From bfcb0ff4740622a359d1ff042b6c959edce95f63 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:46:06 -0700 Subject: [PATCH] feat(telemetry): record proxy requests and provider attempts into the telemetry sink (#45489) * feat(telemetry): record proxy requests and provider attempts into the telemetry sink * fix(telemetry): validate env settings, fix the window and only record the request's own provider calls Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(telemetry): bind the call id from a typed local Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(telemetry): record a finished request into the sink current when it finishes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(telemetry): record every failed provider attempt, including retries and fallbacks on a shared logging object Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(telemetry): report timeouts and connection failures as their own provider status classes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(telemetry): record failed attempts from the per-deployment failure hook instead of a new core logging hook Reverts the async_log_attempt_failure_event hook added to CustomLogger and litellm_logging in 3421b38685. The router already calls async_post_call_failure_deployment_hook once per failed attempt, retries and fallbacks included, so every failed fallback still gets its own row without touching the shared logging object. * fix(telemetry): register the attempt logger only once a telemetry group is turned on A proxy that never opts in now carries no telemetry callback. Once registered the logger stays, returning early while every group is off. --------- Co-authored-by: nate Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .github/workflows/test-unit.yml | 2 +- litellm/proxy/common_request_processing.py | 5 +- litellm/proxy/proxy_server.py | 23 ++ litellm/proxy/telemetry/__init__.py | 0 litellm/proxy/telemetry/attempt_logger.py | 233 +++++++++++++ litellm/proxy/telemetry/blocks.py | 69 ++++ litellm/proxy/telemetry/middleware.py | 189 +++++++++++ litellm/proxy/telemetry/request_context.py | 65 ++++ litellm/proxy/telemetry/runtime.py | 153 +++++++++ litellm/proxy/telemetry/settings.py | 77 +++++ tests/unit/proxy/telemetry/__init__.py | 0 .../proxy/telemetry/test_attempt_logger.py | 313 ++++++++++++++++++ tests/unit/proxy/telemetry/test_blocks.py | 37 +++ tests/unit/proxy/telemetry/test_middleware.py | 203 ++++++++++++ tests/unit/proxy/telemetry/test_runtime.py | 221 +++++++++++++ tests/unit/proxy/telemetry/test_settings.py | 95 ++++++ 16 files changed, 1683 insertions(+), 2 deletions(-) create mode 100644 litellm/proxy/telemetry/__init__.py create mode 100644 litellm/proxy/telemetry/attempt_logger.py create mode 100644 litellm/proxy/telemetry/blocks.py create mode 100644 litellm/proxy/telemetry/middleware.py create mode 100644 litellm/proxy/telemetry/request_context.py create mode 100644 litellm/proxy/telemetry/runtime.py create mode 100644 litellm/proxy/telemetry/settings.py create mode 100644 tests/unit/proxy/telemetry/__init__.py create mode 100644 tests/unit/proxy/telemetry/test_attempt_logger.py create mode 100644 tests/unit/proxy/telemetry/test_blocks.py create mode 100644 tests/unit/proxy/telemetry/test_middleware.py create mode 100644 tests/unit/proxy/telemetry/test_runtime.py create mode 100644 tests/unit/proxy/telemetry/test_settings.py diff --git a/.github/workflows/test-unit.yml b/.github/workflows/test-unit.yml index e043f011440..b020122ee3b 100644 --- a/.github/workflows/test-unit.yml +++ b/.github/workflows/test-unit.yml @@ -450,7 +450,7 @@ jobs: tests/unit/repositories tests/unit/secret_managers tests/unit/skills/test_skills_main.py - tests/unit/telemetry + tests/unit/telemetry tests/unit/proxy/telemetry tests/unit/tracing python-version: "3.12" workers: 2 diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 00729c12cc1..9b900b7c02b 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -121,6 +121,7 @@ from litellm.proxy.native_compaction import with_proxy_compaction_executor from litellm.proxy.route_llm_request import ( route_request, ) +from litellm.proxy.telemetry.request_context import bind_call_id as bind_telemetry_call_id from litellm.proxy.utils import ( # noqa: F401 # legacy module exports ProxyLogging, _check_and_merge_model_level_guardrails, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export @@ -2262,7 +2263,9 @@ class ProxyBaseLLMRequestProcessing: if alias_target is not None: self.data["model"] = alias_target - self.data["litellm_call_id"] = resolve_litellm_call_id(request.headers.get("x-litellm-call-id")) + litellm_call_id: Final = resolve_litellm_call_id(request.headers.get("x-litellm-call-id")) + self.data["litellm_call_id"] = litellm_call_id + bind_telemetry_call_id(litellm_call_id) DDSpanTagger.tag_call_id(self.data.get("litellm_call_id")) DDSpanTagger.tag_request( user_api_key_dict=user_api_key_dict, diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index bdaa4af4e26..190abfea70e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -756,6 +756,10 @@ from litellm.proxy.spend_tracking.spend_event_producer import ( SpendEventProducer, build_spend_event_producer, ) +from litellm.proxy.telemetry.middleware import TelemetryMiddleware +from litellm.proxy.telemetry.runtime import TelemetryRuntime +from litellm.proxy.telemetry.settings import describe_errors as describe_telemetry_errors +from litellm.proxy.telemetry.settings import load_settings as load_telemetry_settings try: from litellm.proxy.enterprise_billing.billing_metrics import ( @@ -1698,6 +1702,17 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[ProxyLifespanState state: Final[ProxyLifespanState] = {"tracing_receiver": receiver} from litellm.proxy.admin_mcp import admin_mcp_lifespan + telemetry_settings: Final = load_telemetry_settings() + if isinstance(telemetry_settings, ValidationError): + verbose_proxy_logger.warning( + "telemetry: invalid settings, leaving it off: %s", describe_telemetry_errors(telemetry_settings) + ) + else: + await telemetry_runtime.start( + litellm_version=version, + settings=telemetry_settings, + register=litellm.logging_callback_manager.add_litellm_callback, + ) try: async with AsyncExitStack() as admin_mcp_stack: try: @@ -1751,6 +1766,7 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[ProxyLifespanState verbose_proxy_logger.error("Error stopping the spend view setup task: %s", e) await _drain_spend_event_producer_on_shutdown() + await telemetry_runtime.stop() # Shutdown event - finish or cancel in-flight scheduled jobs before the shutdown flushes and the DB disconnect if scheduler is not None and scheduler_executor is not None: @@ -2609,6 +2625,13 @@ app.add_middleware( # it sees prisma_client as of the first request rather than import time. sink_factory=lambda: gateway_request_accumulator if prisma_client is not None else None, ) +telemetry_runtime: Final = TelemetryRuntime() +app.add_middleware( + TelemetryMiddleware, + sink_provider=lambda: telemetry_runtime.sink, + spawn=telemetry_runtime.spawn, + settle_timeout_s=lambda: telemetry_runtime.settings.settle_timeout_seconds, +) app.add_middleware(BudgetReservationReleaseMiddleware, release=release_unbound_budget_reservation) app.add_middleware(RedisRequestBatchMiddleware) app.add_middleware(LensInternalMiddleware) diff --git a/litellm/proxy/telemetry/__init__.py b/litellm/proxy/telemetry/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/proxy/telemetry/attempt_logger.py b/litellm/proxy/telemetry/attempt_logger.py new file mode 100644 index 00000000000..f0825b9596c --- /dev/null +++ b/litellm/proxy/telemetry/attempt_logger.py @@ -0,0 +1,233 @@ +from collections.abc import Callable, Mapping +from typing import Final, TypeAlias + +from pydantic import BaseModel, ConfigDict, ValidationError + +from litellm._logging import verbose_proxy_logger +from litellm.exceptions import APIConnectionError, BadRequestError, Timeout +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider +from litellm.proxy.telemetry.blocks import count_blocks +from litellm.proxy.telemetry.request_context import AttemptObservation, RequestAccumulator, current_request +from litellm.telemetry.records import AttemptRecord, BlockCounts, StatusClass, TokenCounts +from litellm.telemetry.sink import TelemetrySink +from litellm.types.utils import CallTypes + +DeploymentHasher: TypeAlias = Callable[[str], str] + + +class _PromptTokensDetails(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + cached_tokens: int | None = None + + +class _Usage(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + prompt_tokens_details: _PromptTokensDetails | None = None + cache_read_input_tokens: int | None = None + + +class _Metadata(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + usage_object: _Usage | None = None + + +class _LoggedAttempt(BaseModel): + """The slice of ``StandardLoggingPayload`` telemetry reads""" + + model_config = ConfigDict(frozen=True, extra="ignore") + + custom_llm_provider: str | None = None + model_id: str | None = None + stream: bool | None = None + startTime: float + endTime: float + completionStartTime: float | None = None + prompt_tokens: int = 0 + completion_tokens: int = 0 + cache_hit: bool | None = None + metadata: _Metadata | None = None + + +class _ModelInfo(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + id: str | None = None + + +class _CallMetadata(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + model_info: _ModelInfo | None = None + + +class _LoggedParams(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + proxy_rejected_before_routing: bool = False + + +class _FailedCall(BaseModel): + """The slice of one deployment attempt's own kwargs telemetry reads""" + + model_config = ConfigDict(frozen=True, extra="ignore") + + model: str | None = None + custom_llm_provider: str | None = None + stream: bool | None = None + metadata: _CallMetadata | None = None + litellm_metadata: _CallMetadata | None = None + + @property + def model_id(self) -> str | None: + infos: Final = (meta.model_info for meta in (self.litellm_metadata, self.metadata) if meta is not None) + return next((info.id for info in infos if info is not None and info.id is not None), None) + + @property + def provider(self) -> str | None: + if self.custom_llm_provider is not None or self.model is None: + return self.custom_llm_provider + try: + return get_llm_provider(self.model)[1] + except BadRequestError: + return None + + +def _rejected_before_routing(litellm_params: object) -> bool: + try: + return _LoggedParams.model_validate(litellm_params).proxy_rejected_before_routing + except ValidationError: + return False + + +def _cache_read_tokens(metadata: _Metadata | None) -> int: + usage: Final = metadata.usage_object if metadata is not None else None + if usage is None: + return 0 + details: Final = usage.prompt_tokens_details + return (details.cached_tokens if details is not None else None) or usage.cache_read_input_tokens or 0 + + +def observe(logged: _LoggedAttempt, *, hash_deployment: DeploymentHasher) -> AttemptObservation: + stream: Final = logged.stream is True + first_token_s: Final = logged.completionStartTime if stream else None + return AttemptObservation( + attempt=AttemptRecord( + provider=logged.custom_llm_provider or "unknown", + provider_status=StatusClass.SUCCESS, + stream=stream, + deployment_hash=hash_deployment(logged.model_id) if logged.model_id else None, + latency_to_first_token_ms=(first_token_s - logged.startTime) * 1000 if first_token_s is not None else None, + ), + succeeded=True, + tokens=TokenCounts( + input=logged.prompt_tokens, + output=logged.completion_tokens, + cache_read=_cache_read_tokens(logged.metadata), + ), + litellm_cache_hit=logged.cache_hit is True, + ) + + +def _failure_status(exception: Exception) -> StatusClass: + if isinstance(exception, Timeout): + return StatusClass.TIMEOUT + if isinstance(exception, APIConnectionError): + return StatusClass.CONNECTION_ERROR + status_code: Final[object] = getattr(exception, "status_code", None) + return StatusClass.from_status_code(status_code if isinstance(status_code, int) else None) + + +def observe_failure( + call: _FailedCall, exception: Exception, *, hash_deployment: DeploymentHasher +) -> AttemptObservation: + model_id: Final = call.model_id + return AttemptObservation( + attempt=AttemptRecord( + provider=call.provider or "unknown", + provider_status=_failure_status(exception), + stream=call.stream is True, + deployment_hash=hash_deployment(model_id) if model_id else None, + ), + succeeded=False, + ) + + +class TelemetryAttemptLogger(CustomLogger): + """Records the provider calls made for the in-flight proxy request itself + + Failures come from the deployment-level hook, which runs once per attempt (retries and fallbacks included); + the request-level failure event runs once per shared logging object, so it would miss all but one + + Calls with another ``litellm_call_id`` (guardrails, moderation, background jobs) and calls outside a request + are skipped + """ + + def __init__( + self, + sink: Callable[[], TelemetrySink | None], + hash_deployment: DeploymentHasher, + *, + blocks_enabled: Callable[[], bool] = lambda: False, + block_counter: Callable[[object], BlockCounts | None] = count_blocks, + ) -> None: + super().__init__() # pyright: ignore[reportUnknownMemberType] # base callback constructor accepts untyped kwargs + self._sink: Final = sink + self._hash_deployment: Final = hash_deployment + self._blocks_enabled: Final = blocks_enabled + self._block_counter: Final = block_counter + + def _owning_request(self, kwargs: Mapping[str, object]) -> tuple[TelemetrySink, RequestAccumulator] | None: + sink: Final = self._sink() + request: Final[RequestAccumulator | None] = current_request.get() + if sink is None or request is None or not request.owns(kwargs.get("litellm_call_id")): + return None + if _rejected_before_routing(kwargs.get("litellm_params")): + return None + return sink, request + + def _add( + self, + sink: TelemetrySink, + request: RequestAccumulator, + kwargs: Mapping[str, object], + observation: AttemptObservation, + ) -> None: + if not observation.litellm_cache_hit: + sink.record_attempt(observation.attempt) + if self._blocks_enabled(): + request.count_blocks_once(lambda: self._block_counter(kwargs.get("messages"))) + request.add(observation) + + async def async_log_success_event( + self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + owner: Final = self._owning_request(kwargs) + if owner is None: + return + try: + logged: Final = _LoggedAttempt.model_validate(kwargs.get("standard_logging_object")) + except ValidationError: + verbose_proxy_logger.debug("telemetry: skipping attempt without a standard logging payload") + return + self._add(*owner, kwargs, observe(logged, hash_deployment=self._hash_deployment)) + + async def async_post_call_failure_deployment_hook( + self, + request_data: Mapping[str, object], + exception: Exception, + call_type: CallTypes | None, + fallback_depth: int | None = None, + ) -> None: + owner: Final = self._owning_request(request_data) + if owner is None: + return + try: + call: Final = _FailedCall.model_validate(request_data) + except ValidationError: + verbose_proxy_logger.debug("telemetry: skipping a failed attempt without call details") + return + self._add(*owner, request_data, observe_failure(call, exception, hash_deployment=self._hash_deployment)) diff --git a/litellm/proxy/telemetry/blocks.py b/litellm/proxy/telemetry/blocks.py new file mode 100644 index 00000000000..46a3f70194b --- /dev/null +++ b/litellm/proxy/telemetry/blocks.py @@ -0,0 +1,69 @@ +from collections import Counter +from collections.abc import Iterator +from typing import Final + +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError + +from litellm.telemetry.records import BlockCounts, BlockType + +_PART_TYPES: Final = { + "text": BlockType.TEXT, + "input_text": BlockType.TEXT, + "image": BlockType.IMAGE, + "image_url": BlockType.IMAGE, + "input_image": BlockType.IMAGE, + "input_audio": BlockType.AUDIO, + "audio": BlockType.AUDIO, + "file": BlockType.FILE, + "document": BlockType.FILE, + "input_file": BlockType.FILE, + "tool_use": BlockType.TOOL_USE, + "tool_result": BlockType.TOOL_RESULT, + "thinking": BlockType.THINKING, + "redacted_thinking": BlockType.THINKING, +} + + +class _Part(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + type: str + + +class _Message(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + role: str + content: str | tuple[_Part, ...] | None = None + tool_calls: tuple[object, ...] | None = None + + +_MESSAGES: Final = TypeAdapter(tuple[_Message, ...]) + + +def _message_blocks(message: _Message) -> Iterator[BlockType]: + match message.content: + case str() if message.role == "tool": + yield BlockType.TOOL_RESULT + case str(): + yield BlockType.TEXT + case tuple() as parts: + yield from (_PART_TYPES.get(part.type, BlockType.OTHER) for part in parts) + case None: + pass + yield from (BlockType.TOOL_USE for _ in message.tool_calls or ()) + + +def _all_blocks(messages: tuple[_Message, ...]) -> Iterator[BlockType]: + for message in messages: + yield from _message_blocks(message) + + +def count_blocks(messages: object) -> BlockCounts | None: + """Content block counts for OpenAI or Anthropic style ``messages``, or None when they are not messages""" + try: + parsed: Final = _MESSAGES.validate_python(messages) + except ValidationError: + return None + counts: Final = Counter(_all_blocks(parsed)) + return BlockCounts(total=counts.total(), by_type=tuple(sorted(counts.items()))) diff --git a/litellm/proxy/telemetry/middleware.py b/litellm/proxy/telemetry/middleware.py new file mode 100644 index 00000000000..0cf0da92996 --- /dev/null +++ b/litellm/proxy/telemetry/middleware.py @@ -0,0 +1,189 @@ +import time +from collections.abc import Callable, Coroutine, Mapping +from dataclasses import dataclass +from typing import Final, TypeAlias + +from pydantic import TypeAdapter, ValidationError +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from litellm._logging import verbose_proxy_logger +from litellm.proxy.middleware.billable_request_metrics_middleware import classify_billable_request +from litellm.proxy.telemetry.request_context import AttemptObservation, RequestAccumulator, current_request +from litellm.telemetry.records import BlockCounts, RequestRecord, StatusClass, TokenCounts +from litellm.telemetry.sink import TelemetrySink + +RUST_RESPONSE_HEADER: Final = b"x-litellm-rust" + +Spawn: TypeAlias = Callable[[Coroutine[None, None, None]], None] + + +@dataclass(frozen=True, slots=True) +class ResponseTiming: + status_code: int + stream: bool + to_headers_ms: float | None + to_first_byte_ms: float | None + handled_by_rust: bool = False + + +def _final_observation(observations: tuple[AttemptObservation, ...]) -> AttemptObservation | None: + succeeded: Final = tuple(observation for observation in observations if observation.succeeded) + return (succeeded or observations or (None,))[-1] + + +def build_request_record( + *, + endpoint: str, + timing: ResponseTiming, + header_keys: frozenset[str], + observations: tuple[AttemptObservation, ...], + blocks: BlockCounts | None = None, +) -> RequestRecord: + final: Final = _final_observation(observations) + return RequestRecord( + endpoint=endpoint, + stream=timing.stream, + litellm_status=StatusClass.from_status_code(timing.status_code), + provider=final.attempt.provider if final is not None else None, + deployment_hash=final.attempt.deployment_hash if final is not None else None, + provider_status=final.attempt.provider_status if final is not None else StatusClass.NONE, + litellm_cache_hit=final is not None and final.litellm_cache_hit, + handled_by_rust=timing.handled_by_rust, + provider_cache_hit=final is not None and final.tokens.cache_read > 0, + provider_attempts=sum(not observation.litellm_cache_hit for observation in observations), + tokens=final.tokens if final is not None else TokenCounts(), + latency_to_headers_ms=timing.to_headers_ms, + latency_to_first_byte_ms=timing.to_first_byte_ms, + blocks=blocks, + header_keys=header_keys, + ) + + +_HEADERS: Final = TypeAdapter(tuple[tuple[bytes, bytes], ...]) + + +def _headers(asgi_mapping: Mapping[str, object]) -> tuple[tuple[bytes, bytes], ...]: + try: + return _HEADERS.validate_python(asgi_mapping.get("headers", ())) + except ValidationError: + return () + + +def _str_field(asgi_mapping: Mapping[str, object], key: str, default: str) -> str: + value: Final = asgi_mapping.get(key) + return value if isinstance(value, str) else default + + +def _handled_by_rust(headers: tuple[tuple[bytes, bytes], ...]) -> bool: + return any(name.lower() == RUST_RESPONSE_HEADER and value == b"true" for name, value in headers) + + +def _is_event_stream(headers: tuple[tuple[bytes, bytes], ...]) -> bool: + return any(name.lower() == b"content-type" and value.startswith(b"text/event-stream") for name, value in headers) + + +class _ResponseObserver: + def __init__(self, clock: Callable[[], float]) -> None: + self._clock: Final = clock + self.started: Final = clock() + self.status_code: int = 500 + self.stream: bool = False + self.handled_by_rust: bool = False + self.headers_at: float | None = None + self.first_body_at: float | None = None + + def observe(self, message: Mapping[str, object]) -> None: + match message.get("type"): + case "http.response.start": + status: Final = message.get("status") + self.status_code = status if isinstance(status, int) else self.status_code + self.headers_at = self._clock() + headers: Final = _headers(message) + self.stream = _is_event_stream(headers) + self.handled_by_rust = _handled_by_rust(headers) + case "http.response.body" if self.first_body_at is None and message.get("body"): + self.first_body_at = self._clock() + case _: + pass + + def timing(self) -> ResponseTiming: + return ResponseTiming( + status_code=self.status_code, + stream=self.stream, + to_headers_ms=_elapsed_ms(self.started, self.headers_at), + to_first_byte_ms=_elapsed_ms(self.started, self.first_body_at), + handled_by_rust=self.handled_by_rust, + ) + + +def _elapsed_ms(started: float, at: float | None) -> float | None: + return (at - started) * 1000 if at is not None else None + + +class TelemetryMiddleware: + """Times each LLM, MCP and A2A request at the ASGI layer and joins it with the provider attempts it made""" + + def __init__( + self, + app: ASGIApp, + sink_provider: Callable[[], TelemetrySink | None], + spawn: Spawn, + *, + settle_timeout_s: Callable[[], float] = lambda: 2.0, + clock: Callable[[], float] = time.perf_counter, + ) -> None: + self.app: Final = app + self._sink_provider: Final = sink_provider + self._spawn: Final = spawn + self._settle_timeout_s: Final = settle_timeout_s + self._clock: Final = clock + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + sink: Final = self._sink_provider() if scope["type"] == "http" else None + classification: Final = ( + classify_billable_request(_str_field(scope, "path", ""), _str_field(scope, "method", "POST")) + if sink is not None + else None + ) + if sink is None or classification is None: + await self.app(scope, receive, send) + return + + observer: Final = _ResponseObserver(self._clock) + accumulator: Final = RequestAccumulator() + + async def send_wrapper(message: Message) -> None: + observer.observe(message) + await send(message) + + token: Final = current_request.set(accumulator) + try: + await self.app(scope, receive, send_wrapper) + finally: + current_request.reset(token) + header_keys: Final = frozenset(name.decode("latin-1").lower() for name, _ in _headers(scope)) + self._spawn(self._finish(classification[1], observer.timing(), header_keys, accumulator)) + + async def _finish( + self, + endpoint: str, + timing: ResponseTiming, + header_keys: frozenset[str], + accumulator: RequestAccumulator, + ) -> None: + await accumulator.wait_for_success(self._settle_timeout_s()) + sink: Final = self._sink_provider() + if sink is None: + return + try: + sink.record_request( + build_request_record( + endpoint=endpoint, + timing=timing, + header_keys=header_keys, + observations=accumulator.observations, + blocks=accumulator.blocks, + ) + ) + except Exception: # noqa: BLE001 -- telemetry must never surface into request handling + verbose_proxy_logger.debug("telemetry: failed to record request for %s", endpoint, exc_info=True) diff --git a/litellm/proxy/telemetry/request_context.py b/litellm/proxy/telemetry/request_context.py new file mode 100644 index 00000000000..b37dcad5f86 --- /dev/null +++ b/litellm/proxy/telemetry/request_context.py @@ -0,0 +1,65 @@ +import asyncio +import contextlib +from collections.abc import Callable +from contextvars import ContextVar +from dataclasses import dataclass, field +from typing import Final + +from litellm.telemetry.records import AttemptRecord, BlockCounts, TokenCounts + + +@dataclass(frozen=True, slots=True) +class AttemptObservation: + attempt: AttemptRecord + succeeded: bool + tokens: TokenCounts = field(default_factory=TokenCounts) + litellm_cache_hit: bool = False + + +class RequestAccumulator: + """Collects the provider attempts logged for one proxy request's own ``litellm_call_id``""" + + def __init__(self) -> None: + self._observations: tuple[AttemptObservation, ...] = () + self._succeeded: Final = asyncio.Event() + self._call_id: str | None = None + self._blocks: BlockCounts | None = None + self._blocks_counted: bool = False + + def bind(self, call_id: str) -> None: + if self._call_id is None: + self._call_id = call_id + + def owns(self, call_id: object) -> bool: + return self._call_id is not None and call_id == self._call_id + + @property + def blocks(self) -> BlockCounts | None: + return self._blocks + + def count_blocks_once(self, count: Callable[[], BlockCounts | None]) -> None: + if not self._blocks_counted: + self._blocks_counted = True + self._blocks = count() + + @property + def observations(self) -> tuple[AttemptObservation, ...]: + return self._observations + + def add(self, observation: AttemptObservation) -> None: + self._observations = (*self._observations, observation) + if observation.succeeded: + self._succeeded.set() + + async def wait_for_success(self, timeout_s: float) -> None: + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(self._succeeded.wait(), timeout_s) + + +current_request: Final[ContextVar[RequestAccumulator | None]] = ContextVar("litellm_telemetry_request", default=None) + + +def bind_call_id(call_id: str) -> None: + request: Final = current_request.get() + if request is not None: + request.bind(call_id) diff --git a/litellm/proxy/telemetry/runtime.py b/litellm/proxy/telemetry/runtime.py new file mode 100644 index 00000000000..a3746ad8bb7 --- /dev/null +++ b/litellm/proxy/telemetry/runtime.py @@ -0,0 +1,153 @@ +import asyncio +import contextlib +import hashlib +import hmac +import secrets +import uuid +from collections.abc import Awaitable, Callable, Coroutine +from typing import Final, TypeAlias + +import httpx + +from litellm._logging import verbose_proxy_logger +from litellm.llms.custom_httpx.http_handler import ( + get_async_httpx_client, # pyright: ignore[reportUnknownVariableType] # legacy params: dict signature +) +from litellm.proxy.common_utils.encrypt_decrypt_utils import get_salt_key +from litellm.proxy.telemetry.attempt_logger import TelemetryAttemptLogger +from litellm.proxy.telemetry.settings import FLUSH_INTERVAL_SECONDS, EnvPolicy, TelemetrySettings, env_policy +from litellm.telemetry.aggregate import AggregatingSink +from litellm.telemetry.consent import OFF, ConsentGatedSink, TelemetryConsent +from litellm.telemetry.http_exporter import HttpExporter +from litellm.telemetry.records import InstanceInfo, TelemetryGroup +from litellm.telemetry.sink import Exporter +from litellm.types.llms.custom_http import httpxSpecialProvider + + +def shared_http_client() -> httpx.AsyncClient: + return get_async_httpx_client(httpxSpecialProvider.LoggingCallback, params={"timeout": 10.0}).client + + +StoredConsent: TypeAlias = Callable[[], Awaitable[TelemetryConsent | None]] + + +async def nothing_stored() -> TelemetryConsent | None: + return None + + +def proxy_hash_secret() -> bytes: + salt_key: Final = get_salt_key() + return salt_key.encode() if salt_key is not None else secrets.token_bytes(32) + + +def deployment_hasher(secret: bytes) -> Callable[[str], str]: + key: Final = hashlib.sha256(b"litellm-telemetry-deployment:" + secret).digest() + return lambda model_id: hmac.new(key, model_id.encode(), hashlib.sha256).hexdigest()[:16] + + +class TelemetryRuntime: + """Owns the proxy's telemetry sink, re-reads which groups are on before every window, and flushes each window""" + + def __init__(self) -> None: + self.sink: ConsentGatedSink | None = None + self.settings: TelemetrySettings = TelemetrySettings.model_construct() + self.policy: EnvPolicy | None = None + self._attach_logger: Callable[[], None] | None = None + self._flush_task: asyncio.Task[None] | None = None + self._flushing: asyncio.Future[None] | None = None + self._pending: Final[set[asyncio.Task[None]]] = set() # mutable-ok: strong refs keep finalizers alive + + def spawn(self, coroutine: Coroutine[None, None, None]) -> None: + task: Final = asyncio.create_task(coroutine) + self._pending.add(task) + task.add_done_callback(self._pending.discard) + + async def start( + self, + *, + litellm_version: str, + settings: TelemetrySettings, + register: Callable[[TelemetryAttemptLogger], None], + stored: StoredConsent = nothing_stored, + http_client: Callable[[], httpx.AsyncClient] = shared_http_client, + hash_secret: Callable[[], bytes] = proxy_hash_secret, + flush_interval_s: float = FLUSH_INTERVAL_SECONDS, + ) -> None: + self.settings = settings + policy: Final = env_policy(settings) + self.policy = policy + if policy.locked_off: + return + if settings.endpoint is None: + return + exporter: Final = HttpExporter(http_client(), str(settings.endpoint)) + instance: Final = InstanceInfo(instance_id=uuid.uuid4().hex, litellm_version=litellm_version) + self._attach_logger = lambda: register( + TelemetryAttemptLogger( + lambda: self.sink, deployment_hasher(hash_secret()), blocks_enabled=self._counts_blocks + ) + ) + await self._refresh(policy, stored, exporter, instance) + self._flush_task = asyncio.create_task(self._flush_every(policy, stored, exporter, instance, flush_interval_s)) + + def _counts_blocks(self) -> bool: + sink: Final = self.sink + return sink is not None and sink.consent.allows(TelemetryGroup.EVENT_DETAILS) + + async def _refresh( + self, policy: EnvPolicy, stored: StoredConsent, exporter: Exporter, instance: InstanceInfo + ) -> None: + current: Final = self.sink + try: + consent: Final = policy.effective(await stored()) + except Exception: # noqa: BLE001 -- a failed settings read keeps the current groups rather than flapping off + verbose_proxy_logger.debug("telemetry: could not read stored settings", exc_info=True) + return + if current is not None and current.consent == consent: + return + if consent == OFF: + self.sink = None + return + gated: Final = ConsentGatedSink(AggregatingSink(exporter), consent) + gated.set_instance(instance) + self._attach_logger_once() + self.sink = gated + + def _attach_logger_once(self) -> None: + """Registers the attempt logger the first time a group is on, so a proxy that never opts in carries no callback. + It stays registered after groups are turned off, returning early, since removing it could skip in-flight calls""" + attach: Final = self._attach_logger + self._attach_logger = None + if attach is not None: + attach() + + async def _flush_every( + self, policy: EnvPolicy, stored: StoredConsent, exporter: Exporter, instance: InstanceInfo, interval_s: float + ) -> None: + while True: + await asyncio.sleep(interval_s) + await self._flush_window() + await self._refresh(policy, stored, exporter, instance) + + async def _flush_window(self) -> None: + flushing: Final = asyncio.ensure_future(self._flush_current()) + self._flushing = flushing + await asyncio.shield(flushing) + + async def _flush_current(self) -> None: + sink: Final = self.sink + if sink is not None: + await sink.flush() + + async def stop(self) -> None: + flush_task: Final = self._flush_task + if flush_task is not None: + flush_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await flush_task + flushing: Final = self._flushing + if flushing is not None: + await asyncio.gather(flushing, return_exceptions=True) + await asyncio.gather(*self._pending, return_exceptions=True) + await self._flush_current() + self.sink = None diff --git a/litellm/proxy/telemetry/settings.py b/litellm/proxy/telemetry/settings.py new file mode 100644 index 00000000000..b10ecac9e79 --- /dev/null +++ b/litellm/proxy/telemetry/settings.py @@ -0,0 +1,77 @@ +"""Environment policy for telemetry: the veto, the pinned groups, and which variables an admin should be told about""" + +from dataclasses import dataclass +from typing import Annotated, Final + +from pydantic import AfterValidator, Field, HttpUrl, ValidationError +from pydantic_settings import BaseSettings, SettingsConfigDict + +from litellm.telemetry.consent import OFF, TelemetryConsent, parse_consent + +ENV_PREFIX: Final = "LITELLM_TELEMETRY_" +FLUSH_INTERVAL_SECONDS: Final = 300.0 + + +def _pinned_consent(groups: str) -> TelemetryConsent: + consent: Final = parse_consent(groups.split(",")) + if not isinstance(consent, TelemetryConsent): + raise ValueError(consent.message()) # noqa: TRY004 # pydantic only turns ValueError into a ValidationError + return consent + + +def _valid_groups(groups: str | None) -> str | None: + if groups is not None: + _ = _pinned_consent(groups) + return groups + + +class TelemetrySettings(BaseSettings): + """``LITELLM_TELEMETRY_*`` env vars""" + + model_config = SettingsConfigDict(env_prefix=ENV_PREFIX, case_sensitive=False, extra="ignore", frozen=True) + + disabled: bool = False + groups: Annotated[str | None, AfterValidator(_valid_groups)] = None + endpoint: HttpUrl | None = None + settle_timeout_seconds: float = Field(default=2.0, ge=0) + + +def load_settings() -> TelemetrySettings | ValidationError: + try: + return TelemetrySettings() + except ValidationError as e: + return e + + +def describe_errors(error: ValidationError) -> str: + """Field names and messages only: the rejected value of ``endpoint`` can carry a token""" + return "; ".join( + f"{ENV_PREFIX}{'.'.join(map(str, detail['loc'])).upper()}: {detail['msg']}" + for detail in error.errors(include_input=False, include_url=False) + ) + + +@dataclass(frozen=True, slots=True) +class EnvPolicy: + vetoed: bool + pinned: TelemetryConsent | None + set_variables: tuple[str, ...] + + def effective(self, stored: TelemetryConsent | None) -> TelemetryConsent: + if self.vetoed: + return OFF + if self.pinned is not None: + return self.pinned + return stored if stored is not None else OFF + + @property + def locked_off(self) -> bool: + return self.vetoed or self.pinned == OFF + + +def env_policy(settings: TelemetrySettings) -> EnvPolicy: + return EnvPolicy( + vetoed=settings.disabled, + pinned=_pinned_consent(settings.groups) if settings.groups is not None else None, + set_variables=tuple(sorted(f"{ENV_PREFIX}{name.upper()}" for name in settings.model_fields_set)), + ) diff --git a/tests/unit/proxy/telemetry/__init__.py b/tests/unit/proxy/telemetry/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/proxy/telemetry/test_attempt_logger.py b/tests/unit/proxy/telemetry/test_attempt_logger.py new file mode 100644 index 00000000000..a171cf2b75d --- /dev/null +++ b/tests/unit/proxy/telemetry/test_attempt_logger.py @@ -0,0 +1,313 @@ +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime +from typing import Final + +import pytest + +import litellm +from litellm import Router +from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging +from litellm.proxy.telemetry.attempt_logger import TelemetryAttemptLogger +from litellm.proxy.telemetry.request_context import RequestAccumulator, bind_call_id, current_request +from litellm.telemetry.records import ( + AttemptRecord, + BlockCounts, + BlockType, + InstanceInfo, + RequestRecord, + StatusClass, + TokenCounts, + UIEvent, +) +from litellm.types.llms.openai import AllMessageValues, ChatCompletionUserMessage + + +@dataclass +class _RecordingSink: + attempts: tuple[AttemptRecord, ...] = () + + def set_instance(self, info: InstanceInfo) -> None: ... + + def record_request(self, record: RequestRecord) -> None: ... + + def record_attempt(self, record: AttemptRecord) -> None: + self.attempts = (*self.attempts, record) + + def record_ui_event(self, event: UIEvent) -> None: ... + + async def flush(self) -> None: ... + + +_CALL_ID: Final = "call-1" + + +def _request() -> RequestAccumulator: + request: Final = RequestAccumulator() + request.bind(_CALL_ID) + return request + + +def _payload() -> Mapping[str, object]: + return { + "custom_llm_provider": "anthropic", + "model_id": "deployment-1", + "stream": True, + "startTime": 100.0, + "endTime": 101.5, + "completionStartTime": 100.25, + "prompt_tokens": 30, + "completion_tokens": 7, + "cache_hit": None, + "metadata": {"usage_object": {"prompt_tokens_details": {"cached_tokens": 20}}}, + } + + +@pytest.mark.asyncio +async def test_a_logged_stream_success_becomes_an_attempt_and_joins_the_in_flight_request() -> None: + sink: Final = _RecordingSink() + logger: Final = TelemetryAttemptLogger( + lambda: sink, hash_deployment=lambda model_id: f"h({model_id})", blocks_enabled=lambda: True + ) + request: Final = _request() + token: Final = current_request.set(request) + try: + await logger.async_log_success_event( + { + "litellm_call_id": _CALL_ID, + "standard_logging_object": _payload(), + "messages": [{"role": "user", "content": "hi"}], + }, + None, + None, + None, + ) + finally: + current_request.reset(token) + + expected: Final = AttemptRecord( + provider="anthropic", + provider_status=StatusClass.SUCCESS, + stream=True, + deployment_hash="h(deployment-1)", + latency_to_first_token_ms=250.0, + ) + assert sink.attempts == (expected,) + (observation,) = request.observations + assert observation.attempt == expected + assert observation.succeeded + assert observation.tokens == TokenCounts(input=30, output=7, cache_read=20) + assert request.blocks == BlockCounts(total=1, by_type=((BlockType.TEXT, 1),)) + + +def _failed_call(provider: str = "anthropic", model_id: str = "deployment-1") -> dict[str, object]: + return { + "litellm_call_id": _CALL_ID, + "model": f"{provider}/claude", + "stream": True, + "metadata": {"model_info": {"id": model_id}}, + } + + +@pytest.mark.parametrize( + ("exception", "expected_status"), + [ + (litellm.RateLimitError("429", "anthropic", "claude"), StatusClass.CLIENT_ERROR), + (litellm.ServiceUnavailableError("503", "anthropic", "claude"), StatusClass.SERVER_ERROR), + (ValueError("no status"), StatusClass.NONE), + (litellm.Timeout("timed out", "claude", "anthropic"), StatusClass.TIMEOUT), + (litellm.APIConnectionError("refused", "anthropic", "claude"), StatusClass.CONNECTION_ERROR), + ], +) +@pytest.mark.asyncio +async def test_a_failed_attempt_carries_the_provider_status_class( + exception: Exception, expected_status: StatusClass +) -> None: + sink: Final = _RecordingSink() + logger: Final = TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: f"h({model_id})") + token: Final = current_request.set(_request()) + try: + await logger.async_post_call_failure_deployment_hook(_failed_call(), exception, None) + finally: + current_request.reset(token) + assert sink.attempts == ( + AttemptRecord( + provider="anthropic", provider_status=expected_status, stream=True, deployment_hash="h(deployment-1)" + ), + ) + + +@pytest.mark.asyncio +async def test_every_failed_fallback_on_a_shared_logging_object_is_its_own_attempt( + monkeypatch: pytest.MonkeyPatch, +) -> None: + sink: Final = _RecordingSink() + logger: Final = TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id) + monkeypatch.setattr(litellm, "callbacks", [logger]) + router: Final = Router( + model_list=[ + { + "model_name": "a", + "litellm_params": {"model": "anthropic/a", "api_key": "k", "mock_response": "litellm.RateLimitError"}, + "model_info": {"id": "a"}, + }, + { + "model_name": "b", + "litellm_params": { + "model": "bedrock/b", + "api_key": "k", + "mock_response": "litellm.InternalServerError", + }, + "model_info": {"id": "b"}, + }, + { + "model_name": "c", + "litellm_params": {"model": "openai/c", "api_key": "k", "mock_timeout": True, "timeout": 0.01}, + "model_info": {"id": "c"}, + }, + ], + fallbacks=[{"a": ["b", "c"]}], + num_retries=0, + ) + messages: Final[list[AllMessageValues]] = [ChatCompletionUserMessage(role="user", content="hi")] + shared_logging_obj: Final = LitellmLogging( + model="a", + messages=messages, + stream=False, + call_type="acompletion", + start_time=datetime(2026, 1, 1), + litellm_call_id=_CALL_ID, + function_id=_CALL_ID, + ) + request: Final = _request() + token: Final = current_request.set(request) + try: + with pytest.raises(litellm.RateLimitError): + await router.acompletion( # pyright: ignore[reportUnknownMemberType] # Router.acompletion's overloads return partially untyped responses + model="a", messages=messages, litellm_call_id=_CALL_ID, litellm_logging_obj=shared_logging_obj + ) + finally: + current_request.reset(token) + assert [(a.provider, a.deployment_hash, a.provider_status) for a in sink.attempts] == [ + ("anthropic", "a", StatusClass.CLIENT_ERROR), + ("bedrock", "b", StatusClass.SERVER_ERROR), + ("openai", "c", StatusClass.TIMEOUT), + ] + assert len(request.observations) == 3 + + +@pytest.mark.asyncio +async def test_a_request_level_failure_event_is_not_counted_again() -> None: + sink: Final = _RecordingSink() + request: Final = _request() + token: Final = current_request.set(request) + try: + logger = TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id) + await logger.async_log_failure_event( # pyright: ignore[reportUnknownMemberType] # the base callback's signature is untyped + _failed_call(), None, None, None + ) + finally: + current_request.reset(token) + assert sink.attempts == () + assert request.observations == () + + +@pytest.mark.asyncio +async def test_a_call_without_a_standard_logging_payload_is_skipped() -> None: + sink: Final = _RecordingSink() + await TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id).async_log_success_event( + {}, None, None, None + ) + assert sink.attempts == () + + +@pytest.mark.asyncio +async def test_a_litellm_cache_hit_joins_the_request_but_is_not_a_provider_attempt() -> None: + sink: Final = _RecordingSink() + logger: Final = TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: f"h({model_id})") + request: Final = _request() + token: Final = current_request.set(request) + try: + await logger.async_log_success_event( + {"litellm_call_id": _CALL_ID, "standard_logging_object": {**_payload(), "cache_hit": True}}, + None, + None, + None, + ) + finally: + current_request.reset(token) + + assert sink.attempts == () + (observation,) = request.observations + assert observation.litellm_cache_hit + + +@pytest.mark.asyncio +async def test_a_call_outside_any_proxy_request_is_not_recorded() -> None: + sink: Final = _RecordingSink() + await TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id).async_log_success_event( + {"litellm_call_id": _CALL_ID, "standard_logging_object": _payload()}, None, None, None + ) + assert sink.attempts == () + + +@pytest.mark.parametrize("call_id", ["guardrail-call", None]) +@pytest.mark.asyncio +async def test_a_guardrail_or_moderation_call_inside_a_request_is_not_recorded(call_id: str | None) -> None: + sink: Final = _RecordingSink() + request: Final = _request() + token: Final = current_request.set(request) + try: + await TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id).async_log_success_event( + {"litellm_call_id": call_id, "standard_logging_object": _payload()}, None, None, None + ) + finally: + current_request.reset(token) + assert sink.attempts == () + assert request.observations == () + + +@pytest.mark.asyncio +async def test_calls_before_the_proxy_binds_its_call_id_are_not_recorded() -> None: + sink: Final = _RecordingSink() + request: Final = RequestAccumulator() + token: Final = current_request.set(request) + logger: Final = TelemetryAttemptLogger(lambda: sink, hash_deployment=lambda model_id: model_id) + try: + await logger.async_log_success_event( + {"litellm_call_id": "pre-call-guardrail", "standard_logging_object": _payload()}, None, None, None + ) + bind_call_id(_CALL_ID) + bind_call_id("later-call") + await logger.async_log_success_event( + {"litellm_call_id": _CALL_ID, "standard_logging_object": _payload()}, None, None, None + ) + finally: + current_request.reset(token) + assert len(sink.attempts) == 1 + assert len(request.observations) == 1 + + +@pytest.mark.parametrize("enabled", [True, False]) +@pytest.mark.asyncio +async def test_blocks_are_counted_once_per_request_and_only_when_enabled(enabled: bool) -> None: + counted: list[object] = [] # mutable-ok: records each messages payload handed to the block counter + + def _counter(messages: object) -> BlockCounts: + counted.append(messages) + return BlockCounts(total=1, by_type=((BlockType.TEXT, 1),)) + + sink: Final = _RecordingSink() + request: Final = _request() + logger: Final = TelemetryAttemptLogger( + lambda: sink, hash_deployment=lambda model_id: model_id, blocks_enabled=lambda: enabled, block_counter=_counter + ) + kwargs: Final = {"litellm_call_id": _CALL_ID, "standard_logging_object": _payload(), "messages": ["m"]} + token: Final = current_request.set(request) + try: + await logger.async_post_call_failure_deployment_hook(kwargs, ValueError("first attempt failed"), None) + await logger.async_log_success_event(kwargs, None, None, None) + finally: + current_request.reset(token) + assert counted == ([["m"]] if enabled else []) + assert (request.blocks is not None) is enabled diff --git a/tests/unit/proxy/telemetry/test_blocks.py b/tests/unit/proxy/telemetry/test_blocks.py new file mode 100644 index 00000000000..d48501d81a8 --- /dev/null +++ b/tests/unit/proxy/telemetry/test_blocks.py @@ -0,0 +1,37 @@ +from typing import Final + +import pytest + +from litellm.proxy.telemetry.blocks import count_blocks +from litellm.telemetry.records import BlockCounts, BlockType + + +def test_counts_string_content_part_lists_tool_calls_and_tool_messages() -> None: + messages: Final = [ + {"role": "system", "content": "be brief"}, + { + "role": "user", + "content": [ + {"type": "text", "text": "what is in this picture"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}, + {"type": "something_new"}, + ], + }, + {"role": "assistant", "content": None, "tool_calls": [{"id": "a"}, {"id": "b"}]}, + {"role": "tool", "content": "42", "tool_call_id": "a"}, + ] + assert count_blocks(messages) == BlockCounts( + total=7, + by_type=( + (BlockType.IMAGE, 1), + (BlockType.OTHER, 1), + (BlockType.TEXT, 2), + (BlockType.TOOL_RESULT, 1), + (BlockType.TOOL_USE, 2), + ), + ) + + +@pytest.mark.parametrize("messages", [None, "a prompt string", [{"content": "no role"}]]) +def test_anything_that_is_not_a_message_list_has_no_block_counts(messages: object) -> None: + assert count_blocks(messages) is None diff --git a/tests/unit/proxy/telemetry/test_middleware.py b/tests/unit/proxy/telemetry/test_middleware.py new file mode 100644 index 00000000000..6274f29a952 --- /dev/null +++ b/tests/unit/proxy/telemetry/test_middleware.py @@ -0,0 +1,203 @@ +from collections.abc import Callable, Coroutine +from dataclasses import dataclass +from typing import Final + +import pytest +from httpx import ASGITransport, AsyncClient +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse, Response, StreamingResponse +from starlette.routing import Route + +from litellm.proxy.telemetry.middleware import TelemetryMiddleware +from litellm.proxy.telemetry.request_context import AttemptObservation, current_request +from litellm.telemetry.records import ( + AttemptRecord, + InstanceInfo, + RequestRecord, + StatusClass, + TokenCounts, + UIEvent, +) + + +@dataclass +class _RecordingSink: + requests: tuple[RequestRecord, ...] = () + + def set_instance(self, info: InstanceInfo) -> None: ... + + def record_request(self, record: RequestRecord) -> None: + self.requests = (*self.requests, record) + + def record_attempt(self, record: AttemptRecord) -> None: ... + + def record_ui_event(self, event: UIEvent) -> None: ... + + async def flush(self) -> None: ... + + +@dataclass +class _Spawned: + coroutines: tuple[Coroutine[None, None, None], ...] = () + + def __call__(self, coroutine: Coroutine[None, None, None]) -> None: + self.coroutines = (*self.coroutines, coroutine) + + async def drain(self) -> None: + for coroutine in self.coroutines: + await coroutine + + +@dataclass +class _Clock: + now: float = 0.0 + + def __call__(self) -> float: + self.now += 0.1 + return self.now + + +def _attempt(provider: str, status: StatusClass, *, succeeded: bool, tokens: TokenCounts) -> AttemptObservation: + return AttemptObservation( + attempt=AttemptRecord(provider=provider, provider_status=status, stream=False), + succeeded=succeeded, + tokens=tokens, + ) + + +async def _retried_then_served(request: Request) -> Response: + accumulator: Final = current_request.get() + assert accumulator is not None + accumulator.add(_attempt("azure", StatusClass.SERVER_ERROR, succeeded=False, tokens=TokenCounts())) + accumulator.add(_attempt("openai", StatusClass.SUCCESS, succeeded=True, tokens=TokenCounts(input=12, output=3))) + return JSONResponse({"ok": True}) + + +async def _streamed(request: Request) -> Response: + async def chunks(): + yield b"data: 1\n\n" + yield b"data: [DONE]\n\n" + + return StreamingResponse(chunks(), media_type="text/event-stream") + + +async def _served_by_rust(request: Request) -> Response: + return JSONResponse({"ok": True}, headers={"x-litellm-rust": "true"}) + + +async def _rejected(request: Request) -> Response: + return JSONResponse({"error": "bad key"}, status_code=401) + + +def _client( + sink: _RecordingSink | Callable[[], _RecordingSink | None], spawned: _Spawned, *, settle_timeout_s: float = 0.0 +) -> AsyncClient: + app: Final = Starlette( + routes=[ + Route("/v1/chat/completions", _retried_then_served, methods=["POST"]), + Route("/v1/messages", _streamed, methods=["POST"]), + Route("/v1/embeddings", _rejected, methods=["POST"]), + Route("/v1/responses", _served_by_rust, methods=["POST"]), + Route("/health", _rejected, methods=["GET"]), + ] + ) + wrapped: Final = TelemetryMiddleware( + app, + sink_provider=sink if callable(sink) else lambda: sink, + spawn=spawned, + settle_timeout_s=lambda: settle_timeout_s, + clock=_Clock(), + ) + return AsyncClient(transport=ASGITransport(app=wrapped), base_url="http://proxy") + + +@pytest.mark.asyncio +async def test_a_retried_request_reports_the_serving_attempt_and_the_attempt_count() -> None: + sink: Final = _RecordingSink() + spawned: Final = _Spawned() + async with _client(sink, spawned) as client: + response: Final = await client.post( + "/v1/chat/completions", json={}, headers={"Anthropic-Beta": "x", "Authorization": "Bearer sk-1"} + ) + assert response.status_code == 200, response.text + await spawned.drain() + + (record,) = sink.requests + assert record.endpoint == "/chat/completions" + assert (record.provider, record.provider_status, record.provider_attempts) == ("openai", StatusClass.SUCCESS, 2) + assert record.litellm_status is StatusClass.SUCCESS + assert record.tokens == TokenCounts(input=12, output=3) + assert record.stream is False + assert record.latency_to_first_byte_ms is not None + assert {"anthropic-beta", "authorization"} <= record.header_keys + + +@pytest.mark.asyncio +async def test_a_stream_reports_time_to_headers_before_the_first_byte() -> None: + sink: Final = _RecordingSink() + spawned: Final = _Spawned() + async with _client(sink, spawned) as client: + response: Final = await client.post("/v1/messages", json={}) + assert response.status_code == 200, response.text + await spawned.drain() + + (record,) = sink.requests + assert record.stream is True + assert record.latency_to_headers_ms is not None + assert record.latency_to_first_byte_ms is not None + assert record.latency_to_headers_ms < record.latency_to_first_byte_ms + assert (record.provider, record.provider_status, record.provider_attempts) == (None, StatusClass.NONE, 0) + + +@pytest.mark.asyncio +async def test_a_request_rejected_before_any_provider_call_has_no_provider_status() -> None: + sink: Final = _RecordingSink() + spawned: Final = _Spawned() + async with _client(sink, spawned, settle_timeout_s=0.01) as client: + response: Final = await client.post("/v1/embeddings", json={}) + assert response.status_code == 401, response.text + await spawned.drain() + + (record,) = sink.requests + assert (record.litellm_status, record.provider_status) == (StatusClass.CLIENT_ERROR, StatusClass.NONE) + + +@pytest.mark.asyncio +async def test_routes_outside_llm_mcp_and_a2a_are_not_recorded() -> None: + sink: Final = _RecordingSink() + spawned: Final = _Spawned() + async with _client(sink, spawned) as client: + await client.get("/health") + assert spawned.coroutines == () + assert sink.requests == () + + +@pytest.mark.asyncio +async def test_only_a_response_carrying_the_rust_header_counts_as_handled_by_rust() -> None: + sink: Final = _RecordingSink() + spawned: Final = _Spawned() + async with _client(sink, spawned) as client: + rust_response: Final = await client.post("/v1/responses", json={}) + python_response: Final = await client.post("/v1/messages", json={}) + assert (rust_response.status_code, python_response.status_code) == (200, 200) + await spawned.drain() + assert [(record.endpoint, record.handled_by_rust) for record in sink.requests] == [ + ("/responses", True), + ("/v1/messages", False), + ] + + +@pytest.mark.asyncio +async def test_a_request_finishing_after_the_groups_change_is_recorded_into_the_current_sink() -> None: + before: Final = _RecordingSink() + after: Final = _RecordingSink() + sinks: Final = [before] # mutable-ok: the runtime swaps its sink between windows + spawned: Final = _Spawned() + async with _client(lambda: sinks[-1], spawned) as client: + response: Final = await client.post("/v1/chat/completions", json={}) + assert response.status_code == 200, response.text + sinks.append(after) + await spawned.drain() + + assert (before.requests, len(after.requests)) == ((), 1) diff --git a/tests/unit/proxy/telemetry/test_runtime.py b/tests/unit/proxy/telemetry/test_runtime.py new file mode 100644 index 00000000000..5ba5aafb6f1 --- /dev/null +++ b/tests/unit/proxy/telemetry/test_runtime.py @@ -0,0 +1,221 @@ +import asyncio +from typing import Final + +import httpx +import pytest +from pydantic import HttpUrl + +from litellm.proxy.telemetry.attempt_logger import TelemetryAttemptLogger +from litellm.proxy.telemetry.runtime import TelemetryRuntime, deployment_hasher +from litellm.proxy.telemetry.settings import TelemetrySettings +from litellm.telemetry.consent import ConsentGatedSink, TelemetryConsent +from litellm.telemetry.records import AttemptRecord, InstanceInfo, RequestRecord, TelemetryGroup, UIEvent + +_ENDPOINT: Final = HttpUrl("http://telemetry.invalid/v1/reports") + + +def _offline() -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(lambda _request: httpx.Response(204))) + + +_HEARTBEAT: Final = TelemetryConsent(frozenset({TelemetryGroup.HEARTBEAT})) + + +async def _stored_heartbeat() -> TelemetryConsent | None: + return _HEARTBEAT + + +async def _broken_store() -> TelemetryConsent | None: + raise ConnectionError("db down") + + +@pytest.mark.parametrize( + "settings", + [ + TelemetrySettings(), + TelemetrySettings(groups="", endpoint=_ENDPOINT), + TelemetrySettings(groups="heartbeat"), + TelemetrySettings(groups="heartbeat", endpoint=_ENDPOINT, disabled=True), + ], +) +@pytest.mark.asyncio +async def test_telemetry_stays_off_unless_valid_groups_and_an_endpoint_are_set_and_not_vetoed( + settings: TelemetrySettings, +) -> None: + registered: Final[list[TelemetryAttemptLogger]] = [] # mutable-ok: captures the register callback + runtime: Final = TelemetryRuntime() + await runtime.start(litellm_version="1.0.0", settings=settings, register=registered.append, http_client=_offline) + assert runtime.sink is None + assert registered == [] + + +@pytest.mark.asyncio +async def test_pinned_groups_register_the_attempt_logger_and_stop_cleanly() -> None: + registered: Final[list[TelemetryAttemptLogger]] = [] # mutable-ok: captures the register callback + runtime: Final = TelemetryRuntime() + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(groups="HEARTBEAT", endpoint=_ENDPOINT), + flush_interval_s=3600, + register=registered.append, + http_client=_offline, + ) + assert runtime.sink is not None and runtime.sink.consent == _HEARTBEAT + assert len(registered) == 1 + await runtime.stop() + assert runtime.sink is None + + +@pytest.mark.asyncio +async def test_stored_groups_apply_when_no_groups_are_pinned_and_a_veto_ignores_them() -> None: + enabled: Final = TelemetryRuntime() + await enabled.start( + litellm_version="1.0.0", + settings=TelemetrySettings(endpoint=_ENDPOINT), + flush_interval_s=3600, + register=lambda _logger: None, + http_client=_offline, + stored=_stored_heartbeat, + ) + vetoed: Final = TelemetryRuntime() + await vetoed.start( + litellm_version="1.0.0", + settings=TelemetrySettings(endpoint=_ENDPOINT, disabled=True), + register=lambda _logger: None, + http_client=_offline, + stored=_stored_heartbeat, + ) + assert enabled.sink is not None and enabled.sink.consent == _HEARTBEAT + assert vetoed.sink is None + await enabled.stop() + + +@pytest.mark.asyncio +async def test_a_failing_settings_store_leaves_telemetry_off_instead_of_crashing_startup() -> None: + runtime: Final = TelemetryRuntime() + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(endpoint=_ENDPOINT), + flush_interval_s=3600, + register=lambda _logger: None, + http_client=_offline, + stored=_broken_store, + ) + assert runtime.sink is None + await runtime.stop() + + +@pytest.mark.asyncio +async def test_the_attempt_logger_is_registered_only_once_a_group_is_turned_on() -> None: + registered: Final[list[TelemetryAttemptLogger]] = [] # mutable-ok: captures the register callback + stored: list[TelemetryConsent | None] = [None] # mutable-ok: the stored groups the admin changes between windows + + async def _stored() -> TelemetryConsent | None: + return stored[0] + + runtime: Final = TelemetryRuntime() + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(endpoint=_ENDPOINT), + flush_interval_s=0.01, + register=registered.append, + http_client=_offline, + stored=_stored, + ) + await asyncio.sleep(0.05) + assert runtime.sink is None + assert registered == [] + + for consent in (_HEARTBEAT, None, _HEARTBEAT): + stored[0] = consent + await asyncio.sleep(0.05) + assert (runtime.sink is not None) is (consent is not None) + assert len(registered) == 1 + await runtime.stop() + + +def test_deployment_hashes_are_stable_per_install_and_differ_across_installs() -> None: + assert deployment_hasher(b"secret-a")("model-1") == deployment_hasher(b"secret-a")("model-1") + assert deployment_hasher(b"secret-a")("model-1") != deployment_hasher(b"secret-b")("model-1") + assert "model-1" not in deployment_hasher(b"secret-a")("model-1") + + +@pytest.mark.asyncio +async def test_stop_waits_for_in_flight_request_finalizers_before_the_last_flush() -> None: + runtime: Final = TelemetryRuntime() + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(groups="heartbeat", endpoint=_ENDPOINT), + flush_interval_s=3600, + register=lambda _logger: None, + http_client=_offline, + ) + finished: Final[list[bool]] = [] # mutable-ok: records that the finalizer ran to completion + + async def _finalizer() -> None: + await asyncio.sleep(0.01) + finished.append(True) + + runtime.spawn(_finalizer()) + await runtime.stop() + assert finished == [True] + + +@pytest.mark.asyncio +async def test_stopping_a_runtime_that_never_started_is_a_no_op() -> None: + runtime: Final = TelemetryRuntime() + await runtime.stop() + assert runtime.sink is None + + +class _HeldFlushSink: + def __init__(self) -> None: + self.started: Final = asyncio.Event() + self.release: Final = asyncio.Event() + self.flushes: Final[list[bool]] = [] # mutable-ok: records each flush that ran to completion + + def set_instance(self, info: InstanceInfo) -> None: ... + + def record_request(self, record: RequestRecord) -> None: ... + + def record_attempt(self, record: AttemptRecord) -> None: ... + + def record_ui_event(self, event: UIEvent) -> None: ... + + async def flush(self) -> None: + self.started.set() + await self.release.wait() + self.flushes.append(True) + + +@pytest.mark.asyncio +async def test_stop_during_an_export_lets_that_export_finish() -> None: + runtime: Final = TelemetryRuntime() + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(groups="heartbeat", endpoint=_ENDPOINT), + flush_interval_s=0.001, + register=lambda _logger: None, + http_client=_offline, + ) + held: Final = _HeldFlushSink() + runtime.sink = ConsentGatedSink(held, _HEARTBEAT) + await held.started.wait() + stopping: Final = asyncio.create_task(runtime.stop()) + await asyncio.sleep(0) + held.release.set() + await stopping + assert len(held.flushes) == 2 + + +async def test_settle_timeout_reads_the_configured_value_after_start_and_the_default_before() -> None: + runtime: Final = TelemetryRuntime() + assert runtime.settings.settle_timeout_seconds == 2.0 + await runtime.start( + litellm_version="1.0.0", + settings=TelemetrySettings(groups="heartbeat", endpoint=_ENDPOINT, settle_timeout_seconds=7.5), + register=lambda _logger: None, + http_client=_offline, + ) + assert runtime.settings.settle_timeout_seconds == 7.5 + await runtime.stop() diff --git a/tests/unit/proxy/telemetry/test_settings.py b/tests/unit/proxy/telemetry/test_settings.py new file mode 100644 index 00000000000..cc219d2583e --- /dev/null +++ b/tests/unit/proxy/telemetry/test_settings.py @@ -0,0 +1,95 @@ +import pytest + +from pydantic import ValidationError + +from litellm.proxy.telemetry.settings import EnvPolicy, TelemetrySettings, describe_errors, env_policy, load_settings +from litellm.telemetry.consent import OFF, TelemetryConsent +from litellm.telemetry.records import TelemetryGroup + +_HEARTBEAT = TelemetryConsent(frozenset({TelemetryGroup.HEARTBEAT})) +_STORED = TelemetryConsent(frozenset({TelemetryGroup.HEARTBEAT, TelemetryGroup.REQUEST_SUCCESS})) + + +def _policy(settings: TelemetrySettings) -> EnvPolicy: + policy = env_policy(settings) + assert isinstance(policy, EnvPolicy), policy + return policy + + +def test_without_env_vars_the_stored_groups_apply_and_nothing_is_reported_as_set() -> None: + policy = _policy(TelemetrySettings()) + assert policy.effective(_STORED) is _STORED + assert policy.effective(None) == OFF + assert policy.set_variables == () + assert policy.locked_off is False + + +def test_the_veto_wins_over_both_pinned_and_stored_groups() -> None: + policy = _policy(TelemetrySettings(disabled=True, groups="heartbeat")) + assert policy.effective(_STORED) == OFF + assert policy.locked_off is True + + +def test_pinned_groups_override_stored_groups() -> None: + policy = _policy(TelemetrySettings(groups="Heartbeat, ")) + assert policy.effective(_STORED) == _HEARTBEAT + assert policy.locked_off is False + + +def test_an_empty_groups_variable_pins_telemetry_off() -> None: + policy = _policy(TelemetrySettings(groups="")) + assert policy.effective(_STORED) == OFF + assert policy.locked_off is True + + +@pytest.mark.parametrize( + ("groups", "message"), + [ + ("heartbeat,full", "unknown telemetry group 'full'"), + ("token_info", "telemetry group 'token_info' needs 'request_success' turned on too"), + ], +) +def test_invalid_pinned_groups_fail_settings_validation( + monkeypatch: pytest.MonkeyPatch, groups: str, message: str +) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_GROUPS", groups) + error = load_settings() + assert isinstance(error, ValidationError) + assert message in describe_errors(error) + + +@pytest.mark.parametrize("endpoint", ["not a url", "ftp://telemetry.example/v1", "telemetry.example/v1"]) +def test_a_non_http_endpoint_fails_settings_validation(monkeypatch: pytest.MonkeyPatch, endpoint: str) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_ENDPOINT", endpoint) + assert isinstance(load_settings(), ValidationError) + + +def test_validation_errors_name_the_variable_but_never_echo_its_value(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_ENDPOINT", "ftp://secret-token@telemetry.example") + error = load_settings() + assert isinstance(error, ValidationError) + described = describe_errors(error) + assert "LITELLM_TELEMETRY_ENDPOINT" in described + assert "secret-token" not in described + + +def test_the_flush_interval_is_not_configurable(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_FLUSH_INTERVAL_SECONDS", "1") + settings = load_settings() + assert isinstance(settings, TelemetrySettings) + assert "flush_interval_seconds" not in TelemetrySettings.model_fields + + +def test_set_variables_lists_every_telemetry_env_var_present_but_never_its_value( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_ENDPOINT", "https://secret-token@telemetry.example") + monkeypatch.setenv("litellm_telemetry_disabled", "1") + policy = _policy(TelemetrySettings()) + assert policy.set_variables == ("LITELLM_TELEMETRY_DISABLED", "LITELLM_TELEMETRY_ENDPOINT") + assert policy.vetoed is True + + +def test_a_negative_settle_timeout_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_TELEMETRY_SETTLE_TIMEOUT_SECONDS", "-1") + assert isinstance(load_settings(), ValidationError)