From 29806dee1647d7fb5cad0996a3fd83046fbbda2c Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Sat, 8 Aug 2026 12:20:29 -0700 Subject: [PATCH] refactor(otel/v2): satisfy the type-discipline and strict-lint ceilings Never-rebound names carry Final, deliberate rebinds carry rebind-ok with the reason, and the endpoint suffix check collapses into one endswith call. Two hook parameters and the constructor kwargs move from Any to object, which is both stricter and what the bodies actually rely on. The remaining constructions keep a mutable-ok naming what needs a concrete dict: pydantic model_copy payloads, the OTel SDK Resource call, and the LRU caches that are mutable by design. --- .../integrations/otel/destination_logger.py | 31 +++--- litellm/integrations/otel/emitter.py | 8 +- litellm/integrations/otel/logger.py | 32 +++--- litellm/integrations/otel/model/config.py | 4 +- .../integrations/otel/plumbing/providers.py | 7 +- litellm/integrations/otel/plumbing/routing.py | 100 ++++++++++-------- litellm/integrations/otel/presets/__init__.py | 10 +- litellm/integrations/otel/presets/agentops.py | 13 ++- litellm/integrations/otel/presets/arize.py | 21 ++-- .../integrations/otel/presets/destinations.py | 6 +- litellm/integrations/otel/presets/langfuse.py | 12 ++- litellm/integrations/otel/presets/levo.py | 2 +- litellm/integrations/otel/presets/phoenix.py | 7 +- litellm/integrations/otel/presets/weave.py | 12 ++- litellm/litellm_core_utils/litellm_logging.py | 19 ++-- .../proxy/_experimental/mcp_server/server.py | 8 +- litellm/proxy/auth/user_api_key_auth.py | 2 +- .../test_logging_exporter_access.py | 32 ++++++ .../proxy/test_litellm_pre_call_utils.py | 22 ++++ 19 files changed, 228 insertions(+), 120 deletions(-) diff --git a/litellm/integrations/otel/destination_logger.py b/litellm/integrations/otel/destination_logger.py index 6c6150474d1..d1d788da05d 100644 --- a/litellm/integrations/otel/destination_logger.py +++ b/litellm/integrations/otel/destination_logger.py @@ -13,7 +13,7 @@ owning logger, and including it here would export the same call twice. from collections.abc import Mapping from datetime import datetime from functools import lru_cache -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Final from opentelemetry.trace import Span @@ -34,14 +34,14 @@ if TYPE_CHECKING: def _vocabulary_config(backend: str) -> OpenTelemetryV2Config: """The backend's span vocabulary with no exporter of its own.""" - preset_fn = PRESET_BY_CALLBACK.get(backend) + preset_fn: Final = PRESET_BY_CALLBACK.get(backend) if preset_fn is None: return OpenTelemetryV2Config() try: - config = preset_fn(allow_missing_credentials=True) + config: Final = preset_fn(allow_missing_credentials=True) except Exception: # noqa: BLE001 # an unbuildable preset still has a usable default vocabulary return OpenTelemetryV2Config() - return config.model_copy(update={"exporters": ()}) + return config.model_copy(update={"exporters": ()}) # mutable-ok: pydantic model_copy takes a plain update mapping class _DestinationOnlyOtel(OpenTelemetryV2): @@ -89,7 +89,7 @@ class _DestinationOnlyOtel(OpenTelemetryV2): Emitting a second one here would deliver the same discovery call twice; returning ``True`` still stops the caller from closing it as an LLM call. """ - raw_payload = kwargs.get("standard_logging_object") + raw_payload: Final = kwargs.get("standard_logging_object") return isinstance(raw_payload, Mapping) and is_mcp_list_tools(raw_payload) def export_to_destinations( @@ -114,15 +114,15 @@ class _DestinationOnlyOtel(OpenTelemetryV2): class AdminDestinationLogger(CustomLogger): """Delivers each request's gen-AI span to the destinations its identity resolved.""" - def __init__(self, **kwargs: Any) -> None: + def __init__(self, **kwargs: object) -> None: # kwargs-ok: CustomLogger's constructor signature super().__init__(**kwargs) self._emitters: dict[str, _DestinationOnlyOtel] = {} # mutable-ok: bounded per-backend emitter cache def _emitter_for(self, backend: str) -> _DestinationOnlyOtel: - existing = self._emitters.get(backend) + existing: Final = self._emitters.get(backend) if existing is not None: return existing - emitter = _DestinationOnlyOtel(config=_vocabulary_config(backend), callback_name=backend) + emitter: Final = _DestinationOnlyOtel(config=_vocabulary_config(backend), callback_name=backend) self._emitters[backend] = emitter return emitter @@ -132,8 +132,11 @@ class AdminDestinationLogger(CustomLogger): start_time: "datetime | float | None", end_time: "datetime | float | None", ) -> None: - owned = otel_v2_owned_backends() - for backend in sorted({d.callback_name for d in request_destinations() if d.callback_name} - owned): + owned: Final = otel_v2_owned_backends() + for backend in sorted( + {d.callback_name for d in request_destinations() if d.callback_name} + - owned # mutable-ok: handed to a model or SDK that needs a concrete dict + ): # mutable-ok: construction is handed to an API that needs a concrete dict try: self._emitter_for(backend).export_to_destinations(kwargs, start_time, end_time) except Exception as exc: # noqa: BLE001 # one destination's failure must not break the request or the others @@ -142,7 +145,7 @@ class AdminDestinationLogger(CustomLogger): async def async_log_success_event( self, kwargs: "Mapping[str, Any]", - response_obj: Any, + response_obj: object, start_time: "datetime | float | None", end_time: "datetime | float | None", ) -> None: @@ -151,7 +154,7 @@ class AdminDestinationLogger(CustomLogger): async def async_log_failure_event( self, kwargs: "Mapping[str, Any]", - response_obj: Any, + response_obj: object, start_time: "datetime | float | None", end_time: "datetime | float | None", ) -> None: @@ -165,7 +168,7 @@ def admin_destination_logger() -> AdminDestinationLogger: def register_admin_destination_logger() -> None: """Put the destination sink on the proxy's async callback lists, once.""" - sink = admin_destination_logger() - for bucket in (litellm._async_success_callback, litellm._async_failure_callback): + sink: Final = admin_destination_logger() + for bucket in (litellm._async_success_callback, litellm._async_failure_callback): # pyright: ignore[reportPrivateUsage] # the proxy's own callback buckets; the sink must register on them like any built-in logger if not any(callback is sink for callback in bucket): bucket.append(sink) diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index 70c02b47cc9..c176213be30 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -238,16 +238,16 @@ class SpanEmitter: # alone coalesced the successful attempt into the failed one it replaced and the # retried call reached its destinations only as the failure. The response id splits # the attempts apart while a sync+async double-firing of one attempt still shares it. - attempt_id = data.response_id if isinstance(data, LLMCallSpanData) else None - dedup_key = ( + attempt_id: Final = data.response_id if isinstance(data, LLMCallSpanData) else None + dedup_key: Final = ( (f"{data.identity.call_id}:{attempt_id}" if attempt_id else data.identity.call_id) if isinstance(data, (LLMCallSpanData, MCPToolCallSpanData)) else None ) if self._seen(dedup_key, role): return None - name = _NAME_BUILDERS[role](data) - first: Span | None = None + name: Final = _NAME_BUILDERS[role](data) + first: Span | None = None # rebind-ok: set to the first span produced below for tracer in tracers: span = self.start_span( role, diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 2a98b82d7d0..19502dfd01e 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -212,7 +212,7 @@ class OpenTelemetryV2(CustomLogger): other backend, and each one's own exporter went dark. Matching the name as well lets each backend register itself while still de-duplicating a repeat of itself. """ - already_otel = any( + already_otel: Final = any( cb.__class__.__module__.startswith(_OTEL_MODULES) and (not per_backend or getattr(cb, "callback_name", None) == self.callback_name) for cb in callbacks @@ -290,15 +290,15 @@ class OpenTelemetryV2(CustomLogger): # by the previous attempt is cleared here; leaving it made the close callback # short-circuit and the successful attempt after a failure went untraced. self._closed_call_ids.pop(call_id, None) - start_time_ns = to_ns(datetime.now()) - spans: tuple[Span, ...] = () + start_time_ns: Final = to_ns(datetime.now()) + spans: tuple[Span, ...] = () # rebind-ok: reassigned on the branch below # Parent to the request's anchored root span (stable across the request), # falling back to ambient on the SDK path. Open the span live only when # that resolves to a recordable parent; otherwise defer to the close # callback (the thread-pool case, where the anchor isn't visible here). parent_context: Final = resolve_request_span_context() if is_recordable_span(get_current_span(parent_context)): - spans = tuple( + spans = tuple( # rebind-ok: replaces the empty default declared above self._emitter.start_span( SpanRole.LLM_CALL, call.provisional_span_name, @@ -415,7 +415,7 @@ class OpenTelemetryV2(CustomLogger): # The tool-call span carries ``gen_ai.operation.name``, so the fan-out processor # treats it as a gen-AI span and skips it; route it to the request's admin # destinations like the LLM-call span, or it reaches only the global exporter. - call = LLMCallEvent.from_dict(kwargs) + call: Final = LLMCallEvent.from_dict(kwargs) self._emitter.emit_fanout( SpanRole.MCP_TOOL_CALL, data, @@ -487,11 +487,11 @@ class OpenTelemetryV2(CustomLogger): """ from litellm.integrations.otel.presets import dynamic_otlp_headers - call = LLMCallEvent.from_dict(kwargs) - call_id = call.call_id + call: Final = LLMCallEvent.from_dict(kwargs) + call_id: Final = call.call_id - carrier = self._open_llm_calls.pop(call_id, None) if call_id else None - payload = call.payload + carrier: Final = self._open_llm_calls.pop(call_id, None) if call_id else None + payload: Final = call.payload # The closed marker guards the carrier-less path only, where it stops a repeat # success/failure callback re-emitting a span that already shipped. An open @@ -503,13 +503,13 @@ class OpenTelemetryV2(CustomLogger): # its own payload id (the provider's response id, falling back to the call id). # Keying on the call id made the successful attempt after a failure look like a # duplicate, so a destination saw only the failure. - emit_key = (payload.get("id") if payload else None) or call_id + emit_key: Final = (payload.get("id") if payload else None) or call_id if carrier is None and emit_key and emit_key in self._closed_call_ids: return None if carrier is None: - destinations = self._destinations_for_backend(call) - own_credentials = bool(dynamic_otlp_headers(self.callback_name, call.dynamic_params)) + destinations: Final = self._destinations_for_backend(call) + own_credentials: Final = bool(dynamic_otlp_headers(self.callback_name, call.dynamic_params)) if call.is_no_upstream_call or payload is None or not (destinations or own_credentials): return None self._mark_closed(emit_key) @@ -522,14 +522,14 @@ class OpenTelemetryV2(CustomLogger): call.dynamic_params, ) - end_time_ns = to_ns(end_time) + end_time_ns: Final = to_ns(end_time) self._mark_closed(emit_key) if payload is None: for span in carrier.spans: span.end(end_time=end_time_ns) return None - data = LLMCallSpanData.from_standard_logging_payload( + data: Final = LLMCallSpanData.from_standard_logging_payload( payload, capture_content=self.config.capture_span_content, time_to_first_chunk_seconds=call.time_to_first_chunk_seconds, @@ -569,7 +569,7 @@ class OpenTelemetryV2(CustomLogger): Two callers: the SDK thread-pool path and the destination-resolver path. Both anchor to the request's root span via the worker-copied context and seed identity Baggage. """ - data = LLMCallSpanData.from_standard_logging_payload( + data: Final = LLMCallSpanData.from_standard_logging_payload( payload, capture_content=self.config.capture_span_content, time_to_first_chunk_seconds=time_to_first_chunk_seconds, @@ -878,7 +878,7 @@ def publish_global_otel_v2_provider( register_admin_destination_logger, ) - logger = select_global_otel_v2_logger(in_memory_loggers, registered=registered) + logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered) set_global_provider(logger._tracer_provider) register_admin_destination_logger() return logger diff --git a/litellm/integrations/otel/model/config.py b/litellm/integrations/otel/model/config.py index 0b51f8e97e8..c991dda2d90 100644 --- a/litellm/integrations/otel/model/config.py +++ b/litellm/integrations/otel/model/config.py @@ -71,7 +71,7 @@ class _OTelV2Flag(BaseSettings): """ if not isinstance(value, str): return value - stripped = value.strip() + stripped: Final = value.strip() if not stripped: return False try: @@ -289,7 +289,7 @@ class OpenTelemetryV2Config(BaseSettings): # span (including prompt and completion content) to stdout synchronously on the # request path. An explicitly chosen exporter (even ``console``), a non-console # kind, or an endpoint all still fold. - console_by_default = self.exporter == "console" and "exporter" not in self.model_fields_set + console_by_default: Final = self.exporter == "console" and "exporter" not in self.model_fields_set if not self.exporters and (self.endpoint or not console_by_default): self.exporters = [ ExporterSpec( diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 48ddcf9518f..ae85359a61f 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -1,6 +1,7 @@ """Provider / exporter factory + the Baggage span processor.""" from collections.abc import Callable, Iterable, Mapping +from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final from opentelemetry import _logs, baggage, metrics @@ -112,7 +113,7 @@ def _otlp_traces_endpoint(endpoint: str | None) -> str | None: return endpoint endpoint = endpoint.rstrip("/") # Splunk Observability uses ``/v2/trace/otlp``; never rewrite it. - if endpoint.endswith("/v1/traces") or "/v2/trace/otlp" in endpoint or endpoint.endswith("/api/trace"): + if endpoint.endswith(("/v1/traces", "/api/trace")) or "/v2/trace/otlp" in endpoint: return endpoint for other_signal in ("/v1/logs", "/v1/metrics"): if endpoint.endswith(other_signal): @@ -120,7 +121,7 @@ def _otlp_traces_endpoint(endpoint: str | None) -> str | None: return endpoint + "/v1/traces" -_GRPC_BACKENDS = frozenset({"arize"}) +_GRPC_BACKENDS: Final = frozenset({"arize"}) def default_otlp_kind_for_backend(callback_name: "str | None") -> str: @@ -133,7 +134,7 @@ def destination_resource_attrs(destination: "OtelDestination") -> Mapping[str, s ``model_id`` / ``arize.project.name``; empty for header-routed backends), read by both export paths so the gen-AI span and its parents share one Resource. """ - return dict(destination.resource_attributes) + return MappingProxyType(dict(destination.resource_attributes)) def parse_headers(raw: str | None) -> dict[str, str]: diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 2101e532de4..5720e5f7053 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -67,7 +67,7 @@ class TenantTracerCache: self._callback_name = callback_name self._tracer_name = tracer_name self._providers: OrderedDict[tuple[object, ...], TracerProvider] = ( - OrderedDict() + OrderedDict() # mutable-ok: LRU cache, mutated by design ) # mutable-ok: bounded LRU tracer-provider cache def _evict_if_full(self) -> None: @@ -96,7 +96,7 @@ class TenantTracerCache: """ if not destinations: return (default,) - groups = tuple( + groups: Final = tuple( self._tracer_for_group(resource_key, group, include_base=False) for resource_key, group in self._group_by_resource(destinations) ) @@ -115,7 +115,7 @@ class TenantTracerCache: headers: Final = dynamic_otlp_headers(self._callback_name, dynamic_params) if not headers: return self.tracers_for(default, destinations) - dynamic = self._credential_scoped_tracer(headers, dynamic_params) + dynamic: Final = self._credential_scoped_tracer(headers, dynamic_params) if not destinations: return (dynamic,) return (dynamic, *self.tracers_for(default, destinations, include_base_on_first=False)) @@ -135,18 +135,20 @@ class TenantTracerCache: """ from litellm.integrations.otel.presets import dynamic_otlp_destination - destination = dynamic_otlp_destination(self._callback_name, dynamic_params) - cache_key: tuple[object, ...] = ( + destination: Final = dynamic_otlp_destination(self._callback_name, dynamic_params) + cache_key: Final[tuple[object, ...]] = ( "dynamic", tuple(sorted(headers.items())), destination.endpoint if destination is not None else None, destination.protocol if destination is not None else None, ) - provider = self._providers.get(cache_key) + provider = self._providers.get(cache_key) # rebind-ok: replaced below on a cache miss if provider is not None: self._providers.move_to_end(cache_key) else: - provider = build_tracer_provider(self._config_with_headers(headers, dynamic_params)) + provider = build_tracer_provider( # rebind-ok: reassigned on the branch below + self._config_with_headers(headers, dynamic_params) + ) # rebind-ok: cache miss self._providers[cache_key] = provider self._evict_if_full() return get_tracer(provider, self._tracer_name) @@ -171,18 +173,25 @@ class TenantTracerCache: same builder an equivalent admin destination would use so the team reaches the account its ``callback_vars`` name. """ - header_str = ",".join(f"{key}={value}" for key, value in headers.items()) - owns_exporter = any( + header_str: Final = ",".join(f"{key}={value}" for key, value in headers.items()) + owns_exporter: Final = any( spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS for spec in self._config.exporters ) if not owns_exporter: return self._config.model_copy( - update={"exporters": [*self._config.exporters, *self._synthesized_exporter(header_str, dynamic_params)]} + update={ # mutable-ok: handed to a model or SDK that needs a concrete dict + "exporters": [ + *self._config.exporters, + *self._synthesized_exporter(header_str, dynamic_params), + ] # mutable-ok: handed to a model or SDK that needs a concrete dict + } # mutable-ok: pydantic model_copy takes a plain update mapping ) - exporters = [ + exporters: Final = [ ( - spec.model_copy(update={"headers": header_str}) + spec.model_copy( + update={"headers": header_str} # mutable-ok: handed to a model or SDK that needs a concrete dict + ) # mutable-ok: pydantic model_copy takes a plain update mapping if spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS else spec ) @@ -204,7 +213,7 @@ class TenantTracerCache: """ from litellm.integrations.otel.presets import dynamic_otlp_destination - destination = dynamic_otlp_destination(self._callback_name, dynamic_params) + destination: Final = dynamic_otlp_destination(self._callback_name, dynamic_params) if destination is None or not destination.endpoint: return () return ( @@ -228,12 +237,12 @@ class TenantTracerCache: destination_resource_attrs, ) - groups: OrderedDict[tuple[tuple[str, str], ...], list[OtelDestination]] = ( - OrderedDict() - ) # mutable-ok: insertion-order grouping accumulator, frozen before return + groups: Final[ # mutable-ok: insertion-order grouping accumulator, frozen before return + OrderedDict[tuple[tuple[str, str], ...], list[OtelDestination]] + ] = OrderedDict() for destination in destinations: key = tuple(sorted(destination_resource_attrs(destination).items())) - groups.setdefault(key, []).append(destination) + groups.setdefault(key, []).append(destination) # mutable-ok: grouping accumulator, frozen before return return tuple((key, tuple(group)) for key, group in sorted(groups.items())) def _tracer_for_group( @@ -243,16 +252,16 @@ class TenantTracerCache: *, include_base: bool, ) -> Tracer: - cache_key: tuple[object, ...] = ( + cache_key: Final[tuple[object, ...]] = ( resource_key, tuple(sorted((d.endpoint, tuple(sorted(d.headers.items())), d.protocol or "") for d in group)), include_base, ) - provider = self._providers.get(cache_key) + provider = self._providers.get(cache_key) # rebind-ok: replaced below on a cache miss if provider is not None: self._providers.move_to_end(cache_key) else: - provider = build_tracer_provider( + provider = build_tracer_provider( # rebind-ok: cache miss self._config_with_destinations(tuple(group), include_base_exporters=include_base) ) self._providers[cache_key] = provider @@ -289,8 +298,8 @@ class TenantTracerCache: destination_resource_attrs, ) - kind = self._owned_otlp_kind() - appended = tuple( + kind: Final = self._owned_otlp_kind() + appended: Final = tuple( ExporterSpec( kind=d.protocol or kind, endpoint=d.endpoint, @@ -299,22 +308,27 @@ class TenantTracerCache: ) for d in destinations ) - base_exporters = (*self._config.exporters,) if include_base_exporters else () - merged_resource_attrs = { + base_exporters: Final = (*self._config.exporters,) if include_base_exporters else () + merged_resource_attrs: Final = { # mutable-ok: handed to a model or SDK that needs a concrete dict **self._config.resource_attributes, - **{key: value for d in destinations for key, value in destination_resource_attrs(d).items()}, + **{ # mutable-ok: handed to a model or SDK that needs a concrete dict + key: value for d in destinations for key, value in destination_resource_attrs(d).items() + }, # mutable-ok: construction is handed to an API that needs a concrete dict } return self._config.model_copy( - update={ - "exporters": [*base_exporters, *appended], + update={ # mutable-ok: pydantic model_copy takes a plain update mapping + "exporters": [ # mutable-ok: handed to a model or SDK that needs a concrete dict + *base_exporters, + *appended, + ], # mutable-ok: pydantic model_copy takes a plain update mapping "resource_attributes": merged_resource_attrs, } ) -_MAX_CACHED_PROCESSORS = 256 +_MAX_CACHED_PROCESSORS: Final = 256 -_GENAI_SPAN_ATTR = "gen_ai.operation.name" +_GENAI_SPAN_ATTR: Final = "gen_ai.operation.name" def _processor_key(destination: OtelDestination) -> "tuple[str, tuple[tuple[str, str], ...], str | None]": @@ -322,7 +336,7 @@ def _processor_key(destination: OtelDestination) -> "tuple[str, tuple[tuple[str, def _is_genai_span(span: ReadableSpan) -> bool: - attributes = span.attributes or {} + attributes: Final = span.attributes or {} # mutable-ok: read-only fallback for a span with no attributes return _GENAI_SPAN_ATTR in attributes @@ -333,10 +347,12 @@ def _with_destination_resource(span: ReadableSpan, destination: OtelDestination) destination_resource_attrs, ) - extra = destination_resource_attrs(destination) + extra: Final = destination_resource_attrs(destination) if not extra: return span - merged = Resource.create({**dict(span.resource.attributes), **extra}) + merged: Final = Resource.create( + {**dict(span.resource.attributes), **extra} # mutable-ok: handed to a model or SDK that needs a concrete dict + ) # mutable-ok: the OTel SDK takes a concrete attribute mapping return _ResourceWrappedReadableSpan(span, merged) @@ -371,14 +387,14 @@ class TenantFanOutSpanProcessor(SpanProcessor): def __init__(self, owner_callback_name: str | None) -> None: self._owner = owner_callback_name self._processors: OrderedDict[tuple, SpanProcessor] = ( - OrderedDict() + OrderedDict() # mutable-ok: LRU cache, mutated by design ) # mutable-ok: bounded LRU span-processor cache def on_start(self, span: Span, parent_context: Context | None = None) -> None: return None def on_end(self, span: ReadableSpan) -> None: - destinations = request_destinations() + destinations: Final = request_destinations() if not destinations: return if _is_genai_span(span): @@ -409,7 +425,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): self._processors.clear() def force_flush(self, timeout_millis: int = 30000) -> bool: - all_ok = True + all_ok = True # rebind-ok: cleared by any failing exporter below # Snapshot before iterating (see ``shutdown``): a concurrent ``on_end`` mutating # the processor cache must not abort the flush and drop the remaining destinations' # buffered spans. @@ -422,28 +438,28 @@ class TenantFanOutSpanProcessor(SpanProcessor): return all_ok def _processor_for(self, destination: OtelDestination) -> SpanProcessor | None: - key = _processor_key(destination) - cached = self._processors.get(key) + key: Final = _processor_key(destination) + cached: Final = self._processors.get(key) if cached is not None: self._processors.move_to_end(key) return cached from litellm.integrations.otel.plumbing.providers import ( - _exporter_from_spec, + _exporter_from_spec, # pyright: ignore[reportPrivateUsage] # shared with providers.py inside the otel plumbing package default_otlp_kind_for_backend, ) from litellm.integrations.otel.plumbing.providers import ( - _processor_for as _build_processor, + _processor_for as _build_processor, # pyright: ignore[reportPrivateUsage] # shared with providers.py inside the otel plumbing package ) try: - spec = ExporterSpec( + spec: Final = ExporterSpec( kind=destination.protocol or default_otlp_kind_for_backend(destination.callback_name), endpoint=destination.endpoint, headers=destination.header_string(), owner=None, ) - exporter = _exporter_from_spec(spec) - processor = _build_processor(exporter, use_simple=False) + exporter: Final = _exporter_from_spec(spec) + processor: Final = _build_processor(exporter, use_simple=False) except Exception as exc: # noqa: BLE001 # a malformed destination spec must not break fan-out; skip this destination verbose_logger.debug( "OTel V2 fan-out: failed to build processor for %s: %s", diff --git a/litellm/integrations/otel/presets/__init__.py b/litellm/integrations/otel/presets/__init__.py index 0296de458cc..2dbe3b8e2de 100644 --- a/litellm/integrations/otel/presets/__init__.py +++ b/litellm/integrations/otel/presets/__init__.py @@ -16,7 +16,7 @@ those per-request headers. """ from collections.abc import Callable -from typing import Final, TYPE_CHECKING +from typing import TYPE_CHECKING, Final from litellm.integrations.otel.presets.agentops import agentops_preset from litellm.integrations.otel.presets.arize import arize_dynamic_headers, arize_preset @@ -39,7 +39,7 @@ if TYPE_CHECKING: #: routing). Only integrations that support dynamic credentials appear here — #: Arize-Phoenix/Langtrace/Levo/AgentOps/generic don't, so they use the logger's #: default tracer. -DYNAMIC_HEADERS_BY_CALLBACK: dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]] = { +DYNAMIC_HEADERS_BY_CALLBACK: Final[dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]]] = { "arize": arize_dynamic_headers, "langfuse_otel": langfuse_dynamic_headers, "weave_otel": weave_dynamic_headers, @@ -76,13 +76,15 @@ def dynamic_otlp_destination( if callback_name not in DYNAMIC_HEADERS_BY_CALLBACK or not dynamic_params: return None - values = {str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str)} + values: Final = { # mutable-ok: handed to a model or SDK that needs a concrete dict + str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str) + } # mutable-ok: handed to a model or SDK that needs a concrete dict return build_destination(callback_name or "", values) #: Callback name → preset. The ``Preset`` annotation makes mypy verify every #: registered value matches the preset interface. -PRESET_BY_CALLBACK: dict[str, Preset] = { +PRESET_BY_CALLBACK: Final[dict[str, Preset]] = { # mutable-ok: module registry, read-only after import "agentops": agentops_preset, "arize": arize_preset, "arize_phoenix": phoenix_preset, diff --git a/litellm/integrations/otel/presets/agentops.py b/litellm/integrations/otel/presets/agentops.py index 38f80aab36c..e7755f19be2 100644 --- a/litellm/integrations/otel/presets/agentops.py +++ b/litellm/integrations/otel/presets/agentops.py @@ -50,21 +50,28 @@ def agentops_preset( """ settings: Final = _AgentOpsSettings() base: Final = config_overrides or OpenTelemetryV2Config() - global_exporter = ( + global_exporter: Final = ( () if allow_missing_credentials and not settings.api_key else ( ExporterSpec( kind=_AGENTOPS_EXPORTER_KIND, endpoint=_AGENTOPS_ENDPOINT, - options=({"api_key": settings.api_key} if settings.api_key else None), + options=( + {"api_key": settings.api_key} + if settings.api_key + else None # mutable-ok: handed to a model or SDK that needs a concrete dict + ), # mutable-ok: the exporter spec carries a concrete options mapping owner=ExporterOwner.AGENTOPS, ), ) ) return base.model_copy( update={ - "exporters": [*base.exporters, *global_exporter], + "exporters": [ + *base.exporters, + *global_exporter, + ], # mutable-ok: pydantic model_copy takes a plain update mapping "resource_attributes": { **base.resource_attributes, "service.name": settings.service_name, diff --git a/litellm/integrations/otel/presets/arize.py b/litellm/integrations/otel/presets/arize.py index fb6a49cb101..8fc9bc48b60 100644 --- a/litellm/integrations/otel/presets/arize.py +++ b/litellm/integrations/otel/presets/arize.py @@ -14,7 +14,7 @@ from litellm.integrations.otel.model.config import ( from litellm.integrations.otel.presets.utils import ensure_mappers from litellm.types.utils import StandardCallbackDynamicParams -ARIZE_PUBLIC_OTLP_ENDPOINT = "https://otlp.arize.com/v1" +ARIZE_PUBLIC_OTLP_ENDPOINT: Final = "https://otlp.arize.com/v1" class _ArizeSettings(BaseSettings): @@ -33,11 +33,11 @@ def arize_preset( allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: arize_cfg: Final = _V1ArizeLogger.get_arize_config() - settings = _ArizeSettings() - has_own_endpoint = bool(settings.grpc_endpoint or settings.http_endpoint) - has_credentials = bool(arize_cfg.space_id or arize_cfg.space_key or arize_cfg.api_key) + settings: Final = _ArizeSettings() + has_own_endpoint: Final = bool(settings.grpc_endpoint or settings.http_endpoint) + has_credentials: Final = bool(arize_cfg.space_id or arize_cfg.space_key or arize_cfg.api_key) base: Final = config_overrides or OpenTelemetryV2Config() - global_exporter = ( + global_exporter: Final = ( () if allow_missing_credentials and not has_credentials and not has_own_endpoint else ( @@ -51,7 +51,10 @@ def arize_preset( ) return base.model_copy( update={ - "exporters": [*base.exporters, *global_exporter], + "exporters": [ + *base.exporters, + *global_exporter, + ], # mutable-ok: pydantic model_copy takes a plain update mapping "mapper_names": ensure_mappers(base.mapper_names, "openinference"), "resource_attributes": { **base.resource_attributes, @@ -77,9 +80,11 @@ def _arize_headers(arize_cfg, settings: "_ArizeSettings", has_own_endpoint: bool def arize_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]: """Per-request Arize OTLP headers from team/key dynamic params.""" - headers: dict[str, str] = {} + headers: dict[ # rebind-ok: reassigned on the branch below + str, str + ] = {} # rebind-ok: populated by the optional-credential branches below # mutable-ok: construction is handed to an API that needs a concrete dict # ``arize_space_key`` is the suggested param and wins over ``arize_space_id``. - space = params.get("arize_space_key") or params.get("arize_space_id") + space: Final = params.get("arize_space_key") or params.get("arize_space_id") if space: headers["arize-space-id"] = space api_key: Final = params.get("arize_api_key") diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index 8ff8145d23e..84a1c474468 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -20,7 +20,9 @@ from litellm.integrations.langfuse.langfuse_otel import ( LangfuseOtelLogger, ) from litellm.integrations.otel.model.destination import OtelDestination -from litellm.integrations.weave.weave_otel import _get_weave_authorization_header +from litellm.integrations.weave.weave_otel import ( + _get_weave_authorization_header, # pyright: ignore[reportPrivateUsage] # reuse the backend's own header builder rather than duplicating its auth scheme +) def _parse_header_string(raw: str) -> Mapping[str, str]: @@ -40,7 +42,7 @@ def _langfuse_destination(values: Mapping[str, str]) -> OtelDestination | None: return None host: Final = values.get("langfuse_host") endpoint: Final = _langfuse_endpoint(host) if host else LANGFUSE_CLOUD_US_ENDPOINT - auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key) + auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key) # pyright: ignore[reportPrivateUsage] # reuse the backend's own header builder rather than duplicating its auth scheme return OtelDestination(endpoint=endpoint, headers=MappingProxyType({"Authorization": auth})) diff --git a/litellm/integrations/otel/presets/langfuse.py b/litellm/integrations/otel/presets/langfuse.py index 3118aa12a53..6a7198ce108 100644 --- a/litellm/integrations/otel/presets/langfuse.py +++ b/litellm/integrations/otel/presets/langfuse.py @@ -19,12 +19,12 @@ def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, public_key: Final = params.get("langfuse_public_key") secret_key: Final = params.get("langfuse_secret_key") if public_key and secret_key: - return { + return { # mutable-ok: handed to a model or SDK that needs a concrete dict "Authorization": _V1Langfuse._get_langfuse_authorization_header( public_key=public_key, secret_key=secret_key ) } - return {} + return {} # mutable-ok: handed to a model or SDK that needs a concrete dict def langfuse_preset( @@ -33,13 +33,15 @@ def langfuse_preset( allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: base: Final = config_overrides or OpenTelemetryV2Config() - mappers = ensure_mappers(base.mapper_names, "langfuse") + mappers: Final = ensure_mappers(base.mapper_names, "langfuse") try: - cfg = _V1Langfuse.get_langfuse_otel_config() + cfg: Final = _V1Langfuse.get_langfuse_otel_config() except Exception: if not allow_missing_credentials: raise - return base.model_copy(update={"mapper_names": mappers}) + return base.model_copy( + update={"mapper_names": mappers} # mutable-ok: handed to a model or SDK that needs a concrete dict + ) # mutable-ok: pydantic model_copy takes a plain update mapping kind: Final = cfg.exporter if isinstance(cfg.exporter, str) else "otlp_http" return base.model_copy( update={ diff --git a/litellm/integrations/otel/presets/levo.py b/litellm/integrations/otel/presets/levo.py index 1d118e5223e..1eb2759b82c 100644 --- a/litellm/integrations/otel/presets/levo.py +++ b/litellm/integrations/otel/presets/levo.py @@ -17,7 +17,7 @@ def levo_preset( ) -> OpenTelemetryV2Config: base: Final = config_overrides or OpenTelemetryV2Config() try: - cfg = _V1Levo.get_levo_config() + cfg: Final = _V1Levo.get_levo_config() except Exception: if not allow_missing_credentials: raise diff --git a/litellm/integrations/otel/presets/phoenix.py b/litellm/integrations/otel/presets/phoenix.py index 18af6a8f62b..aca09bbb4f6 100644 --- a/litellm/integrations/otel/presets/phoenix.py +++ b/litellm/integrations/otel/presets/phoenix.py @@ -34,7 +34,7 @@ def phoenix_preset( headers: Final = cfg.otlp_auth_headers if hasattr(cfg, "otlp_auth_headers") else None project_name: Final = _PhoenixSettings().project_name base: Final = config_overrides or OpenTelemetryV2Config() - global_exporter = ( + global_exporter: Final = ( ExporterSpec( kind=cfg.protocol if hasattr(cfg, "protocol") else "otlp_http", endpoint=cfg.endpoint, @@ -44,7 +44,10 @@ def phoenix_preset( ) return base.model_copy( update={ - "exporters": [*base.exporters, *global_exporter], + "exporters": [ + *base.exporters, + *global_exporter, + ], # mutable-ok: pydantic model_copy takes a plain update mapping "mapper_names": ensure_mappers(base.mapper_names, "openinference"), "resource_attributes": { **base.resource_attributes, diff --git a/litellm/integrations/otel/presets/weave.py b/litellm/integrations/otel/presets/weave.py index 2a46ba50c24..f28f6b9824e 100644 --- a/litellm/integrations/otel/presets/weave.py +++ b/litellm/integrations/otel/presets/weave.py @@ -17,7 +17,9 @@ from litellm.types.utils import StandardCallbackDynamicParams def weave_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]: """Per-request Weave OTLP headers from team/key dynamic params.""" - headers: dict[str, str] = {} + headers: dict[ # rebind-ok: reassigned on the branch below + str, str + ] = {} # rebind-ok: populated by the optional-credential branches below # mutable-ok: construction is handed to an API that needs a concrete dict api_key: Final = params.get("wandb_api_key") if api_key: headers["Authorization"] = _get_weave_authorization_header(api_key=api_key) @@ -33,13 +35,15 @@ def weave_preset( allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: base: Final = config_overrides or OpenTelemetryV2Config() - mappers = ensure_mappers(base.mapper_names, "openinference", "weave") + mappers: Final = ensure_mappers(base.mapper_names, "openinference", "weave") try: - weave_cfg = get_weave_otel_config() + weave_cfg: Final = get_weave_otel_config() except Exception: if not allow_missing_credentials: raise - return base.model_copy(update={"mapper_names": mappers}) + return base.model_copy( + update={"mapper_names": mappers} # mutable-ok: handed to a model or SDK that needs a concrete dict + ) # mutable-ok: pydantic model_copy takes a plain update mapping return base.model_copy( update={ "exporters": [ diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 140c02025fe..d4dd2c09fb1 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -4025,11 +4025,16 @@ def _init_custom_logger_compatible_class( for callback in _in_memory_loggers: if type(callback) is OpenTelemetryV2: return callback - settings = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration) - config = OpenTelemetryV2Config(**settings) + settings: Final = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration) + config = OpenTelemetryV2Config(**settings) # rebind-ok: refined below when no exporter resolved if not config.exporters: - config = OpenTelemetryV2Config(**{**settings, "exporter": config.exporter}) - otel_logger_v2 = OpenTelemetryV2(config=config) + config = OpenTelemetryV2Config( # rebind-ok: reassigned on the branch below + **{ + **settings, + "exporter": config.exporter, + } # mutable-ok: handed to a model or SDK that needs a concrete dict + ) # mutable-ok: handed to a model or SDK that needs a concrete dict + otel_logger_v2: Final = OpenTelemetryV2(config=config) _in_memory_loggers.append(otel_logger_v2) _maybe_auto_initialize_arize_phoenix(_in_memory_loggers) return otel_logger_v2 @@ -4393,7 +4398,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name: return callback try: - config = preset_fn(allow_missing_credentials=False) + config: Final = preset_fn(allow_missing_credentials=False) except Exception: return None if not config.exporters: @@ -4407,7 +4412,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom "register a logging destination for it.", callback_name, ) - v2_logger = OpenTelemetryV2(config=config, callback_name=callback_name) + v2_logger: Final = OpenTelemetryV2(config=config, callback_name=callback_name) _in_memory_loggers.append(v2_logger) return v2_logger @@ -4430,7 +4435,7 @@ def otel_v2_owned_backends() -> frozenset[str]: """ from litellm.integrations.otel.logger import OpenTelemetryV2 - dispatched = litellm._async_success_callback + dispatched: Final = litellm._async_success_callback return frozenset( name for logger in _in_memory_loggers diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index ee0ad5b614d..c32e06d3ea5 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -4790,7 +4790,9 @@ if MCP_AVAILABLE: return stored - async def _refresh_request_otel_destinations(user_api_key_auth: Any) -> None: + async def _refresh_request_otel_destinations( + user_api_key_auth: object, + ) -> None: """Re-resolve the caller's admin destinations for THIS JSON-RPC message. A stateful MCP session dispatches every later message on descendants of the task @@ -4803,7 +4805,9 @@ if MCP_AVAILABLE: if user_api_key_auth is None: return try: - from litellm.proxy.litellm_pre_call_utils import _apply_admin_logging_exporters + from litellm.proxy.litellm_pre_call_utils import ( + _apply_admin_logging_exporters, # pyright: ignore[reportPrivateUsage] # the MCP path applies the same pre-call resolution as the HTTP path + ) await _apply_admin_logging_exporters(user_api_key_auth) except Exception as exc: # noqa: BLE001 # a resolver failure must not break the MCP call diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 9c35902689b..c5ce64483cf 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -1035,7 +1035,7 @@ async def _hoist_request_destinations(request: Request, user_api_key_dict: UserA set_request_destinations, ) from litellm.proxy.litellm_pre_call_utils import ( - _resolve_logging_exporters, + _resolve_logging_exporters, # pyright: ignore[reportPrivateUsage] # resolver lives in the pre-call module it is hoisted from ) destinations_raw, _backends = await _resolve_logging_exporters(user_api_key_dict) diff --git a/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py b/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py index cc66bc57c6b..9312b63ac59 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py +++ b/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py @@ -329,3 +329,35 @@ async def test_disclosure_agrees_with_the_resolver_on_a_shared_target(monkeypatc assert len(destinations) == 1 assert resolved_logging_exporter_names(None, None) == ("dup-one", "dup-two") + + + +def test_backend_without_a_preset_routes_under_generic(): + """Regression: a backend outside ``PRESET_BY_CALLBACK`` is routed as ``generic``. + + Only a registered name gets an ``OpenTelemetryV2`` logger, and that logger is what + emits the gen-AI span to its destinations. Routing an unregistered name under itself + delivered the proxy-internal spans but never the LLM call. Registered names keep + their own routing so each backend's attribute vocabulary is preserved.""" + from litellm.integrations.otel.presets import PRESET_BY_CALLBACK + + unknown = CredentialItem( + credential_name="self-hosted", + credential_values={"otel_endpoint": "http://collector.example/v1/traces"}, + credential_info={"credential_type": "logging", "description": "honeycomb", "access": {"global": True}}, + ) + resolved = destination_for_credential(unknown) + assert resolved is not None + assert resolved[0] == "generic" + assert resolved[1].endpoint == "http://collector.example/v1/traces" + + for registered in ("arize", "langfuse_otel", "generic"): + assert registered in PRESET_BY_CALLBACK + known = CredentialItem( + credential_name="lf", + credential_values={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, + credential_info={"credential_type": "logging", "description": "langfuse_otel", "access": {"global": True}}, + ) + known_resolved = destination_for_credential(known) + assert known_resolved is not None + assert known_resolved[0] == "langfuse_otel" diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py index f8664f06853..62694f1a318 100644 --- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py +++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py @@ -6917,3 +6917,25 @@ async def test_apply_admin_logging_exporters_degrades_when_flag_on_without_opent env={**os.environ, "PYTHONPATH": repo_root, "LITELLM_OTEL_V2": "true"}, ) assert "REQUEST_PATH_OK" in result.stdout, f"flag-on request path raised:\n{result.stderr[-3000:]}" + + + +@pytest.mark.asyncio +async def test_apply_admin_logging_exporters_registers_on_failure(_seeded_logging_credentials, monkeypatch): + """An admin-owned destination must capture a FAILED upstream call, not only a + successful one. + + The destination sink is one process-wide logger, so it has to sit on the failure + list as well as the success list; registering it on success alone means a + 401/timeout never reaches the destination and the trace lands with no error + gen-AI span. Registration is idempotent. + """ + import litellm + from litellm.integrations.otel.destination_logger import admin_destination_logger + from litellm.integrations.otel.logger import publish_global_otel_v2_provider + + sink = admin_destination_logger() + publish_global_otel_v2_provider([], lambda provider: None) + publish_global_otel_v2_provider([], lambda provider: None) + for bucket in (litellm._async_success_callback, litellm._async_failure_callback): + assert sum(1 for callback in bucket if callback is sink) == 1