test(e2e): add datadog log-delivery e2e scenarios

Drive real traffic through the proxy and read the events back out of the
Datadog Logs Search API to prove delivery instead of trusting the proxy. The
DatadogClient reads DD_API_KEY, DD_SITE and DD_APP_KEY from the environment and
polls the search API for a per-run marker stamped into each request

Three scenarios cover the contract: a batch of gpt-5.5 chat completions shipped
with no drops, the /v1/responses path on a claude-haiku-4-5 deployment
exercising the Responses-to-Anthropic translation, and a failed request that
must land as an error event. The e2e compose config enables the datadog (and
prometheus) callback so the delivery path is actually exercised, and the suite
skips unless all three credentials are set

tests/e2e/docker-compose.yml

@@ -11,6 +11,7 @@ configs:
         drop_params: true
         num_retries: 3
         request_timeout: 600
+        callbacks: ["prometheus", "datadog"]
         cache: true
         cache_params:
           type: redis

tests/e2e/logging/conftest.py

@@ -1,37 +1,39 @@
-"""Fixtures for the Datadog logging suite.
+"""Fixtures for the logging e2e suite.
 
-These tests drive the Datadog batch-send path (#25663) directly against the real
-Datadog logs intake with synthetic events - no LLM calls, no proxy, no log
-read-back - so they need only the shipping credentials DD_API_KEY + DD_SITE
-(DD_SERVICE is an optional tag). No Datadog Application key is required, and they
-skip when the shipping credentials are absent from the environment.
+The Datadog tests exercise the real delivery path: the proxy ships each request's
+StandardLoggingPayload to Datadog on its ``datadog`` callback, and the tests read
+those events back out of the Datadog Logs Search API to prove they landed. That
+read-back needs all three credentials - ``DD_API_KEY`` and ``DD_SITE`` (which the
+proxy also ships with) plus ``DD_APP_KEY`` (the Logs Search API rejects reads that
+carry only an API key) - so the suite skips when any is absent.
+
+The shared ``resources`` / ``scoped_key`` fixtures come from the root e2e conftest
+via the GatewayProvider protocol (LoggingClient exposes ``.gateway``).
 """
 
-import os
-
 import pytest
 
-from logging_client import LoggingClient, build_logging_client
+from logging_client import DatadogClient, LoggingClient, build_logging_client
 
 
 def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line(
         "markers",
-        "covers: registry cell a test covers, e.g. logging.datadog.success.writes_object",
+        "covers: registry cell a test covers, e.g. logging.datadog.success.exports_metric",
     )
 
 
 @pytest.fixture(scope="session")
 def client() -> LoggingClient:
-    """The logging suite's client: holds the shared Gateway so `resources` /
-    `scoped_key` clean up keys, and adds `/metrics` scraping."""
+    """The logging suite's client: holds the shared Gateway (so `resources` /
+    `scoped_key` clean up keys), the Datadog read-back client, and `/metrics`
+    scraping."""
     return build_logging_client()
 
 
-@pytest.fixture
-def datadog_creds() -> None:
-    """Gate the suite on the Datadog shipping credentials. The DataDogLogger is built
-    inside each async test, not here, because its __init__ schedules a periodic-flush
-    task via asyncio.create_task and so needs a running event loop."""
-    if not (os.getenv("DD_API_KEY") and os.getenv("DD_SITE")):
-        pytest.skip("set DD_API_KEY and DD_SITE to run the Datadog logging suite")
+@pytest.fixture(scope="session")
+def datadog(client: LoggingClient) -> DatadogClient:
+    """The Datadog read-back client, or skip when the credentials are absent."""
+    if client.datadog is None:
+        pytest.skip("set DD_API_KEY, DD_SITE and DD_APP_KEY to run the Datadog logging suite")
+    return client.datadog

tests/e2e/logging/logging_client.py

@@ -1,48 +1,211 @@
-"""Client for the logging e2e suite: drive traffic and scrape the proxy's
-Prometheus ``/metrics`` endpoint.
+"""Client for the logging e2e suite.
 
-Holds the shared Gateway so the ``resources`` fixture cleans up keys it creates.
-``/metrics`` is exposed as plaintext (not a typed JSON body), so scraping goes
-through ``transport.probe`` and returns the raw exposition text for a Prometheus
-parser to read.
+Two jobs live here. The first is driving traffic through the proxy and scraping
+its Prometheus ``/metrics`` endpoint (plaintext, so it goes through
+``transport.probe``). The second is verifying Datadog delivery end to end: the
+proxy ships every request's StandardLoggingPayload to the Datadog logs intake on
+its ``datadog`` success/failure callback, and ``DatadogClient`` reads those events
+back out through the Datadog Logs Search API to prove they actually landed.
+
+The read-back is a real external call, so it still goes through the shared
+``HttpTransport`` (the only sanctioned path to ``requests``) - just pointed at the
+Datadog API host instead of the proxy. Verification is a poll, not a push: the
+proxy batches and flushes asynchronously (every 5s or at ``DD_MAX_BATCH_SIZE``)
+and Datadog then indexes for search, so a test drives traffic and waits for the
+marker it stamped to become searchable rather than "flushing" anything itself.
 """
 
 from __future__ import annotations
 
