From 4141c3e5f2820d6ea131363a9fddc8b14bee5ebe Mon Sep 17 00:00:00 2001 From: Anik Bhattacharjee Date: Thu, 3 Sep 2026 12:51:42 -0400 Subject: [PATCH] LCORE-1822: Enable OpenTelemetry delivery E2E test with mock OTLP collector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the Behave step definitions for the previously-skipped OpenTelemetry E2E scenario and turns the test on, so CI now verifies end-to-end that the Lightspeed Core Stack actually delivers telemetry to an OTLP backend. The scenario asserts that a `responses` request carrying a `safety_identifier` marker results in that marker reaching a collector. Two pieces were needed to make it real: 1. **A mock OTLP/HTTP collector** to receive and assert on exports. 2. **Instrumentation** so the marker (`safety_identifier`) is actually emitted on a span — previously it was only forwarded to the model provider, so the scenario could never pass and was tagged `@skip`. --- docker-compose-library.yaml | 19 +++ docker-compose.yaml | 19 +++ src/app/endpoints/responses.py | 20 ++- src/utils/otel_tracing.py | 1 + tests/e2e/features/environment.py | 43 ++++- tests/e2e/features/opentelemetry.feature | 3 +- tests/e2e/features/steps/opentelemetry.py | 156 ++++++++++++++++++ tests/e2e/mock_otel_collector/Dockerfile | 5 + tests/e2e/mock_otel_collector/README.md | 53 ++++++ tests/e2e/mock_otel_collector/server.py | 143 ++++++++++++++++ tests/e2e/utils/utils.py | 58 +++++++ .../app/endpoints/responses_otel_helpers.py | 2 + .../unit/app/endpoints/test_responses_otel.py | 48 ++++++ 13 files changed, 559 insertions(+), 11 deletions(-) create mode 100644 tests/e2e/features/steps/opentelemetry.py create mode 100644 tests/e2e/mock_otel_collector/Dockerfile create mode 100644 tests/e2e/mock_otel_collector/README.md create mode 100644 tests/e2e/mock_otel_collector/server.py diff --git a/docker-compose-library.yaml b/docker-compose-library.yaml index 0d750b071..6c55e5c42 100755 --- a/docker-compose-library.yaml +++ b/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: diff --git a/docker-compose.yaml b/docker-compose.yaml index 16dcda1bc..5dda03461 100755 --- a/docker-compose.yaml +++ b/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: diff --git a/src/app/endpoints/responses.py b/src/app/endpoints/responses.py index 38110287d..42f49d0a2 100644 --- a/src/app/endpoints/responses.py +++ b/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) diff --git a/src/utils/otel_tracing.py b/src/utils/otel_tracing.py index f24b35311..9ae1d4a0a 100644 --- a/src/utils/otel_tracing.py +++ b/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" diff --git a/tests/e2e/features/environment.py b/tests/e2e/features/environment.py index 4e449f86b..458a5dc17 100644 --- a/tests/e2e/features/environment.py +++ b/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. diff --git a/tests/e2e/features/opentelemetry.feature b/tests/e2e/features/opentelemetry.feature index 8d8f79a59..96cbe73c5 100644 --- a/tests/e2e/features/opentelemetry.feature +++ b/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 diff --git a/tests/e2e/features/steps/opentelemetry.py b/tests/e2e/features/steps/opentelemetry.py new file mode 100644 index 000000000..789595d0b --- /dev/null +++ b/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" + ) diff --git a/tests/e2e/mock_otel_collector/Dockerfile b/tests/e2e/mock_otel_collector/Dockerfile new file mode 100644 index 000000000..6e2195273 --- /dev/null +++ b/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"] diff --git a/tests/e2e/mock_otel_collector/README.md b/tests/e2e/mock_otel_collector/README.md new file mode 100644 index 000000000..04895c127 --- /dev/null +++ b/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=` | Report whether `` 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`. diff --git a/tests/e2e/mock_otel_collector/server.py b/tests/e2e/mock_otel_collector/server.py new file mode 100644 index 000000000..5497e4ae9 --- /dev/null +++ b/tests/e2e/mock_otel_collector/server.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""Minimal mock OpenTelemetry (OTLP/HTTP) collector for E2E tests. + +Accepts OTLP/HTTP exports from the Lightspeed Core Stack and buffers the raw +request bodies in memory so Behave steps can assert that telemetry was +delivered. Uses only the Python standard library. + +Endpoints +--------- +- ``POST /v1/*`` : Receive an OTLP export (traces/logs/metrics). The body is + buffered and the request is answered with an empty ``application/x-protobuf`` + 200, which the OTLP/HTTP exporter accepts as success. +- ``GET /received`` : Report how many exports have been buffered. With + ``?contains=`` it reports whether that substring appears in any buffered + payload (raw-byte search; OTLP protobuf stores string fields as UTF-8, so a + plaintext marker embedded in an attribute value is found). +- ``POST /reset`` : Clear the buffer (used at the start of a scenario). +- ``GET /health`` : Liveness probe returning ``{"status": "ok"}``. + +Run as ``python server.py [port]``; default port is 4318 (OTLP/HTTP). + +The exporter must be configured for HTTP/protobuf, e.g.:: + + OTEL_EXPORTER_OTLP_ENDPOINT=http://mock-otel:4318 + OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf +""" + +import json +import sys +import threading +from collections import deque +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +# Resource bounds so a reachable caller cannot exhaust process memory by sending +# large or repeated exports (the Compose service publishes port 4318). The deque +# caps the entry count (evicting oldest); the per-body cap bounds each entry. +_MAX_BODY_BYTES = 5 * 1024 * 1024 # reject a single OTLP body larger than 5 MiB +_MAX_ENTRIES = 1000 # keep at most this many buffered exports + +# Buffered export bodies shared across handler threads, oldest first. deque and +# its append/clear are thread-safe, but iteration is not, so reads and writes are +# guarded by a lock. +_received: "deque[bytes]" = deque(maxlen=_MAX_ENTRIES) +_lock = threading.Lock() + + +class Handler(BaseHTTPRequestHandler): + """HTTP handler buffering OTLP exports and answering test queries.""" + + def _send_json(self, status: int, data: dict) -> None: + """Send a JSON response with the given status code.""" + body = json.dumps(data).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # pylint: disable=invalid-name + """Serve the health probe and the buffered-data query endpoint.""" + path, _, query = self.path.partition("?") + if path == "/health": + self._send_json(200, {"status": "ok"}) + return + if path == "/received": + self._handle_received_query(query) + return + self._send_json(404, {"error": "not found"}) + + def _handle_received_query(self, query: str) -> None: + """Answer a count or substring query over the buffered exports.""" + contains = None + for pair in query.split("&"): + key, sep, value = pair.partition("=") + if sep and key == "contains": + contains = value + break + + if contains: + needle = contains.encode("utf-8") + with _lock: + matches = sum(1 for body in _received if needle in body) + count = len(_received) + self._send_json( + 200, + { + "contains": contains, + "found": matches > 0, + "matches": matches, + "count": count, + }, + ) + return + + with _lock: + count = len(_received) + self._send_json(200, {"count": count}) + + def do_POST(self) -> None: # pylint: disable=invalid-name + """Buffer OTLP exports and handle the reset control endpoint.""" + path, _, _ = self.path.partition("?") + if path == "/reset": + with _lock: + _received.clear() + self._send_json(200, {"status": "reset"}) + return + + # Any other POST is treated as an OTLP export (e.g. /v1/traces). + raw_length = self.headers.get("Content-Length", "0") + try: + length = int(raw_length) + except ValueError: + self._send_json(400, {"error": "invalid content-length"}) + return + if length < 0 or length > _MAX_BODY_BYTES: + self._send_json(413, {"error": "payload too large"}) + return + + body = self.rfile.read(length) if length else b"" + with _lock: + _received.append(body) # deque(maxlen=...) evicts the oldest entry + + # Acknowledge with an empty protobuf 200, which the OTLP exporter accepts. + self.send_response(200) + self.send_header("Content-Type", "application/x-protobuf") + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, format: str, *args: Any) -> None: + """Suppress default request logging for minimal test output.""" + + +def main() -> None: + """Start the mock OTLP collector on the requested port.""" + port = int(sys.argv[1]) if len(sys.argv) > 1 else 4318 + server = ThreadingHTTPServer(("0.0.0.0", port), Handler) + print(f"Mock OTEL collector on :{port}") + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/tests/e2e/utils/utils.py b/tests/e2e/utils/utils.py index 1025114fe..13340a108 100644 --- a/tests/e2e/utils/utils.py +++ b/tests/e2e/utils/utils.py @@ -515,6 +515,64 @@ def restart_lightspeed_stack_service( os.environ["E2E_SKIP_LLAMA_RESTORE_ON_LCS_RESTART"] = previous +def compose_file_for_mode(is_library_mode: bool) -> str: + """Return the absolute Compose file path for the active deployment mode. + + Parameters: + is_library_mode: True when the run uses the in-process (library) OGX. + + Returns: + Absolute path to ``docker-compose-library.yaml`` in library mode, + otherwise ``docker-compose.yaml``. + """ + name = "docker-compose-library.yaml" if is_library_mode else "docker-compose.yaml" + return absolute_repo_path(name) + + +def force_recreate_compose_service( + service: str, *, is_library_mode: bool, timeout: int = 300 +) -> None: + """Force-recreate a single Compose service so it picks up current env vars. + + Runs ``docker compose -f up -d --force-recreate --no-deps `` + from the repo root. Needed when a container must be rebuilt with environment + variables that are only read at container-creation time (e.g. the OTEL + exporter configuration), which a plain ``docker restart`` would not apply. + + Parameters: + service: Compose service name to recreate (e.g. ``lightspeed-stack``). + is_library_mode: Selects the Compose file (see ``compose_file_for_mode``). + timeout: Seconds to allow the compose command to run. + + Raises: + AssertionError: If the compose command exits with a non-zero status. + """ + cmd = [ + "docker", + "compose", + "-f", + compose_file_for_mode(is_library_mode), + "up", + "-d", + "--force-recreate", + "--no-deps", + service, + ] + result = subprocess.run( + cmd, + cwd=absolute_repo_path("."), + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + if result.stdout: + print(result.stdout, end="") + if result.returncode != 0: + print(result.stderr, end="") + raise AssertionError(f"`{' '.join(cmd)}` failed with code {result.returncode}") + + def wait_for_lightspeed_stack_http_ready( max_attempts: int = 80, delay_s: float = 1.5, diff --git a/tests/unit/app/endpoints/responses_otel_helpers.py b/tests/unit/app/endpoints/responses_otel_helpers.py index c6605ed36..706ea2729 100644 --- a/tests/unit/app/endpoints/responses_otel_helpers.py +++ b/tests/unit/app/endpoints/responses_otel_helpers.py @@ -286,6 +286,7 @@ async def run_responses_setup_smoke( *, stream: bool, input_text: str = "What is Kubernetes?", + safety_identifier: str | None = None, ) -> ReadableSpan: """Run the handler through setup and return the root span.""" patch_responses_otel_tracers(mocker, tracer, minimal_config) @@ -306,6 +307,7 @@ async def run_responses_setup_smoke( store=False, conversation=OTEL_CONV_ID, generate_topic_summary=False, + safety_identifier=safety_identifier, ), auth=MOCK_AUTH, mcp_headers={}, diff --git a/tests/unit/app/endpoints/test_responses_otel.py b/tests/unit/app/endpoints/test_responses_otel.py index b1de65e81..0d00e0e62 100644 --- a/tests/unit/app/endpoints/test_responses_otel.py +++ b/tests/unit/app/endpoints/test_responses_otel.py @@ -226,6 +226,54 @@ async def test_root_setup_attributes_and_validation_event( ) assert_root_setup_attributes(root, input_text=INPUT_TEXT) + @pytest.mark.asyncio + async def test_safety_identifier_recorded_raw_when_present( + self, + mocker: MockerFixture, + dummy_request: Request, + minimal_config: AppConfig, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """safety_identifier is recorded verbatim (not anonymized) on the root span.""" + tracer, exporter = otel + root = await run_responses_setup_smoke( + mocker, + dummy_request, + tracer, + minimal_config, + exporter, + stream=False, + input_text=INPUT_TEXT, + safety_identifier="e2e-otel-delivery-marker", + ) + assert root.attributes is not None + assert ( + root.attributes[SpanAttributes.SAFETY_IDENTIFIER] + == "e2e-otel-delivery-marker" + ) + + @pytest.mark.asyncio + async def test_safety_identifier_absent_when_not_provided( + self, + mocker: MockerFixture, + dummy_request: Request, + minimal_config: AppConfig, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """No safety_identifier attribute is set when the request omits it.""" + tracer, exporter = otel + root = await run_responses_setup_smoke( + mocker, + dummy_request, + tracer, + minimal_config, + exporter, + stream=False, + input_text=INPUT_TEXT, + ) + assert root.attributes is not None + assert SpanAttributes.SAFETY_IDENTIFIER not in root.attributes + @pytest.mark.asyncio async def test_streaming_root_span_closed_on_setup_error( self,