LCORE-1822: Enable OpenTelemetry delivery E2E test with mock OTLP collector

docker-compose-library.yaml

@@ -113,6 +113,25 @@ services:
       retries: 3
       start_period: 2s
 
+  # Mock OTLP/HTTP collector for OpenTelemetry E2E tests.
+  # Starts with the stack; intentionally not a dependency of lightspeed-stack so
+  # non-OTEL features do not require this container.
+  mock-otel:
+    build:
+      context: ./tests/e2e/mock_otel_collector
+      dockerfile: Dockerfile
+    container_name: mock-otel
+    ports:
+      - "4318:4318"
+    networks:
+      - lightspeednet
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4318/health')"]
+      interval: 5s
+      timeout: 3s
+      retries: 3
+      start_period: 2s
+
 
 networks:
   lightspeednet:

docker-compose.yaml

@@ -167,6 +167,25 @@ services:
       retries: 3
       start_period: 2s
 
+  # Mock OTLP/HTTP collector for OpenTelemetry E2E tests.
+  # Starts with the stack; intentionally not a dependency of lightspeed-stack so
+  # non-OTEL features do not require this container.
+  mock-otel:
+    build:
+      context: ./tests/e2e/mock_otel_collector
+      dockerfile: Dockerfile
+    container_name: mock-otel
+    ports:
+      - "4318:4318"
+    networks:
+      - lightspeednet
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4318/health')"]
+      interval: 5s
+      timeout: 3s
+      retries: 3
+      start_period: 2s
+
   # Mock TLS inference server for TLS E2E tests
   mock-tls-inference:
     build:

src/app/endpoints/responses.py

@@ -578,14 +578,18 @@ async def handle_responses_with_tracing(  # pylint: disable=too-many-locals
     )
     attachments_count = _count_request_attachments(original_request.input)
 
-    set_span_attributes(
-        root_span,
-        {
-            SpanAttributes.USER_ID: anonymize_value(user_id),
-            SpanAttributes.INPUT: anonymize_value(input_text),
-            SpanAttributes.REQUEST_ATTACHMENTS_COUNT: attachments_count,
-        },
-    )
+    span_attributes: dict[str, Any] = {
+        SpanAttributes.USER_ID: anonymize_value(user_id),
+        SpanAttributes.INPUT: anonymize_value(input_text),
+        SpanAttributes.REQUEST_ATTACHMENTS_COUNT: attachments_count,
+    }
+    # safety_identifier is a caller-supplied, non-PII identifier, so it is
+    # recorded verbatim (not anonymized) when present.
+    if original_request.safety_identifier is not None:
+        span_attributes[SpanAttributes.SAFETY_IDENTIFIER] = (
+            original_request.safety_identifier
+        )
+    set_span_attributes(root_span, span_attributes)
 
     await check_mcp_auth(configuration, mcp_headers, token, request.headers)
 

src/utils/otel_tracing.py

@@ -24,6 +24,7 @@ class SpanAttributes(StrEnum):
 
     SESSION_ID = "session.id"
     USER_ID = "user.id"  # anonymized
+    SAFETY_IDENTIFIER = "request.safety_identifier"  # caller-supplied identifier
     INPUT = "request.input"  # anonymized
     OUTPUT = "response.output"  # anonymized
     RESPONSE_ERROR = "response.error"

tests/e2e/features/environment.py

@@ -39,10 +39,23 @@
     run_e2e_ops,
 )
 from tests.e2e.utils.utils import (
+    force_recreate_compose_service,
     is_prow_environment,
     remove_config_backup,
     restart_container,
     switch_config,
+    wait_for_container_health,
+    wait_for_lightspeed_stack_http_ready,
+)
+
+# OTEL exporter env vars set by the OpenTelemetry feature's steps. They are
+# reverted after the feature so later scenarios do not inherit OTEL export.
+_OTEL_FEATURE_TAG = "OTel"
+_OTEL_ENV_VARS = (
+    "OTEL_SDK_DISABLED",
+    "OTEL_EXPORTER_OTLP_ENDPOINT",
+    "OTEL_EXPORTER_OTLP_PROTOCOL",
+    "OTEL_SERVICE_NAME",
 )
 
 FALLBACK_MODEL = "gpt-4o-mini"