+import os
+import time
 from dataclasses import dataclass
 
+from pydantic import BaseModel, Field
+
+from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, REQUEST_TIMEOUT
 from e2e_gateway import Gateway, build_gateway
-from e2e_http import NoBody, unwrap
+from e2e_http import Headers, NoBody, Result, StreamingResponse, Success, unwrap
 from models import ChatBody, ChatMessage, ChatResponse, KeyGenerateBody
+from transport import HttpTransport
+
+
+class DatadogSearchHeaders(Headers):
+    """Auth + content type for the Datadog Logs API. Datadog authenticates reads
+    with an API key plus an *application* key (writes/intake need only the API
+    key), sent as their own headers rather than a bearer token."""
+
+    dd_api_key: str = Field(serialization_alias="DD-API-KEY")
+    dd_application_key: str = Field(serialization_alias="DD-APPLICATION-KEY")
+    content_type: str = Field(default="application/json", serialization_alias="Content-Type")
+
+
+class DatadogSearchFilter(BaseModel):
+    query: str
+    from_: str = Field(default="now-15m", serialization_alias="from")
+    to: str = "now"
+
+
+class DatadogSearchPage(BaseModel):
+    limit: int = 100
+
+
+class DatadogSearchBody(BaseModel):
+    """POST /api/v2/logs/events/search body. ``sort=-timestamp`` returns newest
+    first so a small ``page.limit`` still sees the events a test just produced."""
+
+    filter: DatadogSearchFilter
+    page: DatadogSearchPage = DatadogSearchPage()
+    sort: str = "-timestamp"
+
+
+class DatadogLogAttributes(BaseModel):
+    message: str | None = None
+    status: str | None = None
+    service: str | None = None
+    tags: list[str] = []
+    timestamp: str | None = None
+
+
+class DatadogLogEvent(BaseModel):
+    id: str | None = None
+    attributes: DatadogLogAttributes | None = None
+
+
+class DatadogSearchResponse(BaseModel):
+    data: list[DatadogLogEvent] = []
+
+
+DD_ERROR = "error"
+
+
+class ResponsesBody(BaseModel):
+    """Minimal POST /v1/responses body: the fields a logging test needs to drive a
+    real completion through the Responses API and get it shipped to Datadog."""
+
+    model: str
+    input: str
+
+
+@dataclass(frozen=True, slots=True)
+class DatadogClient:
+    """Reads litellm's shipped logs back out of Datadog to prove delivery.
+
+    Wraps an ``HttpTransport`` aimed at the Datadog API host (``api.<DD_SITE>``);
+    every call still flows through the shared e2e_http layer, so no test touches
+    ``requests``. Built from ``DD_API_KEY`` / ``DD_SITE`` / ``DD_APP_KEY`` in the
+    environment (see ``build_datadog_client``)."""
+
+    transport: HttpTransport
+    api_key: str
+    app_key: str
+    poll_timeout: float = POLL_TIMEOUT
+    poll_interval: float = POLL_INTERVAL
+
+    def _headers(self) -> DatadogSearchHeaders:
+        return DatadogSearchHeaders(dd_api_key=self.api_key, dd_application_key=self.app_key)
+
+    def search(self, query: str, *, limit: int = 100, window: str = "now-15m") -> list[DatadogLogEvent]:
+        """Every log event Datadog currently returns for ``query`` (free-text over
+        the log message plus facets like ``status:error``). Never raises: a failed
+        read yields an empty list so the caller keeps polling to its deadline."""
+        result: Result[DatadogSearchResponse] = self.transport.post(
+            "/api/v2/logs/events/search",
+            headers=self._headers(),
+            json=DatadogSearchBody(
+                filter=DatadogSearchFilter(query=query, from_=window),
+                page=DatadogSearchPage(limit=limit),
+            ),
+            response_type=DatadogSearchResponse,
+        )
+        match result:
+            case Success(data=payload):
+                return payload.data
+            case _:
+                return []
+
+    def poll_for_events(self, query: str, *, min_count: int = 1, window: str = "now-15m") -> list[DatadogLogEvent]:
+        """Poll the search API until at least ``min_count`` events match ``query``
+        or the deadline passes; returns whatever was seen on the last read."""
+        deadline = time.monotonic() + self.poll_timeout
+        events: list[DatadogLogEvent] = []
+        while time.monotonic() < deadline:
+            events = self.search(query, window=window)
+            if len(events) >= min_count:
+                return events
+            time.sleep(self.poll_interval)
+        return events
 
 
 @dataclass(frozen=True, slots=True)
 class LoggingClient:
     gateway: Gateway
