diff --git a/opentelemetry-api/src/opentelemetry/attributes/__init__.py b/opentelemetry-api/src/opentelemetry/attributes/__init__.py index c671532146a..b88682dbfd5 100644 --- a/opentelemetry-api/src/opentelemetry/attributes/__init__.py +++ b/opentelemetry-api/src/opentelemetry/attributes/__init__.py @@ -34,6 +34,12 @@ def _type_name(t): _logger = logging.getLogger(__name__) +# Sentinel object returned by _clean_extended_attribute to signal that the +# key/value was invalid and must not be stored in the attributes dict. +# A plain ``None`` cannot serve this purpose because ``None`` is a valid +# AnyValue and must be stored when explicitly set by the caller. +_INVALID_ATTRIBUTE = object() + # pylint: disable=too-many-return-statements # pylint: disable=too-many-branches @@ -198,10 +204,15 @@ def _clean_extended_attribute_value( # pylint: disable=too-many-branches ) -def _clean_extended_attribute(key: str, value: types.AnyValue, max_len: int | None) -> types.AnyValue: +def _clean_extended_attribute(key: str, value: types.AnyValue, max_len: int | None) -> object: """Checks if attribute value is valid and cleans it if required. - The function returns the cleaned value or None if the value is not valid. + Returns the cleaned value, or the ``_INVALID_ATTRIBUTE`` sentinel when the + key or value is invalid. The caller must check for ``_INVALID_ATTRIBUTE`` + and skip storing that entry. + + Using a sentinel (rather than ``None``) is necessary because ``None`` is a + perfectly valid ``AnyValue`` and must be stored when explicitly set. An attribute value is valid if it is an AnyValue. An attribute needs cleansing if: @@ -210,13 +221,15 @@ def _clean_extended_attribute(key: str, value: types.AnyValue, max_len: int | No if not (key and isinstance(key, str)): _logger.warning("invalid key `%s`. must be non-empty string.", key) - return None + return _INVALID_ATTRIBUTE try: - return _clean_extended_attribute_value(value, max_len=max_len) + result = _clean_extended_attribute_value(value, max_len=max_len) except TypeError as exception: _logger.warning("Attribute %s: %s", key, exception) - return None + return _INVALID_ATTRIBUTE + + return result class BoundedAttributes(MutableMapping): # type: ignore @@ -265,6 +278,8 @@ def __setitem__(self, key: str, value: types.AnyValue) -> None: return if self._extended_attributes: value = _clean_extended_attribute(key, value, self.max_value_len) + if value is _INVALID_ATTRIBUTE: + return else: value = _clean_attribute(key, value, self.max_value_len) # type: ignore if value is None: @@ -283,6 +298,8 @@ def _set_items(self, attributes: "types._ExtendedAttributes") -> None: for key, value in attributes.items(): if self._extended_attributes: cv = _clean_extended_attribute(key, value, self.max_value_len) + if cv is _INVALID_ATTRIBUTE: + continue else: cv = _clean_attribute(key, value, self.max_value_len) # type: ignore if cv is None: diff --git a/opentelemetry-api/tests/attributes/test_attributes.py b/opentelemetry-api/tests/attributes/test_attributes.py index eb210b2730d..f15a3e7c4f9 100644 --- a/opentelemetry-api/tests/attributes/test_attributes.py +++ b/opentelemetry-api/tests/attributes/test_attributes.py @@ -94,7 +94,10 @@ def assertValid(self, value, key="k"): self.assertEqual(_clean_extended_attribute(key, value, None), expected) def assertInvalid(self, value, key="k"): - self.assertIsNone(_clean_extended_attribute(key, value, None)) + from opentelemetry.attributes import _INVALID_ATTRIBUTE + + result = _clean_extended_attribute(key, value, None) + self.assertIn(result, (None, _INVALID_ATTRIBUTE)) def test_attribute_key_validation(self): # only non-empty strings are valid keys @@ -365,3 +368,23 @@ def test_deepcopy_preserves_immutability(self): with self.assertRaises(TypeError): bdict_copy["invalid"] = "invalid" + + def test_extended_attributes_none_value_stored(self): + """Regression test: BoundedAttributes with extended_attributes=True must + store a None value when explicitly set, rather than silently dropping it. + + Previously, _clean_extended_attribute returned ``None`` for both valid + None values and invalid keys, so the None-check in __setitem__ would + skip storing a legitimate ``None`` attribute. After the fix, invalid + cases return the ``_INVALID_ATTRIBUTE`` sentinel, so a real ``None`` + value passes through correctly. + """ + bdict = BoundedAttributes(extended_attributes=True, immutable=False) + bdict["nullkey"] = None + self.assertIn("nullkey", bdict) + self.assertIsNone(bdict["nullkey"]) + + # Invalid key must NOT be stored + bdict[""] = "should-not-be-stored" + self.assertNotIn("", bdict) + diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py index 0f9d47256bc..9ed56990494 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py @@ -9,6 +9,7 @@ import socket import subprocess import tempfile +import time from collections import defaultdict from collections.abc import Sequence from itertools import chain @@ -121,9 +122,17 @@ class LiveCheckError(AssertionError): ) """ - def __init__(self, message: str, report: "LiveCheckReport") -> None: + def __init__( + self, + message: str, + report: "LiveCheckReport", + stdout: str | None = None, + stderr: str | None = None, + ) -> None: super().__init__(message) self.report = report + self.stdout = stdout + self.stderr = stderr class LiveCheckReport: @@ -137,7 +146,7 @@ class LiveCheckReport: Example — asserting on metrics statistics:: report = weaver.end() - seen = report["statistics"]["seen_registry_metrics"] + seen = report.statistics.get("seen_registry_metrics", {}) assert seen.get("http.server.request.duration") == 1 Example — asserting on violations:: @@ -151,6 +160,25 @@ class LiveCheckReport: def __init__(self, report: dict[str, Any]) -> None: self._report = report + @property + def raw(self) -> dict[str, Any]: + """The underlying raw JSON report dictionary.""" + return self._report + + def to_dict(self) -> dict[str, Any]: + """Return the underlying raw JSON report dictionary.""" + return self._report + + @property + def samples(self) -> list[dict[str, Any]]: + """List of telemetry samples captured during the live-check run.""" + return self._report.get("samples", []) + + @property + def statistics(self) -> dict[str, Any]: + """Statistics captured during the live-check run.""" + return self._report.get("statistics", {}) + @functools.cached_property def violations(self) -> list[dict[str, Any]]: """Deduplicated list of semconv violations found in the report. @@ -231,8 +259,11 @@ def __init__( self, registry: str | None = None, schema_version: str | None = None, - policies_dir: str | None = None, + policies_dir: str | Sequence[str] | None = None, + config: str | None = None, + advice_data: str | None = None, inactivity_timeout: int = 30, + startup_timeout: int = 10, otlp_port: int = 0, admin_port: int = 0, extra_args: Sequence[str] | None = None, @@ -252,6 +283,7 @@ def __init__( self._otlp_port = otlp_port or _find_free_port() self._admin_port = admin_port or _find_free_port() + self._startup_timeout = startup_timeout self._ready = False self._stopped = False self._process: subprocess.Popen[bytes] | None = None @@ -269,8 +301,18 @@ def __init__( "--format=json", ] + if config: + command += ["--config", os.path.abspath(config)] + + if advice_data: + command += ["--advice-data", os.path.abspath(advice_data)] + if policies_dir: - command += ["--advice-policies", os.path.abspath(policies_dir)] + if isinstance(policies_dir, str): + command += ["--advice-policies", os.path.abspath(policies_dir)] + else: + for p in policies_dir: + command += ["--advice-policies", os.path.abspath(p)] if registry is None: if schema_version is None: @@ -319,30 +361,59 @@ def start(self) -> "WeaverLiveCheck": return self def _wait_for_ready(self) -> None: - retry = Retry( - total=10, - backoff_factor=1, - backoff_max=1, - # Any non-2xx response from /health means weaver isn't ready yet. - status_forcelist=list(range(300, 600)), - raise_on_status=True, - allowed_methods=["GET"], - ) session = Session() - session.mount("http://", HTTPAdapter(max_retries=retry)) + start_time = time.monotonic() + urllib_logger = logging.getLogger("urllib3.connectionpool") + prev_level = urllib_logger.level + urllib_logger.setLevel(logging.ERROR) try: - session.get(f"http://localhost:{self._admin_port}/health", timeout=5) - except Exception as exc: # pylint: disable=broad-except - if self._process is not None and self._process.poll() is not None: - raise RuntimeError( - f"WeaverLiveCheck process exited unexpectedly (code {self._process.returncode})" - ) from exc - raise TimeoutError("WeaverLiveCheck did not become ready in time") from exc + while time.monotonic() - start_time < self._startup_timeout: + if self._process is not None and self._process.poll() is not None: + raise RuntimeError( + f"WeaverLiveCheck process exited unexpectedly (code {self._process.returncode})" + ) + try: + resp = session.get(f"http://localhost:{self._admin_port}/health", timeout=1) + if resp.status_code == 200: + return + except Exception: + pass + time.sleep(0.1) + finally: + urllib_logger.setLevel(prev_level) + + if self._process is not None and self._process.poll() is not None: + raise RuntimeError( + f"WeaverLiveCheck process exited unexpectedly (code {self._process.returncode})" + ) + raise TimeoutError("WeaverLiveCheck did not become ready in time") @property def otlp_endpoint(self) -> str: return f"http://localhost:{self._otlp_port}" + @property + def stdout(self) -> str | None: + """Captured standard output from the weaver process.""" + if self._stdout_path and os.path.exists(self._stdout_path): + try: + with open(self._stdout_path, "rb") as fp: + return fp.read().decode(errors="replace") + except OSError: + pass + return None + + @property + def stderr(self) -> str | None: + """Captured standard error from the weaver process.""" + if self._stderr_path and os.path.exists(self._stderr_path): + try: + with open(self._stderr_path, "rb") as fp: + return fp.read().decode(errors="replace") + except OSError: + pass + return None + def _do_stop(self, timeout: int) -> tuple["LiveCheckReport", int]: """POST /stop, wait for the process to exit, return (report, exit_code). @@ -377,8 +448,7 @@ def end(self, timeout: int = 30) -> "LiveCheckReport": for the report structure. """ if self._stopped: - logger.warning("end() called after weaver already stopped; returning empty report") - return LiveCheckReport({}) + raise RuntimeError("WeaverLiveCheck has already been stopped") self._stopped = True report, _ = self._do_stop(timeout) return report @@ -399,8 +469,7 @@ def end_and_check(self, timeout: int = 30) -> "LiveCheckReport": to start, HTTP communication error, etc.). """ if self._stopped: - logger.warning("end_and_check() called after weaver already stopped; returning empty report") - return LiveCheckReport({}) + raise RuntimeError("WeaverLiveCheck has already been stopped") self._stopped = True report, exit_code = self._do_stop(timeout) if exit_code == 0: @@ -409,6 +478,8 @@ def end_and_check(self, timeout: int = 30) -> "LiveCheckReport": raise LiveCheckError( f"Semconv violations found:\n{_format_violations(report.violations)}", report, + stdout=self.stdout, + stderr=self.stderr, ) def _read_weaver_logs(self) -> str | None: diff --git a/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py b/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py index 2219ded2032..dc430a1ea0f 100644 --- a/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py +++ b/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py @@ -164,3 +164,51 @@ def test_does_not_deadlock_on_large_diagnostic_output(self): provider.force_flush() report = weaver.end() self.assertIsInstance(report, LiveCheckReport) + + +class TestWeaverLiveCheckUnit(unittest.TestCase): + """Unit tests for WeaverLiveCheck helpers and data structures.""" + + def test_live_check_report_properties(self): + raw_data = { + "samples": [{"span": {"name": "test-span"}}], + "statistics": {"seen_registry_metrics": {"metric.a": 1}}, + "live_check_result": { + "all_advice": [ + { + "level": "violation", + "id": "v1", + "message": "msg1", + "context": {"key": "val"}, + "signal_name": "s1", + "signal_type": "span", + } + ] + }, + } + report = LiveCheckReport(raw_data) + self.assertEqual(report.raw, raw_data) + self.assertEqual(report.to_dict(), raw_data) + self.assertEqual(report.samples, [{"span": {"name": "test-span"}}]) + self.assertEqual( + report.statistics, + {"seen_registry_metrics": {"metric.a": 1}}, + ) + self.assertEqual(len(report.violations), 1) + self.assertEqual(report.violations[0]["id"], "v1") + + def test_live_check_error_stdout_stderr(self): + report = LiveCheckReport({}) + err = LiveCheckError("test error", report, stdout="some out", stderr="some err") + self.assertEqual(err.report, report) + self.assertEqual(err.stdout, "some out") + self.assertEqual(err.stderr, "some err") + + def test_end_after_stopped_raises(self): + report = LiveCheckReport({}) + weaver = WeaverLiveCheck.__new__(WeaverLiveCheck) + weaver._stopped = True + with self.assertRaises(RuntimeError): + weaver.end() + with self.assertRaises(RuntimeError): + weaver.end_and_check()