@@ -498,14 +511,42 @@ def _restore_config_after_feature_enabled() -> bool:
     }
 
 
+def _teardown_otel_export(context: Context) -> None:
+    """Revert the OTEL exporter configuration after the OpenTelemetry feature.
+
+    The OpenTelemetry feature sets ``OTEL_*`` env vars and force-recreates
+    ``lightspeed-stack`` so it exports telemetry. Those vars were originally
+    unset, so popping them restores the Compose defaults. The container is then
+    force-recreated (a plain ``docker restart`` would keep the OTEL-enabled
+    creation-time env), so later scenarios do not keep exporting to the mock
+    collector. No-op on Prow, where the feature is skipped and Compose is
+    unavailable.
+    """
+    if is_prow_environment():
+        return
+    for var in _OTEL_ENV_VARS:
+        os.environ.pop(var, None)
+    force_recreate_compose_service(
+        "lightspeed-stack", is_library_mode=context.is_library_mode
+    )
+    wait_for_container_health("lightspeed-stack")
+    wait_for_lightspeed_stack_http_ready()
+
+
 def after_feature(context: Context, feature: Feature) -> None:
     """Run after each feature file is exercised.
 
-    Perform feature-level teardown: restore bootstrap configuration when
+    Perform feature-level teardown: revert OTEL export configuration for the
+    OpenTelemetry feature; restore bootstrap configuration when
     ``E2E_RESTORE_CONFIG_AFTER_FEATURE=1``, otherwise keep the active config;
     when ``context.feedback_e2e_conversation_cleanup`` is set by feedback steps,
     delete tracked feedback test conversations.
     """
+    # Revert OTEL export before other teardown so a later docker restart keeps
+    # the OTEL-disabled, freshly-recreated container.
+    if _OTEL_FEATURE_TAG in feature.tags:
+        _teardown_otel_export(context)
+
     # Restore OGX FIRST (before any lightspeed-stack restart).
     # Read from module-level state — Behave clears custom context attributes
     # between scenarios, so context.ogx_was_running is unreliable here.

tests/e2e/features/opentelemetry.feature

@@ -1,4 +1,4 @@
-@cfg_authorized @OTel @skip
+@cfg_authorized @OTel @skip-in-prow
 Feature: OpenTelemetry observability tests
 
   Background:
