feat(otel): xtest tracing plugin + otdf-local Jaeger wiring (DSPX-3635)

otdf-local/src/otdf_local/cli.py

@@ -94,14 +94,27 @@ def up(
         bool,
         typer.Option("--no-provision", help="Skip provisioning step"),
     ] = False,
+    tracing: Annotated[
+        bool,
+        typer.Option(
+            "--tracing",
+            help="Start Jaeger and export OpenTelemetry traces from platform/KAS",
+        ),
+    ] = False,
 ) -> None:
     """Start the test environment.
 
     By default starts all services: docker (keycloak, postgres), platform, and all KAS instances.
     """
     settings = get_settings()
+    # Enable tracing on the shared settings so every service (docker profile,
+    # platform/KAS config generation) picks it up.
+    settings.tracing = tracing
     settings.ensure_directories()
 
+    if tracing:
+        print_info(f"Tracing enabled — Jaeger UI at {settings.jaeger_ui_url}")
+
     # Parse services to start
     if services:
         service_list = [s.strip().lower() for s in services.split(",")]
@@ -591,6 +604,16 @@ def env(
         root_key = get_nested(platform_config, "services.kas.root_key")
         if root_key:
             env_vars["OT_ROOT_KEY"] = root_key
+
+        # If the running platform is exporting traces, point clients (pytest,
+        # SDK CLIs) at the same collector so their spans join the trace.
+        if get_nested(platform_config, "server.trace.enabled"):
+            endpoint = get_nested(
+                platform_config,
+                "server.trace.provider.otlp.endpoint",
+                settings.otlp_endpoint,
+            )
+            env_vars["OTEL_EXPORTER_OTLP_ENDPOINT"] = endpoint
     except Exception as e:
         print_warning(f"Could not read root key from platform config: {e}")
 

otdf-local/src/otdf_local/config/settings.py

@@ -2,6 +2,7 @@
 
 from functools import lru_cache
 from pathlib import Path
+from typing import Any
 
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -146,6 +147,25 @@ def docker_compose_file(self) -> Path:
     # Log level
     log_level: str = "info"
 
+    # OpenTelemetry tracing (opt-in via `otdf-local up --tracing`)
+    tracing: bool = False
+    otlp_endpoint: str = "localhost:4317"
+    jaeger_ui_url: str = "http://localhost:16686"
+
+    def trace_config_updates(self) -> dict[str, Any]:
+        """Platform/KAS config overrides that export OTel traces to the local
+        collector (Jaeger). Empty when tracing is disabled so the generated
+        config is unchanged for normal runs.
+        """
+        if not self.tracing:
+            return {}
+        return {
+            "server.trace.enabled": True,
+            "server.trace.provider.name": "otlp",
+            "server.trace.provider.otlp.endpoint": self.otlp_endpoint,
+            "server.trace.provider.otlp.insecure": True,
+        }
+
     def get_kas_port(self, name: str) -> int:
         """Get port for a KAS instance."""
         return Ports.get_kas_port(name)

otdf-local/src/otdf_local/services/docker.py

@@ -32,13 +32,22 @@ def service_type(self) -> ServiceType:
     def health_url(self) -> str:
         return f"http://localhost:{Ports.KEYCLOAK}/auth/realms/master"
 
+    def _compose_cmd(self, *args: str) -> list[str]:
+        """Build a `docker compose` command, opting into the `tracing` profile
+        (which includes the Jaeger all-in-one container) when tracing is enabled.
+        """
+        cmd = ["docker", "compose", "-f", str(self._compose_file)]
+        if self.settings.tracing:
+            cmd += ["--profile", "tracing"]
+        return cmd + list(args)
+
     def start(self) -> bool:
         """Start Docker compose services."""
         if not self._compose_file.exists():
             return False
 
         result = subprocess.run(
-            ["docker", "compose", "-f", str(self._compose_file), "up", "-d"],
+            self._compose_cmd("up", "-d"),
             capture_output=True,
             text=True,
             cwd=self._compose_file.parent,
@@ -51,7 +60,7 @@ def stop(self) -> bool:
             return False
 
         result = subprocess.run(
-            ["docker", "compose", "-f", str(self._compose_file), "down"],
+            self._compose_cmd("down"),
             capture_output=True,
             text=True,
             cwd=self._compose_file.parent,

otdf-local/src/otdf_local/services/kas.py

@@ -81,6 +81,9 @@ def _generate_config(self) -> Path:
             # registered_kas_uri should NOT have /kas suffix
             updates["services.kas.registered_kas_uri"] = f"http://localhost:{self.port}"
 
+        # Export traces to the local collector (Jaeger) when tracing is enabled
+        updates.update(self.settings.trace_config_updates())
+
         copy_yaml_with_updates(template_path, config_path, updates)
         return config_path
 

otdf-local/src/otdf_local/services/platform.py

@@ -67,6 +67,9 @@ def _generate_config(self) -> Path:
             "logger.output": logger_output,
         }
 
+        # Export traces to the local collector (Jaeger) when tracing is enabled
+        updates.update(self.settings.trace_config_updates())
+
         copy_yaml_with_updates(template_path, config_path, updates)
 
         # Set up golden keys for legacy TDF tests

xtest/conftest.py

@@ -52,6 +52,7 @@ def pytest_report_header() -> list[str]:
     "fixtures.keys",
     "fixtures.audit",
     "fixtures.encryption",
+    "fixtures.tracing",
 ]
 
 
@@ -129,6 +130,12 @@ def pytest_addoption(parser: pytest.Parser):
         action="store_true",
         help="skip round-trip tests where all SDKs are released artifacts",
     )
