scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-14 04:09:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 97

feat(telemetry): record proxy requests and provider attempts into the telemetry sink (#45489)

Browse source 
* 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 <nate@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-13 18:46:06 -07:00 • committed by GitHub
parent 18b756fd7d
commit bfcb0ff474
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 1683 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/test-unit.yml vendored
View file

@ -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

5
litellm/proxy/common_request_processing.py
View file

@ -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,

23
litellm/proxy/proxy_server.py
View file

@ -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)

0
litellm/proxy/telemetry/__init__.py Normal file
View file

233
litellm/proxy/telemetry/attempt_logger.py Normal file
View file

@ -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))

69
litellm/proxy/telemetry/blocks.py Normal file
View file

@ -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())))

189
litellm/proxy/telemetry/middleware.py Normal file
View file

@ -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)

65
litellm/proxy/telemetry/request_context.py Normal file
View file

@ -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)

153
litellm/proxy/telemetry/runtime.py Normal file
View file

@ -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

77
litellm/proxy/telemetry/settings.py Normal file
View file

@ -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)),
)

0
tests/unit/proxy/telemetry/__init__.py Normal file
View file

313
tests/unit/proxy/telemetry/test_attempt_logger.py Normal file
View file

@ -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

37
tests/unit/proxy/telemetry/test_blocks.py Normal file
View file

@ -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

203
tests/unit/proxy/telemetry/test_middleware.py Normal file
View file

@ -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)

221
tests/unit/proxy/telemetry/test_runtime.py Normal file
View file

@ -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()

95
tests/unit/proxy/telemetry/test_settings.py Normal file
View file

@ -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)