test(e2e): datadog log delivery for successful chat, messages, and responses (LIT-4447)

tests/e2e/coverage_registry/logging.yaml

@@ -5,7 +5,7 @@
 - {id: logging.s3.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/s3_v2.py", rationale: "Primary audit trail; batch flush no-drop"}
 - {id: logging.s3.failure.writes_object, module: logging, tier: P0, event: failure, assertions: [writes_object], exercised_on: [chat_completions, messages], source: "integrations/s3_v2.py", rationale: "Failed calls persisted for compliance"}
 - {id: logging.gcs_bucket.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/gcs_bucket/gcs_bucket.py", rationale: "GCS parallel to S3"}
-- {id: logging.datadog.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, embeddings], source: "integrations/datadog/datadog.py", rationale: "Powers dashboards/alerts; cardinality regressions common"}
+- {id: logging.datadog.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/datadog/datadog.py", rationale: "Powers dashboards/alerts; cardinality regressions common"}
 - {id: logging.datadog.failure.exports_metric, module: logging, tier: P0, event: failure, assertions: [exports_metric], exercised_on: [chat_completions], source: "integrations/datadog/datadog.py", rationale: "Failure metrics for alerting/SLO"}
 - {id: logging.prometheus.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, embeddings], source: "integrations/prometheus.py", rationale: "Standard OSS metrics; per-key cardinality (existing e2e)"}
 - {id: logging.otel.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/otel/logger.py", rationale: "OTEL spans on every call path"}

tests/e2e/docker-compose.yml

@@ -1,5 +1,41 @@
 # local setup to run e2e tests
 configs:
+  dd_sink_script:
+    content: |
+      # Minimal DataDog logs-intake sink for the logging suite: records every
+      # POST (gunzipping the compressed batches the integration sends) and
+      # replays them as JSON on GET /requests so tests can assert delivery.
+      import gzip, json
+      from http.server import BaseHTTPRequestHandler, HTTPServer
+
+      REQUESTS = []
+
+      class Handler(BaseHTTPRequestHandler):
+          def do_POST(self):
+              body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
+              if self.headers.get("Content-Encoding") == "gzip":
+                  body = gzip.decompress(body)
+              REQUESTS.append({"path": self.path, "body": body.decode("utf-8", "replace")})
+              self.send_response(202)
+              self.end_headers()
+              self.wfile.write(b"{}")
+
+          def do_GET(self):
+              self.send_response(200)
+              if self.path == "/health":
+                  self.send_header("Content-Type", "text/plain")
+                  self.end_headers()
+                  self.wfile.write(b"ok")
+                  return
+              self.send_header("Content-Type", "application/json")
+              self.end_headers()
+              self.wfile.write(json.dumps({"requests": REQUESTS}).encode())
+
+          def log_message(self, *args):
+              pass
+
+      HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+
   litellm_config:
     content: |
       general_settings:
@@ -23,7 +59,7 @@ configs:
         # (PHOENIX_COLLECTOR_HTTP_ENDPOINT below points it at the jaeger service),
         # so gen-AI spans export through a preset-owned provider - the code path
         # where trace splits actually happen - with no cloud credentials needed.
-        callbacks: ["arize_phoenix"]
+        callbacks: ["arize_phoenix", "datadog"]
 
       router_settings:
         routing_strategy: simple-shuffle
@@ -76,10 +112,15 @@ services:
         condition: service_healthy
       jaeger:
         condition: service_healthy
+      dd-sink:
+        condition: service_healthy
     env_file: .env
     environment:
       LITELLM_MASTER_KEY: sk-1234
       STORE_MODEL_IN_DB: "True"
+      DD_API_KEY: local-sink-noauth
+      DD_SITE: datadoghq.com
+      DD_BASE_URL: http://dd-sink:8080
       LITELLM_OTEL_V2: "true"
       PHOENIX_COLLECTOR_HTTP_ENDPOINT: http://jaeger:4318/v1/traces
       PHOENIX_API_KEY: local-jaeger-noauth
@@ -138,3 +179,19 @@ services:
       interval: 3s
       timeout: 3s
       retries: 20
+
+# throwaway DataDog logs-intake sink (records POSTs, replays on GET /requests;
+# see E2E_DD_SINK_URL)
+  dd-sink:
+    image: python:3.12-alpine
+    command: ["python", "/sink.py"]
+    configs:
+      - source: dd_sink_script
+        target: /sink.py
+    ports:
+      - "9915:8080"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"]
+      interval: 3s
+      timeout: 3s
+      retries: 20

tests/e2e/e2e_config.py

@@ -32,6 +32,10 @@
 # read exported spans back through it.
 OTEL_QUERY_URL = os.environ.get("E2E_OTEL_QUERY_URL", "http://localhost:16686").rstrip("/")
 
+# Query URL of the compose stack's DataDog logs-intake sink (the `dd-sink`
+# service records every intake POST and replays them on GET /requests).
+DD_SINK_URL = os.environ.get("E2E_DD_SINK_URL", "http://localhost:9915").rstrip("/")
+
 # Writes on the proxy are eventually consistent (e.g. spend rows flush on
 # proxy_batch_write_at, ~60s). Read-backs poll to this deadline, never sleep-once.
 POLL_TIMEOUT = float(os.environ.get("E2E_POLL_TIMEOUT", "120"))

tests/e2e/logging/conftest.py

@@ -11,6 +11,7 @@
 import pytest
 
 from logging_client import LangfuseCreds, LoggingClient, build_logging_client, load_langfuse_creds
+from datadog_sink import DdSinkReader, build_dd_sink_reader
 from otel_client import OtelReader, build_otel_reader
 
 
@@ -35,6 +36,12 @@ def otel_reader() -> OtelReader:
     return build_otel_reader()
 
 
+@pytest.fixture(scope="session")
+def dd_sink() -> DdSinkReader:
+    """Read-back client for the compose stack's DataDog logs-intake sink."""
+    return build_dd_sink_reader()
+
+
 @pytest.fixture
 def datadog_creds() -> None:
     """Require Datadog shipping credentials. Hard-fail when absent; never skip."""

tests/e2e/logging/datadog_sink.py

@@ -0,0 +1,103 @@
+"""Read-back for the DataDog logging tests: typed models over the compose
+stack's dd-sink service, which records every logs-intake POST the datadog
+callback sends (gunzipped) and replays them as JSON.
+
+Delivery is judged on what the sink actually received, mirroring how the OTEL
+tests read Jaeger; a failed sink query is a hard failure, never an empty
+result. External reads go through ``e2e_http``.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+
+import pytest
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
+
+from e2e_config import DD_SINK_URL, POLL_INTERVAL, POLL_TIMEOUT
+from e2e_http import URL, NoBody, Success, get
+
+
+class DdSinkRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    path: str
+    body: str
+
+
+class DdSinkRequests(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    requests: list[DdSinkRequest] = []
+
+
+class DdLogEvent(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    message: str
+    ddsource: str | None = None
+    service: str | None = None
+    status: str | None = None
+
+
+_EVENT_BATCH: TypeAdapter[list[DdLogEvent]] = TypeAdapter(list[DdLogEvent])
+
+
+def _parse_batch(request: DdSinkRequest) -> list[DdLogEvent]:
+    """The intake accepts an array of events or a single event object."""
+    try:
+        return _EVENT_BATCH.validate_json(request.body)
+    except ValidationError:
+        try:
+            return [DdLogEvent.model_validate_json(request.body)]
+        except ValidationError:
+            pytest.fail(f"dd-sink recorded a non-log body on {request.path}: {request.body[:200]}")
+
+
+@dataclass(frozen=True, slots=True)
+class DdSinkReader:
+    sink_url: str
+
+    def _recorded_requests(self) -> list[DdSinkRequest]:
+        result = get(
+            URL(f"{self.sink_url}/requests"),
+            headers=NoBody(),
+            params=NoBody(),
+            response_type=DdSinkRequests,
+            timeout=30.0,
+        )
+        match result:
+            case Success(data=page):
+                return page.requests
+            case failure:
+                pytest.fail(f"dd-sink query at {self.sink_url} failed: {failure}")
+
+    def events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Every log event across every recorded intake batch whose message
+        carries the marker. More than one hit for one call IS the
+        duplicate-delivery bug, so this never collapses to a single event."""
+        events: list[DdLogEvent] = []
+        for request in self._recorded_requests():
+            if "/api/v2/logs" not in request.path:
+                continue
+            events.extend(event for event in _parse_batch(request) if marker in event.message)
+        return events
+
+    def poll_events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Poll until at least one matching event lands (the callback flushes
+        in periodic batches), then re-read after one more interval so a late
+        duplicate cannot hide from the exactly-one assertion. At the deadline
+        the last result is returned as-is."""
+        deadline = time.monotonic() + POLL_TIMEOUT
+        while time.monotonic() < deadline:
+            events = self.events_for_marker(marker)
+            if events:
+                time.sleep(POLL_INTERVAL)
+                return self.events_for_marker(marker)
+            time.sleep(POLL_INTERVAL)
+        return self.events_for_marker(marker)
+
+
+def build_dd_sink_reader() -> DdSinkReader:
+    return DdSinkReader(sink_url=DD_SINK_URL)

tests/e2e/logging/logging_client.py

@@ -18,7 +18,7 @@
 import os
 import time
 from dataclasses import dataclass
-from typing import Literal
+from typing import Callable, Literal
 
 import pytest
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError
@@ -28,6 +28,7 @@
 from e2e_http import (
     URL,
     AuthHeaders,
+    require_successful_call,
     NoBody,
     StreamingResponse,
     Success,
@@ -617,5 +618,20 @@ def poll_langfuse_trace_observations(
         return self.list_langfuse_observations(creds, trace_id=gen.trace_id) or [gen]
 
 
+def first_ok(client: LoggingClient, send: Callable[[], StreamingResponse]) -> StreamingResponse:
+    """First successful call on a fresh key. A fresh key may briefly 401 until
+    the data plane's auth cache picks it up, so retry on 401 to a deadline; a
+    401 is rejected before the LLM call, so it cannot contaminate delivery or
+    trace assertions. Any other failure is behavior under test and fails hard."""
+    deadline = time.monotonic() + client.gateway.poll_timeout
+    while True:
+        outcome = send()
+        if outcome.ok:
+            return outcome
+        if outcome.status_code != 401 or time.monotonic() >= deadline:
+            require_successful_call(outcome)
+        time.sleep(client.gateway.poll_interval)
+
+
 def build_logging_client() -> LoggingClient:
     return LoggingClient(gateway=build_gateway())