+    parser.addoption(
+        "--tracing",
+        action="store_true",
+        help="wrap each test in an OpenTelemetry span and export it to the OTLP "
+        "collector (also enabled by setting OTEL_EXPORTER_OTLP_ENDPOINT)",
+    )
     parser.addoption(
         "--sdks",
         help=f"select which sdks to run by default, unless overridden; one or more of {englist(typing.get_args(tdfs.sdk_type))}, optionally version-qualified (e.g. go@main, go@v0.18.0, go@*)",

xtest/fixtures/tracing.py

@@ -0,0 +1,142 @@
+"""Pytest fixtures for end-to-end OpenTelemetry tracing.
+
+Wraps each test in a ``pytest.test`` span and exports a ``TRACEPARENT`` into the
+environment so the SDK CLI subprocess (and, through it, platform/KAS) join the
+same trace. On failure the trace's Jaeger URL is printed so the failure links
+directly to the full request chain.
+
+Tracing is opt-in and a strict no-op unless enabled — it activates when either
+``--tracing`` is passed or ``OTEL_EXPORTER_OTLP_ENDPOINT`` is set. When disabled
+nothing is imported or initialized, so normal runs pay no cost.
+
+See ``fixtures/audit.py`` for the sibling log-collection fixture this mirrors.
+"""
+
+import logging
+import os
+from collections.abc import Iterator
+from dataclasses import dataclass
+
+import pytest
+
+logger = logging.getLogger("xtest")
+
+# Default local collector (Jaeger all-in-one, OTLP gRPC) and UI, matching the
+# `tracing` docker-compose profile started by `otdf-local up --tracing`.
+_DEFAULT_OTLP_ENDPOINT = "localhost:4317"
+_DEFAULT_JAEGER_UI = "http://localhost:16686"
+
+
+@dataclass
+class TracingSession:
+    """Holds the initialized tracer and the Jaeger UI base URL for a session."""
+
+    tracer: object  # opentelemetry.trace.Tracer
+    provider: object  # opentelemetry.sdk.trace.TracerProvider
+    jaeger_ui_url: str
+
+
+@pytest.fixture(scope="session")
+def _tracing(request: pytest.FixtureRequest) -> Iterator[TracingSession | None]:
+    """Initialize an OTLP tracer for the session, or yield None when disabled.
+
+    Enabled when ``--tracing`` is passed or ``OTEL_EXPORTER_OTLP_ENDPOINT`` is
+    set; otherwise this is a no-op and no OpenTelemetry code runs.
+    """
+    endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
+    enabled = bool(request.config.getoption("--tracing", default=False)) or bool(
+        endpoint
+    )
+    if not enabled:
+        yield None
+        return
+
+    endpoint = endpoint or _DEFAULT_OTLP_ENDPOINT
+
+    # Imported lazily so the disabled path never requires the OTel packages.
+    from opentelemetry import trace
+    from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+    from opentelemetry.sdk.resources import Resource
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+    provider = TracerProvider(resource=Resource.create({"service.name": "xtest"}))
+    provider.add_span_processor(
+        BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, insecure=True))
+    )
+    trace.set_tracer_provider(provider)
+    logger.info("xtest tracing enabled, exporting to %s", endpoint)
+
+    session = TracingSession(
+        tracer=provider.get_tracer("xtest"),
+        provider=provider,
+        jaeger_ui_url=os.getenv("JAEGER_UI_URL", _DEFAULT_JAEGER_UI),
+    )
+    try:
+        yield session
+    finally:
+        # Flush any buffered spans before the process exits.
+        provider.shutdown()
+
+
+def _test_span_attributes(request: pytest.FixtureRequest) -> dict[str, str]:
+    """Derive span attributes (sdk, container) from the test's parametrization."""
+    attrs: dict[str, str] = {"test.name": request.node.name}
+    callspec = getattr(request.node, "callspec", None)
+    if callspec is None:
+        return attrs
+    params = callspec.params
+    for key in ("encrypt_sdk", "decrypt_sdk", "sdk"):
+        if key in params:
+            attrs["test.sdk"] = str(params[key])
+            break
+    if "container" in params:
+        attrs["test.container"] = str(params["container"])
+    return attrs
+
+
+@pytest.fixture(autouse=True)
+def _trace_test(
+    request: pytest.FixtureRequest, _tracing: TracingSession | None
+) -> Iterator[None]:
+    """Wrap each test in a ``pytest.test`` span and propagate it to subprocesses.
+
+    Sets ``TRACEPARENT`` in the environment for the duration of the test so the
+    SDK CLI (invoked via ``subprocess.run`` in ``tdfs.py``, which copies
+    ``os.environ``) starts a child span under this one. Prints the Jaeger trace
+    URL on failure.
+    """
+    if _tracing is None:
+        yield
+        return
+
+    from opentelemetry import trace
+    from opentelemetry.propagate import inject
+
+    tracer = _tracing.tracer
+    with tracer.start_as_current_span("pytest.test") as span:  # type: ignore[attr-defined]
+        for key, value in _test_span_attributes(request).items():
+            span.set_attribute(key, value)
+
+        # Export the active context so child processes continue this trace.
+        carrier: dict[str, str] = {}
+        inject(carrier)
+        prev_traceparent = os.environ.get("TRACEPARENT")
+        if "traceparent" in carrier:
+            os.environ["TRACEPARENT"] = carrier["traceparent"]
+
+        trace_id = format(span.get_span_context().trace_id, "032x")
+        trace_url = f"{_tracing.jaeger_ui_url}/trace/{trace_id}"
+        try:
+            yield
+        finally:
+            # Restore prior TRACEPARENT to avoid leaking across tests.
+            if prev_traceparent is None:
+                os.environ.pop("TRACEPARENT", None)
+            else:
+                os.environ["TRACEPARENT"] = prev_traceparent
+
+            rep = getattr(request.node, "rep_call", None)
+            if rep is not None and rep.failed:
+                span.set_status(trace.Status(trace.StatusCode.ERROR))
+                print(f"\nTrace: {trace_url}")

xtest/pyproject.toml

@@ -30,6 +30,9 @@ dependencies = [
     "jsonschema>=4.25.1",
     "jsonschema-specifications>=2025.9.1",
     "MarkupSafe>=3.0.3",
+    "opentelemetry-api>=1.29.0",
+    "opentelemetry-sdk>=1.29.0",
+    "opentelemetry-exporter-otlp-proto-grpc>=1.29.0",
     "packaging>=26.2",
     "pluggy>=1.6.0",
     "pycparser>=3.0",