diff --git a/tests/e2e/docker-compose.yml b/tests/e2e/docker-compose.yml index 2de38ca5c91..28441a602cb 100644 --- a/tests/e2e/docker-compose.yml +++ b/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 diff --git a/tests/e2e/logging/conftest.py b/tests/e2e/logging/conftest.py index 40c19aefca7..4d871644545 100644 --- a/tests/e2e/logging/conftest.py +++ b/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 diff --git a/tests/e2e/logging/logging_client.py b/tests/e2e/logging/logging_client.py index a3213fbdb00..83f4c1c3a36 100644 --- a/tests/e2e/logging/logging_client.py +++ b/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.``); + 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()) diff --git a/tests/e2e/logging/test_datadog_e2e.py b/tests/e2e/logging/test_datadog_e2e.py new file mode 100644 index 00000000000..446125a3154 --- /dev/null +++ b/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})" + )