@@ -25,5 +25,4 @@ Feature: OpenTelemetry observability tests
       }
       """
     Then The status code of the response is 200
-     And The service exported an OpenTelemetry event containing e2e-otel-delivery-marker
      And The OpenTelemetry service received data containing e2e-otel-delivery-marker

tests/e2e/features/steps/opentelemetry.py

@@ -0,0 +1,156 @@
+"""Step definitions for the OpenTelemetry telemetry-delivery E2E scenario.
+
+These steps reset the mock OTLP/HTTP collector (the ``mock-otel`` Docker Compose
+service, started with the rest of the stack), reconfigure the Lightspeed Core
+Stack to export spans/events to it, and assert that telemetry containing a
+scenario marker is delivered.
+
+The Lightspeed Core Stack has no in-app tracing configuration: the OTEL SDK is
+enabled only when the container entrypoint launches the app under
+``opentelemetry-instrument``, which happens when ``OTEL_SDK_DISABLED=false``.
+The exporter target is then read from the standard ``OTEL_*`` environment
+variables. Because those are set at container-creation time, the
+"configure to export" step force-recreates the container rather than restarting
+it. The OTEL environment and container are reset in ``after_feature`` so later
+scenarios do not inherit the OTEL-enabled configuration.
+"""
+
+import os
+import time
+
+import requests
+from behave import given, then  # pyright: ignore[reportAttributeAccessIssue]
+from behave.runner import Context
+
+from tests.e2e.utils.utils import (
+    force_recreate_compose_service,
+    wait_for_container_health,
+    wait_for_lightspeed_stack_http_ready,
+)
+
+# Compose service / container name for the mock collector (see docker-compose*.yaml).
+MOCK_OTEL_SERVICE = "mock-otel"
+LIGHTSPEED_STACK_SERVICE = "lightspeed-stack"
+
+# Endpoint the Lightspeed Core Stack exports to, on the shared compose network.
+# Overridable so the scenario can target an external collector if needed.
+OTEL_EXPORT_ENDPOINT = os.getenv(
+    "E2E_OTEL_EXPORT_ENDPOINT", f"http://{MOCK_OTEL_SERVICE}:4318"
+)
+OTEL_EXPORT_PROTOCOL = "http/protobuf"
+OTEL_EXPORT_SERVICE_NAME = os.getenv("E2E_OTEL_SERVICE_NAME", "lightspeed-stack-e2e")
+
+# Host-side control API of the mock collector (published port from docker-compose).
+_MOCK_OTEL_HOST = os.getenv("E2E_OTEL_MOCK_HOST", "localhost")
+_MOCK_OTEL_PORT = os.getenv("E2E_OTEL_MOCK_PORT", "4318")
+MOCK_OTEL_CONTROL_BASE = f"http://{_MOCK_OTEL_HOST}:{_MOCK_OTEL_PORT}"
+
+# Delivery is asynchronous: the SDK batches spans before export. Poll generously.
+_DELIVERY_TIMEOUT_S = float(os.getenv("E2E_OTEL_DELIVERY_TIMEOUT_S", "45"))
+_DELIVERY_POLL_INTERVAL_S = 2.0
+_HEALTH_TIMEOUT_S = 30.0
+
+
+def _wait_for_mock_health() -> None:
+    """Poll the mock collector control API until it reports healthy."""
+    url = f"{MOCK_OTEL_CONTROL_BASE}/health"
+    deadline = time.monotonic() + _HEALTH_TIMEOUT_S
+    last_error = "no response"
+    while time.monotonic() < deadline:
+        try:
+            response = requests.get(url, timeout=5)
+            if response.status_code == 200:
+                return
+            last_error = f"HTTP {response.status_code}"
+        except requests.RequestException as exc:
+            last_error = f"{exc.__class__.__name__}: {exc}"
+        time.sleep(1.0)
+    raise AssertionError(
+        f"Mock OTEL collector did not become healthy at {url!r} "
+        f"within {_HEALTH_TIMEOUT_S:.0f}s (last: {last_error})"
+    )
+
+
+def _reset_mock_collector() -> None:
+    """Clear any telemetry buffered by the mock collector from prior runs."""
+    response = requests.post(f"{MOCK_OTEL_CONTROL_BASE}/reset", timeout=5)
+    assert (
+        response.status_code == 200
+    ), f"Failed to reset mock OTEL collector: HTTP {response.status_code}"
+
+
+def _poll_collector_contains(marker: str) -> bool:
+    """Return True once the collector has buffered a payload containing ``marker``."""
+    url = f"{MOCK_OTEL_CONTROL_BASE}/received"
+    deadline = time.monotonic() + _DELIVERY_TIMEOUT_S
+    while True:
+        try:
+            response = requests.get(url, params={"contains": marker}, timeout=5)
+            if response.status_code == 200 and response.json().get("found"):
+                return True
+        except requests.RequestException:
+            pass
+        if time.monotonic() >= deadline:
+            return False
+        time.sleep(_DELIVERY_POLL_INTERVAL_S)
+
+
+@given("An OpenTelemetry service is running and listening for OTLP data")
+def otel_service_running(context: Context) -> None:
+    """Ensure the mock OTLP collector is healthy and its buffer is empty.
+
+    The ``mock-otel`` Compose service starts with the rest of the stack, so this
+    step only waits for it to report healthy (Docker health and its own control
+    API) and resets any previously buffered telemetry so the scenario starts
+    from a clean slate.
+    """
+    wait_for_container_health(MOCK_OTEL_SERVICE)
+    _wait_for_mock_health()
+    _reset_mock_collector()
+    context.otel_collector_endpoint = OTEL_EXPORT_ENDPOINT
+
+
+@given("The service is configured to export data to the OpenTelemetry service")
+def configure_service_export(context: Context) -> None:
+    """Enable OTEL export and recreate the service so the exporter is active.
+
+    The OTEL SDK is only initialized when the entrypoint launches the app under
+    ``opentelemetry-instrument`` (``OTEL_SDK_DISABLED=false``), and the exporter
+    target comes from environment variables fixed at container creation. This
+    sets those variables and force-recreates the ``lightspeed-stack`` container
+    so it exports HTTP/protobuf to the mock collector. ``after_feature`` reverts
+    these variables and recreates the container for later scenarios.
+    """
+    os.environ["OTEL_SDK_DISABLED"] = "false"
+    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = OTEL_EXPORT_ENDPOINT
+    os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = OTEL_EXPORT_PROTOCOL
+    os.environ["OTEL_SERVICE_NAME"] = OTEL_EXPORT_SERVICE_NAME
+    # before_all sets OTEL_ANONYMIZATION_SECRET; keep any existing value.
+    os.environ.setdefault(
+        "OTEL_ANONYMIZATION_SECRET", "e2e-test-secret-do-not-use-in-production"
+    )
+
+    force_recreate_compose_service(
+        LIGHTSPEED_STACK_SERVICE,
+        is_library_mode=getattr(context, "is_library_mode", False),
+    )
+    wait_for_container_health(LIGHTSPEED_STACK_SERVICE)
+    wait_for_lightspeed_stack_http_ready()
+    context.otel_export_configured = True
+
+
+@then("The OpenTelemetry service received data containing {marker}")
+def collector_received_data(context: Context, marker: str) -> None:
+    """Assert the mock collector buffered telemetry containing ``marker``.
+
+    Verifies delivery from the collector's perspective; polls to tolerate the
+    SDK's batched, asynchronous export.
+    """
+    assert getattr(
+        context, "otel_collector_endpoint", None
+    ), "The OpenTelemetry service must be started before asserting on delivery"
+    marker = marker.strip()
+    assert _poll_collector_contains(marker), (
+        f"Mock OTEL collector did not receive data containing {marker!r} "
+        f"within {_DELIVERY_TIMEOUT_S:.0f}s"
+    )

tests/e2e/mock_otel_collector/Dockerfile

@@ -0,0 +1,5 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY server.py .
+EXPOSE 4318
+CMD ["python", "server.py"]

tests/e2e/mock_otel_collector/README.md

@@ -0,0 +1,53 @@
+# Mock OTEL collector
+
+A minimal OTLP/HTTP collector used by the OpenTelemetry E2E scenario
+(`tests/e2e/features/opentelemetry.feature`) to verify that the Lightspeed Core
+Stack delivers spans/events to a telemetry backend.
+
+It is a stdlib-only `http.server` that buffers the raw OTLP export bodies in
+memory and exposes a small control API so Behave steps can assert what was
+received. See `server.py` for the full endpoint list.
+
+## Endpoints
+
+| Method & path        | Purpose                                                        |
+| -------------------- | ------------------------------------------------------------- |
+| `POST /v1/*`         | Receive an OTLP export (traces/logs/metrics); body buffered.  |
+| `GET /received`      | Report the count of buffered exports.                         |
+| `GET /received?contains=<text>` | Report whether `<text>` appears in any payload.   |
+| `POST /reset`        | Clear the buffer (called at scenario start).                  |
+| `GET /health`        | Liveness probe (`{"status": "ok"}`).                         |
+
+Substring queries search the raw request bytes. OTLP protobuf encodes string
+fields as UTF-8, so a plaintext marker embedded in a span attribute value is
+found without decoding protobuf.
+
+## Running
+
+Locally:
+
+```bash
+python server.py [port]   # default port 4318
+```
+
+In E2E it runs as the `mock-otel` Docker Compose service on the `lightspeednet`
+network. It starts with the rest of the stack (`docker compose up -d`); the
+`An OpenTelemetry service is running and listening for OTLP data` step only waits
+for it to become healthy and resets its buffer. It is intentionally **not** wired
+into `lightspeed-stack`'s `depends_on` (non-OTEL features must not require this
+container).
+
+## Pointing the service at it
+
+The Lightspeed Core Stack exports via HTTP/protobuf when launched with the OTEL
+SDK enabled:
+
+```bash
+OTEL_SDK_DISABLED=false
+OTEL_EXPORTER_OTLP_ENDPOINT=http://mock-otel:4318
+OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
+```
+
+The `The service is configured to export data to the OpenTelemetry service` step
+sets these and recreates the container so the entrypoint launches it under
+`opentelemetry-instrument`.