+    datadog: DatadogClient | None
 
     def key_with_alias(self, alias: str, *, models: list[str]) -> str:
-        return self.gateway.generate_key(
-            KeyGenerateBody(key_alias=alias, models=models, user_id=f"e2e-{alias}")
-        )
+        return self.gateway.generate_key(KeyGenerateBody(key_alias=alias, models=models, user_id=f"e2e-{alias}"))
 
     def delete_key(self, key: str) -> None:
         self.gateway.delete_key(key)
 
-    def chat(self, key: str, model: str, text: str) -> ChatResponse:
-        return unwrap(
-            self.gateway.chat(
-                key,
-                ChatBody(
-                    model=model,
+    def chat_result(self, key: str, model: str, text: str) -> Result[ChatResponse]:
+        """The raw tagged-union outcome, so a test can assert on a failure instead
+        of turning it into one."""
+        return self.gateway.chat(
+            key,
+            ChatBody(
+                model=model,
                 messages=[ChatMessage(role="user", content=text)],
                 max_tokens=64,
-                ),
-            )
+            ),
+        )
+
+    def chat(self, key: str, model: str, text: str) -> ChatResponse:
+        return unwrap(self.chat_result(key, model, text))
+
+    def responses(self, key: str, model: str, text: str) -> StreamingResponse:
+        """Drive POST /v1/responses. Returns the raw outcome (the Responses body is
+        provider-native), so a test asserts on status and then verifies the log
+        reached Datadog rather than parsing the completion here."""
+        return self.gateway.transport.send(
+            "/v1/responses",
+            headers=self.gateway.transport.bearer(key),
+            json=ResponsesBody(model=model, input=text),
         )
 
     def scrape_metrics(self) -> str:
         return self.gateway.probe("/metrics", params=NoBody()).body
 
 
+def _datadog_api_base(dd_site: str) -> str:
+    """The Datadog API host for a site: ``us5.datadoghq.com`` -> the
+    ``api.us5.datadoghq.com`` reads host. Tolerates a site given with a scheme or
+    an already-``api.``-prefixed host."""
+    host = dd_site.strip().removeprefix("https://").removeprefix("http://").strip("/")
+    host = host if host.startswith("api.") else f"api.{host}"
+    return f"https://{host}"
+
+
+def build_datadog_client() -> DatadogClient | None:
+    """A ``DatadogClient`` when the read-back credentials are all present, else
+    ``None`` so the suite skips. ``DD_APP_KEY`` is required on top of the
+    ``DD_API_KEY`` / ``DD_SITE`` the proxy ships with, because the Logs Search API
+    rejects reads that carry only an API key."""
+    api_key = os.getenv("DD_API_KEY")
+    app_key = os.getenv("DD_APP_KEY")
+    dd_site = os.getenv("DD_SITE")
+    if not (api_key and app_key and dd_site):
+        return None
+    return DatadogClient(
+        transport=HttpTransport(
+            base_url=_datadog_api_base(dd_site),
+            master_key="",
+            request_timeout=REQUEST_TIMEOUT,
+        ),
+        api_key=api_key,
+        app_key=app_key,
+    )
+
+
 def build_logging_client() -> LoggingClient:
-    return LoggingClient(gateway=build_gateway())
+    return LoggingClient(gateway=build_gateway(), datadog=build_datadog_client())

tests/e2e/logging/test_datadog_e2e.py

@@ -0,0 +1,122 @@
+"""Live e2e: the proxy's ``datadog`` callback ships every request to Datadog.
+
+The proxy is configured with ``datadog`` in its callbacks, so each completion
+(success or failure) has its StandardLoggingPayload batched and flushed to the
+Datadog logs intake. These tests drive real traffic through the proxy, then read
+the events back out of the Datadog Logs Search API to prove they actually landed -
+delivery is verified at the destination, not by trusting the proxy.
+
+Each test stamps a unique marker into the prompts it sends. That marker rides
+along in the logged payload's ``messages``, so a free-text search for it isolates
+exactly this run's events from every other request flowing through the shared
+proxy (and across concurrent CI runs). Delivery is asynchronous - the proxy
+batches and flushes (every 5s or at ``DD_MAX_BATCH_SIZE``) and Datadog then
+indexes for search - so the read-back polls to a deadline instead of reading once.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import unique_marker
+from e2e_http import is_ok, require_successful_call
+from lifecycle import ResourceManager
+from logging_client import DD_ERROR, DatadogClient, LoggingClient
+from models import LiteLLMParamsBody
+
+pytestmark = pytest.mark.e2e
+
+MODEL = "gpt-5.5"
+ANTHROPIC_MODEL = "anthropic/claude-haiku-4-5"
+BATCH_REQUESTS = 10
+RESPONSES_REQUESTS = 5
+
+
+def _probe_prompt(run_marker: str, index: int) -> str:
+    """Unique per request so the cache never collapses two into one, but every
+    prompt carries ``run_marker`` so a single search finds the whole batch."""
+    return f"e2e datadog probe {run_marker} item {index} {unique_marker()}: reply with OK"
+
+
+class TestDatadogLogging:
+    @pytest.mark.covers("logging.datadog.success.exports_metric", exercised_on=["chat_completions"])
+    def test_datadog_can_flush_logs(
+        self, client: LoggingClient, resources: ResourceManager, datadog: DatadogClient
+    ) -> None:
+        """A batch of successful completions is delivered to Datadog with no drops:
+        driving N requests must produce at least N searchable log events carrying
+        this run's marker. A callback that stops shipping, drops on flush, or loses
+        the payload's message content fails here."""
+        run_marker = f"e2edd{unique_marker()}"
+        key = client.key_with_alias(f"e2e-dd-{run_marker}", models=[MODEL])
+        resources.defer(lambda: client.delete_key(key))
+
+        for index in range(BATCH_REQUESTS):
+            response = client.chat(key, MODEL, _probe_prompt(run_marker, index))
+            assert response.choices, f"empty completion for {run_marker} item {index}: {response}"
+
+        events = datadog.poll_for_events(run_marker, min_count=BATCH_REQUESTS)
+        assert len(events) >= BATCH_REQUESTS, (
+            f"Datadog returned {len(events)} events for marker {run_marker}, expected >= {BATCH_REQUESTS}; "
+            "the proxy's datadog callback dropped success logs or stopped shipping them"
+        )
+
+    @pytest.mark.covers("logging.datadog.success.exports_metric", exercised_on=["responses"])
+    def test_datadog_logs_responses_api_with_claude(
+        self, client: LoggingClient, resources: ResourceManager, datadog: DatadogClient
+    ) -> None:
+        """The Responses API on a Claude deployment is delivered to Datadog too:
+        the same no-drop contract as chat, but exercising litellm's
+        Responses-to-Anthropic translation and the /v1/responses logging path. A
+        callback that only ships /chat/completions would fail here."""
+        run_marker = f"e2eddresp{unique_marker()}"
+        model = f"dd-responses-{run_marker}"
+        model_id = client.gateway.create_model(
+            model,
+            LiteLLMParamsBody(model=ANTHROPIC_MODEL, api_key="os.environ/ANTHROPIC_API_KEY"),
+        )
+        resources.defer(lambda: client.gateway.delete_model(model_id))
+        key = client.key_with_alias(f"e2e-ddr-{run_marker}", models=[model])
+        resources.defer(lambda: client.delete_key(key))
+
+        for index in range(RESPONSES_REQUESTS):
+            result = client.responses(key, model, _probe_prompt(run_marker, index))
+            require_successful_call(result)
+
+        events = datadog.poll_for_events(run_marker, min_count=RESPONSES_REQUESTS)
+        assert len(events) >= RESPONSES_REQUESTS, (
+            f"Datadog returned {len(events)} events for responses marker {run_marker}, "
+            f"expected >= {RESPONSES_REQUESTS}; the proxy's datadog callback did not ship the "
+            "/v1/responses logs"
+        )
+
+    @pytest.mark.covers("logging.datadog.failure.exports_metric", exercised_on=["chat_completions"])
+    def test_datadog_logs_request_failures(
+        self, client: LoggingClient, resources: ResourceManager, datadog: DatadogClient
+    ) -> None:
+        """A failed completion is delivered to Datadog and classified as an error.
+        A deployment wired to a bad upstream key makes the provider reject the call;
+        the resulting failure must surface as a Datadog event tagged ``error`` that
+        still carries this run's marker."""
+        run_marker = f"e2eddfail{unique_marker()}"
+        bad_model = f"dd-fail-{run_marker}"
+        model_id = client.gateway.create_model(
+            bad_model,
+            LiteLLMParamsBody(model="openai/gpt-5.5", api_key="sk-invalid-e2e-datadog"),
+        )
+        resources.defer(lambda: client.gateway.delete_model(model_id))
+        key = client.key_with_alias(f"e2e-ddf-{run_marker}", models=[bad_model])
+        resources.defer(lambda: client.delete_key(key))
+
+        result = client.chat_result(key, bad_model, f"e2e datadog failure probe {run_marker}: reply with OK")
+        assert not is_ok(result), f"expected the bad-key deployment to reject the call, got {result}"
+
+        events = datadog.poll_for_events(run_marker, min_count=1)
+        assert events, (
+            f"Datadog returned no events for failed-request marker {run_marker}; "
+            "the proxy's datadog callback did not ship the failure"
+        )
+        statuses = [event.attributes.status for event in events if event.attributes]
+        assert DD_ERROR in statuses, (
+            f"failed request for {run_marker} was shipped to Datadog but not as an error (statuses seen: {statuses})"
+        )