Enhance WeaverLiveCheck runner capabilities (#5526)

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:

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)
+

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:

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()