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