From 5e74515c29d62982694fa50aab5b416c2c252949 Mon Sep 17 00:00:00 2001 From: mrinal-berri Date: Wed, 11 Nov 2026 13:57:41 -0800 Subject: [PATCH] feat(otel): let a team or key OTel v2 destination add its own request metadata keys (#50813) * feat(otel): let a team or key OTel v2 destination add its own request metadata keys A team or key callback can now set baggage_metadata_keys in its callback_vars, a comma-separated list of requester_metadata. paths. The tenant fan-out adds them as litellm.metadata. to that destination's spans only, on top of the global LITELLM_OTEL_BAGGAGE_METADATA_KEYS; the operator's exporter and other destinations never see them. Values are bound when each span ends, an attribute the span already carries wins, and names of proxy identity fields are rejected when saved, along with lists over 20 keys. Resolves LIT-9545 * fix(otel): refuse request headers in a destination's baggage_metadata_keys The proxy copies the request headers into requester_metadata.headers with credential headers masked, but a header a gateway in front of it added (an IAP or OIDC assertion, an internal token) stays as sent. A team or key admin listing requester_metadata.headers. could export it to an endpoint the team controls. Entries under requester_metadata.headers are now rejected when saved, so exporting a header stays an operator decision in LITELLM_OTEL_BAGGAGE_METADATA_KEYS. Also drops an unused import. * fix(otel): hold destination metadata to the configured span limits A destination's extra litellm.metadata.* attributes were added to a span view built from a plain mapping, so they skipped the limits the SDK applies in set_attribute: a 1 MiB caller value survived an operator's OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT. Values are now cut to the length limit, and keys past the attribute-count limit are dropped and counted instead of evicting an attribute the span already carries. The per-destination views are built in one Final tuple ahead of the forwarding loop, since a Final cannot be assigned inside the loop. * fix(otel): apply destination metadata after the destination view, within the provider's limits The span limits were checked against the span as it ended, before the destination view strips prompt content and backend-only attributes, so a model-call span full of llm.input_messages.* could drop the team's keys on a no_content destination that had room. The values are now bound when the span ends and added as the last step before export, counted against what that destination actually sends, so their dropped count also survives the view. The limits come from the tracer provider the fan-out is attached to, read from the live span at start, instead of a separate env-built SpanLimits. * refactor(otel): consolidate span limits, destination validators and metadata key parsing Reads a span's limits in one guarded place that the emitter and the fan-out share, replaces the two parallel lists of per-destination callback-var validators with one mapping, splits and strips a baggage_metadata_keys list once, drops a filter the post-view merge already enforces, and folds the repeated fan-out test setup into one helper. No behavior change. * ci: retrigger checks stranded by the GitHub Actions outage --- litellm/integrations/otel/README.md | 8 +- litellm/integrations/otel/emitter.py | 9 +- litellm/integrations/otel/logger.py | 2 + litellm/integrations/otel/model/baggage.py | 58 +++- .../integrations/otel/model/destination.py | 11 +- litellm/integrations/otel/plumbing/context.py | 19 ++ .../integrations/otel/plumbing/providers.py | 93 ++++++- .../integrations/otel/presets/destinations.py | 2 + .../initialize_dynamic_callback_params.py | 23 +- litellm/proxy/_types.py | 18 +- .../callback_config_validation.py | 34 ++- litellm/proxy/litellm_pre_call_utils.py | 6 +- .../team_callback_endpoints.py | 5 +- litellm/types/utils.py | 7 + .../test_otel_destination_metadata_keys.py | 262 ++++++++++++++++++ .../otel/test_otel_v2_destinations.py | 127 +++++++++ .../test_callback_config_validation.py | 71 ++++- ui/litellm-dashboard/src/lib/http/schema.d.ts | 1 + 18 files changed, 700 insertions(+), 56 deletions(-) create mode 100644 tests/integration/observability/test_otel_destination_metadata_keys.py diff --git a/litellm/integrations/otel/README.md b/litellm/integrations/otel/README.md index 06c2da117b5..70384e49f53 100644 --- a/litellm/integrations/otel/README.md +++ b/litellm/integrations/otel/README.md @@ -325,7 +325,13 @@ nothing here imports outside it: on `langfuse_otel`, `arize`, `weave_otel` and `newrelic` callbacks. - [`baggage.py`](./model/baggage.py) — the single definition of which request-identity values are promoted into Baggage (so child spans inherit them) and under which - attribute keys. + attribute keys. A team or key destination can list extra `requester_metadata.` + paths in `baggage_metadata_keys` in its own `callback_vars`. They are never written + into Baggage: the tenant fan-out adds them as `litellm.metadata.` to that + destination's view of each span, on top of the global list, so the operator's + exporter and other destinations never see them. A key the span already carries + wins, and names of proxy identity fields and `requester_metadata.headers.*` (the + request headers the proxy copies there) are rejected when saved. - [`utils.py`](./model/utils.py) — value coercion, JSON serialization, and extractor-table application, shared across the package. diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index 87a432ee7ce..813ba6aaa73 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -6,8 +6,7 @@ from types import MappingProxyType from typing import Final from opentelemetry.context import Context -from opentelemetry.sdk.trace import ReadableSpan, SpanLimits -from opentelemetry.sdk.trace import Span as SdkSpan +from opentelemetry.sdk.trace import ReadableSpan from opentelemetry.trace import Link, Span, Tracer from opentelemetry.trace.status import Status, StatusCode @@ -36,7 +35,7 @@ from litellm.integrations.otel.model.spans import ( ) from litellm.integrations.otel.plumbing.context import destination_backends from litellm.integrations.otel.plumbing.events import GenAIEventRecorder -from litellm.integrations.otel.plumbing.providers import to_otel_span_kind +from litellm.integrations.otel.plumbing.providers import span_limits, to_otel_span_kind # Roles emit() knows how to name and emit. PROXY_REQUEST and the management # routes are SERVER spans owned by the mounted FastAPI instrumentor, so they @@ -88,9 +87,7 @@ def error_attributes(error: SpanError) -> Mapping[str, AttrValue]: def span_attribute_limit(span: Span) -> int | None: """The attribute count limit ``span`` was built with, ``None`` when unbounded.""" - if not isinstance(span, SdkSpan): - return SpanLimits().max_span_attributes - return span._limits.max_span_attributes # pyright: ignore[reportPrivateUsage] # SDK has no public getter + return span_limits(span).max_span_attributes def _carried_keys( diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 41811f836c7..dfb03211927 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -66,6 +66,7 @@ from litellm.integrations.otel.plumbing.context import ( resolve_request_span_context, resolve_service_span_context, set_request_baggage, + set_request_metadata, set_request_root_span, ) from litellm.integrations.otel.plumbing.events import GenAIEventRecorder @@ -748,6 +749,7 @@ class OpenTelemetryV2(CustomLogger): """ try: identity: Final = RequestIdentity.from_user_api_key_auth(user_api_key_dict, request_metadata) + set_request_metadata(identity.metadata) bag: Final = promoted_baggage( identity, model, diff --git a/litellm/integrations/otel/model/baggage.py b/litellm/integrations/otel/model/baggage.py index 606fd79f218..5832a56b076 100644 --- a/litellm/integrations/otel/model/baggage.py +++ b/litellm/integrations/otel/model/baggage.py @@ -18,7 +18,11 @@ from collections.abc import Callable, Mapping from types import MappingProxyType from typing import Final -from litellm.integrations.otel.model.metadata import RequestIdentity, allowlisted_metadata +from litellm.integrations.otel.model.metadata import ( + REQUESTER_METADATA_PATH, + RequestIdentity, + allowlisted_metadata, +) from litellm.integrations.otel.model.semconv import GenAI, LiteLLM # Attribute key -> value extractor over (identity, request_model, @@ -65,6 +69,58 @@ DEFAULT_BAGGAGE_METADATA_KEYS: Final[tuple[str, ...]] = ( # explicitly allowlisted via ``config.baggage_team_metadata_keys``. DEFAULT_BAGGAGE_TEAM_METADATA_KEYS: Final[tuple[str, ...]] = () +# A team or key OTel v2 destination can add caller metadata keys on top of the +# global ``baggage_metadata_keys``, for its own destination only. Each key names +# a ``requester_metadata.`` path, so only what the caller sent can be read, +# and a key whose exported name is a proxy-owned identity field is refused. The +# proxy copies the request headers into ``requester_metadata.headers``, credentials +# masked but anything a gateway in front of it added left as sent, so exporting a +# header stays an operator decision made in the global list. +MAX_DESTINATION_METADATA_KEYS: Final = 20 +_RESERVED_METADATA_PREFIX: Final = "user_api_key_" +_REQUEST_HEADERS_KEY: Final = "headers" + + +def destination_metadata_keys_error(value: str) -> str | None: + """Why ``value`` is not a usable destination ``baggage_metadata_keys`` list, or ``None``.""" + keys: Final = _listed_keys(value) + named: Final = tuple((key, key.removeprefix(REQUESTER_METADATA_PATH)) for key in keys) + if any(name == key or not name for key, name in named): + return f"Invalid baggage_metadata_keys {value!r}: use a comma-separated list of {REQUESTER_METADATA_PATH} entries" + reserved: Final = tuple(key for key, name in named if _is_reserved_metadata_name(name)) + if reserved: + return ( + f"Invalid baggage_metadata_keys {value!r}: {', '.join(reserved)} would overwrite a proxy identity attribute" + ) + headers: Final = tuple(key for key, name in named if _is_request_header(name)) + if headers: + return ( + f"Invalid baggage_metadata_keys {value!r}: {', '.join(headers)} would export request headers, " + "which only the proxy-wide LITELLM_OTEL_BAGGAGE_METADATA_KEYS can do" + ) + if len(frozenset(keys)) > MAX_DESTINATION_METADATA_KEYS: + return f"Invalid baggage_metadata_keys {value!r}: at most {MAX_DESTINATION_METADATA_KEYS} keys" + return None + + +def parse_destination_metadata_keys(value: object) -> tuple[str, ...] | None: + """A destination's ``baggage_metadata_keys`` as a key tuple, ``None`` when unset or invalid.""" + if not isinstance(value, str) or destination_metadata_keys_error(value) is not None: + return None + return tuple(dict.fromkeys(_listed_keys(value))) + + +def _listed_keys(value: str) -> tuple[str, ...]: + return tuple(key.strip() for key in value.split(",")) + + +def _is_reserved_metadata_name(name: str) -> bool: + return name in DEFAULT_BAGGAGE_METADATA_KEYS or name.startswith(_RESERVED_METADATA_PREFIX) + + +def _is_request_header(name: str) -> bool: + return name.partition(".")[0].strip().lower() == _REQUEST_HEADERS_KEY + def promoted_baggage( identity: RequestIdentity, diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py index e760ba3600b..6af28f7c2a7 100644 --- a/litellm/integrations/otel/model/destination.py +++ b/litellm/integrations/otel/model/destination.py @@ -61,6 +61,13 @@ class OtelDestination(LiteLLMBaseModel): default=None, description="An explicit set replaces the global database-span exclusions; ``None`` follows the global set.", ) + metadata_keys: tuple[str, ...] = Field( + default=(), + description=( + "``requester_metadata.`` paths exported as ``litellm.metadata.`` on this destination's " + "spans only, on top of the global ``baggage_metadata_keys``." + ), + ) def captures_content(self, default: str) -> bool: setting: Final = self.capture_message_content @@ -79,8 +86,8 @@ class OtelDestination(LiteLLMBaseModel): def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]: """Identity for processor reuse, so one destination means one exporter. - ``span_scope``, the sampling rates, ``capture_message_content`` and - ``excluded_db_systems`` are left out on purpose: they decide which spans reach + ``span_scope``, the sampling rates, ``capture_message_content``, + ``excluded_db_systems`` and ``metadata_keys`` are left out on purpose: they decide which spans reach the processor and what they carry, not how the processor exports them, so two views of the same account share one exporter. """ diff --git a/litellm/integrations/otel/plumbing/context.py b/litellm/integrations/otel/plumbing/context.py index b328b281dde..03d53965360 100644 --- a/litellm/integrations/otel/plumbing/context.py +++ b/litellm/integrations/otel/plumbing/context.py @@ -4,6 +4,7 @@ import os from collections.abc import Generator, Mapping from contextlib import contextmanager from contextvars import ContextVar, Token +from types import MappingProxyType from typing import TYPE_CHECKING, Final from opentelemetry import baggage @@ -478,6 +479,24 @@ def request_destinations() -> 'tuple["OtelDestination", ...]': return _request_destinations.get() +# The request's flattened metadata as identity seeding last read it, kept beside the +# destinations so the fan-out can export the extra ``metadata_keys`` a destination asked +# for onto that destination's spans alone, never into the Baggage every exporter reads. +_request_metadata: Final["ContextVar[Mapping[str, str]]"] = ContextVar( + "litellm_otel_request_metadata", default=MappingProxyType({}) +) + + +def set_request_metadata(metadata: Mapping[str, str]) -> None: + """Record the flattened metadata of the request on this context.""" + _request_metadata.set(metadata) + + +def request_metadata() -> Mapping[str, str]: + """The flattened metadata seeded for this request, empty before seeding or outside one.""" + return _request_metadata.get() + + #: ``litellm_settings: otel_tenant_destination_mode`` and its env equivalent. ADDITIVE_DESTINATION_MODE: Final = "additive" OTEL_TENANT_DESTINATION_MODE_ENV: Final = "LITELLM_OTEL_TENANT_DESTINATION_MODE" diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index b5a5b85fc11..0941c065e74 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -24,7 +24,7 @@ from opentelemetry.sdk._logs.export import ( ) from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import Event, ReadableSpan, SpanProcessor, TracerProvider +from opentelemetry.sdk.trace import Event, ReadableSpan, SpanLimits, SpanProcessor, TracerProvider from opentelemetry.sdk.trace import Span as SDKSpan from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, @@ -49,6 +49,7 @@ from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_TRACE_NAME, LangfuseMapper, ) +from litellm.integrations.otel.model.baggage import promoted_metadata from litellm.integrations.otel.model.config import ( CaptureMessageContent, ExporterOwner, @@ -71,6 +72,7 @@ from litellm.integrations.otel.model.semconv import ( from litellm.integrations.otel.model.spans import LiteLLMSpanKind from litellm.integrations.otel.plumbing.context import ( request_destinations, + request_metadata, request_traces_excluded, suppressed_backends, ) @@ -469,9 +471,13 @@ def without_message_content(span: ReadableSpan) -> ReadableSpan: class _CountedSpanView(_SpanView): """A ``_SpanView`` of ``inner`` with other attributes that reports the dropped counts ``inner`` reports.""" - def __init__(self, inner: ReadableSpan, attributes: Attributes) -> None: + def __init__(self, inner: ReadableSpan, attributes: Attributes, dropped_attributes: int = 0) -> None: super().__init__(inner, inner.resource, attributes, inner.events, inner.status, parent=inner.parent) - self._dropped_counts: Final = (inner.dropped_attributes, inner.dropped_events, inner.dropped_links) + self._dropped_counts: Final = ( + inner.dropped_attributes + dropped_attributes, + inner.dropped_events, + inner.dropped_links, + ) @property def dropped_attributes(self) -> int: @@ -486,6 +492,60 @@ class _CountedSpanView(_SpanView): return self._dropped_counts[2] +def span_limits(span: object) -> SpanLimits: + """The limits ``span`` was built with; the env-configured defaults for a span the SDK did not build.""" + if not isinstance(span, SDKSpan): + return SpanLimits() + return span._limits # pyright: ignore[reportPrivateUsage] # the SDK has no public accessor for a span's limits + + +class _PendingMetadataSpan(_CountedSpanView): + """``inner`` as it ended, carrying the destination metadata added once the destination's view is built.""" + + def __init__(self, inner: ReadableSpan, metadata: tuple[tuple[str, str], ...], limits: SpanLimits) -> None: + super().__init__(inner, inner.attributes) + self.destination_metadata: Final = metadata + self.limits: Final = limits + + +def with_destination_metadata( + span: ReadableSpan, metadata_keys: tuple[str, ...], metadata: Mapping[str, str], limits: SpanLimits +) -> ReadableSpan: + """``span`` carrying the ``litellm.metadata.*`` values a destination's own ``metadata_keys`` select. + + Bound when the span ends, so a tree held for sampling keeps its own request's values, along + with the span limits it ended under. :func:`with_pending_metadata` adds them once the + destination's view is built. + """ + if not metadata_keys: + return span + promoted: Final = tuple(promoted_metadata(metadata, metadata_keys).items()) + if not promoted: + return span + return _PendingMetadataSpan(span, promoted, limits) + + +def with_pending_metadata(ended: ReadableSpan, exported: ReadableSpan) -> ReadableSpan: + """``exported``, the destination's view of ``ended``, plus the metadata ``ended`` carries. + + Counted against what the destination exports rather than the span as it ended, since the view + can drop prompt content and backend-specific attributes. An exported attribute always wins, so a + destination key never replaces a value the global allowlist or the proxy's identity wrote. + Values are cut to the length limit, and keys past the count limit are dropped and counted. + """ + if not isinstance(ended, _PendingMetadataSpan): + return exported + attributes: Final = exported.attributes or _NO_ATTRIBUTES + extra: Final = tuple((key, value) for key, value in ended.destination_metadata if key not in attributes) + max_attributes: Final = ended.limits.max_span_attributes + max_length: Final = ended.limits.max_span_attribute_length + room: Final = len(extra) if max_attributes is None else max(0, max_attributes - len(attributes)) + kept: Final = {key: value if max_length is None else value[:max_length] for key, value in extra[:room]} + return _CountedSpanView( + exported, MappingProxyType({**attributes, **kept}), dropped_attributes=len(extra) - len(kept) + ) + + def _is_litellm_span(span: ReadableSpan) -> bool: """Whether LiteLLM's own tracer wrote ``span``, rather than application or instrumentation code.""" scope: Final = span.instrumentation_scope @@ -761,9 +821,11 @@ class TenantFanOutSpanProcessor(SpanProcessor): self._span_roots: weakref.WeakKeyDictionary[SDKSpan, int] = weakref.WeakKeyDictionary() self._verdicts: OrderedDict[_VerdictKey, bool] = OrderedDict() # mutable-ok: bounded LRU self._drain: Final = drain_pool if drain_pool is not None else _DrainPool(capacity=pending_drains) + self._span_limits = SpanLimits() def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: context: Final = span.context + self._span_limits = span_limits(span) with self._lock: request: Final = self._request_of(span, trace.get_current_span(parent_context)) if context is not None: @@ -782,8 +844,11 @@ class TenantFanOutSpanProcessor(SpanProcessor): if span.context is not None: self._roots.pop(span.context.span_id, None) failed: Final = span.status.status_code is StatusCode.ERROR - for destination in request_destinations(): - if ( + metadata: Final = request_metadata() + deliveries: Final = tuple( + (destination, with_destination_metadata(span, destination.metadata_keys, metadata, self._span_limits)) + for destination in request_destinations() + if not ( self._operator_already_writes(span, destination, suppressed) or not _in_scope(span, destination.span_scope) or _is_excluded_database_span( @@ -792,12 +857,13 @@ class TenantFanOutSpanProcessor(SpanProcessor): if destination.excluded_db_systems is None else destination.excluded_db_systems, ) - ): - continue + ) + ) + for destination, view in deliveries: if not _is_sampled(destination): - self._forward(span, destination) + self._forward(view, destination) continue - for held_span, held_destination in self._route(request, span, destination, failed): + for held_span, held_destination in self._route(request, view, destination, failed): self._forward(held_span, held_destination) for held_span, held_destination in self._settle(request, span, failed): self._forward(held_span, held_destination) @@ -834,12 +900,11 @@ class TenantFanOutSpanProcessor(SpanProcessor): if processor is None: return try: - processor.on_end( - _scoped( - _for_destination(span, destination, self._default_capture_for(destination), self._langfuse_mapped), - destination.span_scope, - ) + exported: Final = _scoped( + _for_destination(span, destination, self._default_capture_for(destination), self._langfuse_mapped), + destination.span_scope, ) + processor.on_end(with_pending_metadata(span, exported)) except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc) finally: diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index 984b385a647..0bd40824066 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -208,6 +208,7 @@ def destination_for( service_name: str | None = None, capture_message_content: str | None = None, excluded_db_systems: frozenset[str] | None = None, + metadata_keys: tuple[str, ...] | None = None, ) -> OtelDestination | None: """The destination ``params`` names for ``callback_name``, or ``None``. @@ -240,6 +241,7 @@ def destination_for( span_scope=_span_scope(callback_name, params), capture_message_content=capture_message_content, excluded_db_systems=excluded_db_systems, + metadata_keys=metadata_keys or (), success_sampling_rate=success_sampling_rate, error_sampling_rate=error_sampling_rate, ) diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py index c3769686e45..cdb009fcf77 100644 --- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py +++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py @@ -1,12 +1,16 @@ import re -from collections.abc import Generator, Iterator, Mapping +from collections.abc import Callable, Generator, Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar +from types import MappingProxyType from typing import Any, Final from litellm.types.utils import ( ARIZE_OTLP_PROTOCOLS, + BAGGAGE_METADATA_KEYS_VAR, CAPTURE_MESSAGE_CONTENT_VALUES, + CAPTURE_MESSAGE_CONTENT_VAR, + EXCLUDED_SERVICES_VAR, OTEL_SPAN_SCOPES, TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams, @@ -99,6 +103,14 @@ def validate_capture_message_content_value(value: str) -> None: ) +def validate_baggage_metadata_keys_value(value: str) -> None: + from litellm.integrations.otel.model.baggage import destination_metadata_keys_error + + error: Final = destination_metadata_keys_error(value) + if error is not None: + raise ValueError(error) + + def validate_excluded_services_value(value: str) -> None: from litellm.integrations.otel.model.config import invalid_excluded_service_names @@ -111,6 +123,15 @@ def validate_excluded_services_value(value: str) -> None: ) +OTEL_DESTINATION_VALIDATORS: Final[Mapping[str, Callable[[str], None]]] = MappingProxyType( + { + CAPTURE_MESSAGE_CONTENT_VAR: validate_capture_message_content_value, + EXCLUDED_SERVICES_VAR: validate_excluded_services_value, + BAGGAGE_METADATA_KEYS_VAR: validate_baggage_metadata_keys_value, + } +) + + # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict _supported_callback_params: Final[tuple[str, ...]] = ( "langfuse_public_key", diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 0d40faf55fb..80819542d68 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -31,9 +31,8 @@ from typing_extensions import ( from litellm._uuid import uuid from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( + OTEL_DESTINATION_VALIDATORS, validate_arize_otlp_protocol_value, - validate_capture_message_content_value, - validate_excluded_services_value, validate_langfuse_environment_value, validate_langfuse_span_scope_value, validate_no_callback_env_reference, @@ -96,8 +95,7 @@ from litellm.types.router import AllowedModelRegion, RouterErrors, UpdateRouterC from litellm.types.router_weights import validate_router_settings_dict from litellm.types.secret_managers.main import KeyManagementSystem from litellm.types.utils import ( - CAPTURE_MESSAGE_CONTENT_VAR, - EXCLUDED_SERVICES_VAR, + OTEL_DESTINATION_VARS, AzureSpillover, CallTypes, CostBreakdown, @@ -2136,9 +2134,7 @@ class BlockModelRequest(LiteLLMPydanticObjectBase): model_id: str # required -CALLBACK_VAR_NAMES: Final = frozenset( - {*StandardCallbackDynamicParams.__annotations__, CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR} -) +CALLBACK_VAR_NAMES: Final = frozenset({*StandardCallbackDynamicParams.__annotations__, *OTEL_DESTINATION_VARS}) class AddTeamCallback(LiteLLMPydanticObjectBase): @@ -2153,7 +2149,7 @@ class AddTeamCallback(LiteLLMPydanticObjectBase): for key, value in callback_vars.items(): if key not in CALLBACK_VAR_NAMES: raise ValueError(f"Invalid callback variable: {key}. Must be one of {set(CALLBACK_VAR_NAMES)}") - if key in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) and not isinstance(value, str): + if key in OTEL_DESTINATION_VARS and not isinstance(value, str): raise ValueError(f"{key} must be a string, got {type(value).__name__}") callback_vars[key] = str(value) validate_no_callback_env_reference(key, callback_vars[key], source="key/team callback metadata") @@ -2163,10 +2159,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase): validate_langfuse_span_scope_value(callback_vars[key]) if key == "arize_otlp_protocol": validate_arize_otlp_protocol_value(callback_vars[key]) - if key == CAPTURE_MESSAGE_CONTENT_VAR: - validate_capture_message_content_value(callback_vars[key]) - if key == EXCLUDED_SERVICES_VAR: - validate_excluded_services_value(callback_vars[key]) + if (destination_validator := OTEL_DESTINATION_VALIDATORS.get(key)) is not None: + destination_validator(callback_vars[key]) return values diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py index a486d6665fc..a1c2ff8eb52 100644 --- a/litellm/proxy/common_utils/callback_config_validation.py +++ b/litellm/proxy/common_utils/callback_config_validation.py @@ -14,7 +14,11 @@ from typing import Final, Literal, cast # noqa: TID251 # preserve arbitrary ke from pydantic import TypeAdapter, ValidationError from litellm.litellm_core_utils.credential_accessor import CredentialAccessor -from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR +from litellm.types.utils import ( + BAGGAGE_METADATA_KEYS_VAR, + EXCLUDED_SERVICES_VAR, + OTEL_DESTINATION_VARS, +) _NEWRELIC_CALLBACK: Final = "newrelic" _NEWRELIC_VAR_PREFIX: Final = "newrelic_" @@ -42,18 +46,12 @@ def callback_config_error( ) if langfuse_error is not None: return langfuse_error - from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( - validate_capture_message_content_value, - validate_excluded_services_value, - ) + from litellm.litellm_core_utils.initialize_dynamic_callback_params import OTEL_DESTINATION_VALIDATORS destination_setting_error: Final = next( ( error - for variable_name, validator in ( - (CAPTURE_MESSAGE_CONTENT_VAR, validate_capture_message_content_value), - (EXCLUDED_SERVICES_VAR, validate_excluded_services_value), - ) + for variable_name, validator in OTEL_DESTINATION_VALIDATORS.items() if ( error := _destination_setting_error( variable_name, callback_name, callback_vars, callback_type, validator @@ -269,11 +267,17 @@ def conflicting_callback_var_error( def _same_callback_var_value(variable_name: str, first: str, second: str) -> bool: - if variable_name != EXCLUDED_SERVICES_VAR: - return first == second - from litellm.integrations.otel.model.config import parse_excluded_services + if variable_name == EXCLUDED_SERVICES_VAR: + from litellm.integrations.otel.model.config import parse_excluded_services - return parse_excluded_services(first) == parse_excluded_services(second) + return parse_excluded_services(first) == parse_excluded_services(second) + if variable_name == BAGGAGE_METADATA_KEYS_VAR: + from litellm.integrations.otel.model.baggage import parse_destination_metadata_keys + + first_keys: Final = parse_destination_metadata_keys(first) + second_keys: Final = parse_destination_metadata_keys(second) + return first_keys is not None and second_keys is not None and frozenset(first_keys) == frozenset(second_keys) + return first == second def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None: @@ -331,7 +335,7 @@ def _conflicting_entry_setting_error( return next( ( error - for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + for variable_name in OTEL_DESTINATION_VARS if (error := conflicting_callback_var_error(callback_name, variable_name, callback_vars, earlier)) is not None ), @@ -364,7 +368,7 @@ def _logging_entry_error(entry: object) -> str | None: destination_var_error: Final = next( ( f"{variable_name} must be a string, got {type(callback_vars[variable_name]).__name__}" - for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + for variable_name in OTEL_DESTINATION_VARS if variable_name in callback_vars and not isinstance(callback_vars[variable_name], str) ), None, diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index ff0883b6958..b1eb8f7d8ae 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -332,8 +332,10 @@ from litellm.types.router import BaggageMappedTagsStamp, holds_api_key_secret_re from litellm.types.services import ServiceTypes from litellm.types.utils import ( API_KEY_SECRET_REFERENCE_REJECTION, + BAGGAGE_METADATA_KEYS_VAR, CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR, + OTEL_DESTINATION_VARS, CustomPricingLiteLLMParams, LlmProviders, ProviderSpecificHeader, @@ -342,7 +344,7 @@ from litellm.types.utils import ( SupportedCacheControls, ) -_DESTINATION_ONLY_CALLBACK_VARS: Final[frozenset[str]] = frozenset({CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR}) +_DESTINATION_ONLY_CALLBACK_VARS: Final[frozenset[str]] = frozenset(OTEL_DESTINATION_VARS) _CALLBACK_CREDENTIAL_KEYS: Final = frozenset(StandardCallbackDynamicParams.__annotations__) | frozenset( {TRUSTED_CALLBACK_VARS_FIELD} @@ -1479,6 +1481,7 @@ def resolve_tenant_otel_destinations( def _tenant_destination( name: str, callbacks: Sequence[AddTeamCallback], service_name: str | None ) -> "OtelDestination | None": + from litellm.integrations.otel.model.baggage import parse_destination_metadata_keys from litellm.integrations.otel.model.config import parse_capture_message_content, parse_excluded_services from litellm.integrations.otel.presets.destinations import destination_for @@ -1495,6 +1498,7 @@ def _tenant_destination( service_name, parse_capture_message_content(callback_vars.get(CAPTURE_MESSAGE_CONTENT_VAR)), parse_excluded_services(callback_vars.get(EXCLUDED_SERVICES_VAR)), + parse_destination_metadata_keys(callback_vars.get(BAGGAGE_METADATA_KEYS_VAR)), ) diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index 5b778392664..1d35d34cbcf 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -60,7 +60,7 @@ from litellm.proxy.management_endpoints.team_endpoints import ( # noqa: F401 # ) from litellm.proxy.management_helpers.utils import management_endpoint_wrapper from litellm.repositories.team_repository import TeamRepository -from litellm.types.utils import CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR +from litellm.types.utils import OTEL_DESTINATION_VARS router: Final = APIRouter() @@ -304,6 +304,7 @@ async def add_team_callbacks( An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting. "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out - excluded_services: For OTel v2 callbacks, a comma-separated list of datastore names such as "redis,postgres", or "none" to exclude nothing. An explicit value replaces the global exclusions; omitted follows them + - baggage_metadata_keys: For OTel v2 callbacks, a comma-separated list of request metadata keys such as "requester_metadata.correlation_id", exported as litellm.metadata. on this destination's spans only, on top of the global LITELLM_OTEL_BAGGAGE_METADATA_KEYS. At most 20 keys; names of proxy identity fields (user_api_key_*, requester_ip_address) and request headers (requester_metadata.headers.*) are rejected - gcs_bucket_name: The name of the GCS bucket - gcs_path_service_account: The path to the GCS service account - langsmith_api_key: The API key for the Langsmith callback @@ -376,7 +377,7 @@ async def add_team_callbacks( destination_setting_error: Final = next( ( error - for variable_name in (CAPTURE_MESSAGE_CONTENT_VAR, EXCLUDED_SERVICES_VAR) + for variable_name in OTEL_DESTINATION_VARS if ( error := conflicting_callback_var_error( data.callback_name, variable_name, data.callback_vars, stored_callbacks diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 81d39f848fb..5990135a5cb 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -4029,6 +4029,13 @@ ARIZE_OTLP_PROTOCOLS: Final[frozenset[str]] = frozenset(get_args(ArizeOtlpProtoc CAPTURE_MESSAGE_CONTENT_VAR: Final = "capture_message_content" EXCLUDED_SERVICES_VAR: Final = "excluded_services" +BAGGAGE_METADATA_KEYS_VAR: Final = "baggage_metadata_keys" +#: Team and key callback vars that tune that callback's own OTel v2 destination, never sent to the backend. +OTEL_DESTINATION_VARS: Final[tuple[str, ...]] = ( + CAPTURE_MESSAGE_CONTENT_VAR, + EXCLUDED_SERVICES_VAR, + BAGGAGE_METADATA_KEYS_VAR, +) class CaptureMessageContent(str): diff --git a/tests/integration/observability/test_otel_destination_metadata_keys.py b/tests/integration/observability/test_otel_destination_metadata_keys.py new file mode 100644 index 00000000000..33b18fa4c2e --- /dev/null +++ b/tests/integration/observability/test_otel_destination_metadata_keys.py @@ -0,0 +1,262 @@ +"""A team's or key's own OTel v2 destination can list extra request metadata keys in ``baggage_metadata_keys``, +exported as ``litellm.metadata.*`` on that destination's spans only, on top of the operator's global allowlist.""" + +from __future__ import annotations + +import uuid +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import Final, Literal + +import httpx +import pytest +from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment +from integration._support.otlp_sink import ( + ConnectSink, + Span, + SpanSinks, + audit_configuration_writer, + owned_connect_sink, + owned_sinks, + recorded_spans, + spans_for_trace, +) +from integration._support.process import owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue + +OPERATION: Final = "gen_ai.operation.name" +SERVER_KIND: Final = 2 +METADATA_PREFIX: Final = "litellm.metadata." +TEAM_KEYS: Final = "requester_metadata.tenant_id, requester_metadata.order.id" + +Backend = Literal["newrelic", "arize"] + + +def _chat(_request: Request) -> Reply: + return Reply( + body=( + b'{"id":"chatcmpl-meta","object":"chat.completion","created":1,"model":"gpt-4o-mini",' + b'"choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],' + b'"usage":{"prompt_tokens":7,"completion_tokens":2,"total_tokens":9}}' + ) + ) + + +@dataclass(frozen=True, slots=True) +class Rig: + proxy: Gateway + sinks: SpanSinks + newrelic: ConnectSink + upstream: Wire + + def sink(self, backend: Backend) -> str: + return self.newrelic.control_url if backend == "newrelic" else self.sinks.arize + + def destination(self, backend: Backend, extra: Mapping[str, str]) -> dict[str, JsonValue]: + tag: Final = uuid.uuid4().hex + credentials: Final[Mapping[str, str]] = ( + {"newrelic_api_key": f"nr-{tag}", "newrelic_region": "us"} + if backend == "newrelic" + else {"arize_space_key": f"space-{tag}", "arize_api_key": f"arize-{tag}"} + ) + return { + "callback_name": backend, + "callback_type": "success_and_failure", + "callback_vars": {**credentials, **extra}, + } + + def team_key(self, scenario: Scenario, backend: Backend, extra: Mapping[str, str]) -> str: + team: Final = scenario.team() + attached: Final = self.proxy.request("POST", f"/team/{team}/callback", self.destination(backend, extra)) + assert attached.status_code == 200, attached.text + return scenario.key(team_id=team) + + def send(self, key: str, model: str, metadata: Mapping[str, JsonValue]) -> str: + trace_id: Final = uuid.uuid4().hex + response: Final = self.proxy.request( + "POST", + "/v1/chat/completions", + {"model": model, "messages": [{"role": "user", "content": "hi"}], "metadata": dict(metadata)}, + key=key, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + assert response.status_code == 200, response.text + return trace_id + + def request_spans(self, url: str, since: int, trace_id: str) -> tuple[Span, ...]: + return _server_and_model_spans( + eventually( + lambda: spans_for_trace(recorded_spans(url, since)[1], trace_id), + lambda found: len(_server_and_model_spans(found)) >= 2, + seconds=40, + ) + ) + + +@pytest.fixture(scope="module") +def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + directory: Final = tmp_path_factory.mktemp("destination-metadata-keys") + with ( + wire_server(_chat) as upstream, + owned_sinks(directory / "sinks") as sinks, + owned_connect_sink(directory / "newrelic") as newrelic, + gateway_from_environment() as gateway, + owned_proxy_process( + gateway, + directory, + { + "LITELLM_OTEL_V2": "1", + "OTEL_BSP_SCHEDULE_DELAY": "300", + "LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.correlation_id", + "ARIZE_HTTP_ENDPOINT": sinks.arize + "/v1/traces", + "HTTPS_PROXY": newrelic.proxy_url, + "NO_PROXY": "127.0.0.1,localhost", + "REQUESTS_CA_BUNDLE": newrelic.ca_pem, + }, + config=audit_configuration_writer(sinks)(directory, {}), + remove_environment=("OTEL_ENDPOINT", "OTEL_EXPORTER", "OTEL_EXPORTER_OTLP_ENDPOINT"), + workers=2, + ) as owned, + ): + yield Rig(owned.gateway, sinks, newrelic, upstream) + + +def _server_and_model_spans(spans: tuple[Span, ...]) -> tuple[Span, ...]: + return tuple(span for span in spans if span["kind"] == SERVER_KIND or OPERATION in span["attributes"]) + + +def _exported_metadata(spans: tuple[Span, ...]) -> list[dict[str, JsonValue]]: + return [ + {name: value for name, value in span["attributes"].items() if name.startswith(METADATA_PREFIX)} + for span in spans + ] + + +def _request_metadata(marker: str) -> dict[str, JsonValue]: + return { + "correlation_id": f"corr-{marker}", + "tenant_id": f"tenant-{marker}", + "order": {"id": f"order-{marker}"}, + "unlisted": f"unlisted-{marker}", + } + + +@pytest.mark.parametrize("level", ["team", "key"]) +@pytest.mark.parametrize("backend", ["newrelic", "arize"]) +def test_destination_keys_reach_that_destination_and_not_the_operator( + rig: Rig, backend: Backend, level: Literal["team", "key"] +) -> None: + with rig.proxy.scenario() as scenario: + model: Final = scenario.model(api_base=rig.upstream.url + "/v1") + keys: Final = {"baggage_metadata_keys": TEAM_KEYS} + key: Final = ( + rig.team_key(scenario, backend, keys) + if level == "team" + else scenario.key(metadata={"logging": [rig.destination(backend, keys)]}) + ) + tenant_since: Final = recorded_spans(rig.sink(backend))[0] + operator_since: Final = recorded_spans(rig.sinks.operator)[0] + marker: Final = uuid.uuid4().hex + trace_id: Final = rig.send(key, model, _request_metadata(marker)) + tenant: Final = rig.request_spans(rig.sink(backend), tenant_since, trace_id) + operator: Final = rig.request_spans(rig.sinks.operator, operator_since, trace_id) + + assert sorted(span["kind"] == SERVER_KIND for span in tenant) == [False, True], [span["name"] for span in tenant] + assert _exported_metadata(tenant) == [ + { + "litellm.metadata.correlation_id": f"corr-{marker}", + "litellm.metadata.tenant_id": f"tenant-{marker}", + "litellm.metadata.order.id": f"order-{marker}", + } + ] * len(tenant), [(span["name"], sorted(span["attributes"])) for span in tenant] + assert _exported_metadata(operator) == [{"litellm.metadata.correlation_id": f"corr-{marker}"}] * len(operator), [ + (span["name"], sorted(span["attributes"])) for span in operator + ] + + +@pytest.mark.parametrize("backend", ["newrelic", "arize"]) +def test_a_destination_without_keys_and_a_request_body_cannot_add_them(rig: Rig, backend: Backend) -> None: + with rig.proxy.scenario() as scenario: + model: Final = scenario.model(api_base=rig.upstream.url + "/v1") + listing: Final = rig.team_key(scenario, backend, {"baggage_metadata_keys": TEAM_KEYS}) + sibling: Final = rig.team_key(scenario, backend, {}) + since: Final = recorded_spans(rig.sink(backend))[0] + marker: Final = uuid.uuid4().hex + listed_trace: Final = rig.send(listing, model, _request_metadata(f"listed-{marker}")) + sibling_trace: Final = rig.send( + sibling, + model, + { + **_request_metadata(marker), + "baggage_metadata_keys": TEAM_KEYS, + "callback_vars": {"baggage_metadata_keys": TEAM_KEYS}, + }, + ) + listed: Final = rig.request_spans(rig.sink(backend), since, listed_trace) + unlisted: Final = rig.request_spans(rig.sink(backend), since, sibling_trace) + + assert {"litellm.metadata.tenant_id": f"tenant-listed-{marker}"}.items() <= _exported_metadata(listed)[0].items() + assert _exported_metadata(unlisted) == [{"litellm.metadata.correlation_id": f"corr-{marker}"}] * len(unlisted), [ + (span["name"], sorted(span["attributes"])) for span in unlisted + ] + + +def _attach(rig: Rig, team: str, keys: str, *, key: str | None = None) -> httpx.Response: + return rig.proxy.request( + "POST", + f"/team/{team}/callback", + rig.destination("newrelic", {"baggage_metadata_keys": keys}), + key=key, + ) + + +def test_a_team_admin_can_set_keys_and_a_plain_member_cannot(rig: Rig) -> None: + with rig.proxy.scenario() as scenario: + team: Final = scenario.team() + admin_key: Final = scenario.key(user_id=scenario.member(team, role="admin"), team_id=team) + member_key: Final = scenario.key(user_id=scenario.member(team), team_id=team) + + refused: Final = _attach(rig, team, "requester_metadata.tenant_id", key=member_key) + accepted: Final = _attach(rig, team, "requester_metadata.tenant_id", key=admin_key) + + assert refused.status_code == 403, refused.text + assert accepted.status_code == 200, accepted.text + + +@pytest.mark.parametrize( + ("keys", "reason"), + [ + ("requester_metadata.user_api_key_user_id", "proxy identity attribute"), + ("tenant_id", "requester_metadata."), + ("requester_metadata.headers.x-internal-token", "would export request headers"), + (",".join(f"requester_metadata.k{index}" for index in range(21)), "at most 20 keys"), + ], +) +def test_unusable_keys_are_refused_on_team_and_key_callbacks(rig: Rig, keys: str, reason: str) -> None: + with rig.proxy.scenario() as scenario: + team_response: Final = _attach(rig, scenario.team(), keys) + key_response: Final = rig.proxy.request( + "POST", + "/key/generate", + {"metadata": {"logging": [rig.destination("newrelic", {"baggage_metadata_keys": keys})]}}, + ) + + assert [(response.status_code, reason in response.text) for response in (team_response, key_response)] == [ + (422, True), + (400, True), + ], [team_response.text, key_response.text] + + +def test_unset_keys_leave_the_destination_on_the_global_allowlist(rig: Rig) -> None: + with rig.proxy.scenario() as scenario: + model: Final = scenario.model(api_base=rig.upstream.url + "/v1") + key: Final = rig.team_key(scenario, "newrelic", MappingProxyType({})) + since: Final = recorded_spans(rig.newrelic.control_url)[0] + marker: Final = uuid.uuid4().hex + spans: Final = rig.request_spans( + rig.newrelic.control_url, since, rig.send(key, model, _request_metadata(marker)) + ) + + assert _exported_metadata(spans) == [{"litellm.metadata.correlation_id": f"corr-{marker}"}] * len(spans) diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index 6dd9ff35026..256b9e5e480 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -61,6 +61,7 @@ from litellm.integrations.otel.plumbing.context import ( destination_backends, request_destinations, set_request_destinations, + set_request_metadata, ) from litellm.integrations.otel.plumbing.providers import ( TenantFanOutSpanProcessor, @@ -589,6 +590,132 @@ class TestFanOut: "chat gpt-4", } + def test_destination_metadata_keys_reach_only_that_destination(self): + team, sibling, operator = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter() + exporters: Final = {"http://team.local": team, "http://sibling.local": sibling} + provider: Final = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(operator)) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda destination: SimpleSpanProcessor(exporters[destination.endpoint]) + ) + ) + + def run(): + set_request_destinations( + ( + OtelDestination( + endpoint="http://team.local", + metadata_keys=("requester_metadata.tenant_id", "requester_metadata.order.id"), + ), + OtelDestination(endpoint="http://sibling.local"), + ) + ) + set_request_metadata( + MappingProxyType( + { + "requester_metadata.tenant_id": "tenant-1", + "requester_metadata.order.id": "order-1", + "requester_metadata.unlisted": "nope", + } + ) + ) + emit(provider) + + in_fresh_context(run) + + team_attributes: Final = {key: value for key, value in (team.get_finished_spans()[0].attributes or {}).items()} + assert {key: value for key, value in team_attributes.items() if key.startswith("litellm.metadata.")} == { + "litellm.metadata.tenant_id": "tenant-1", + "litellm.metadata.order.id": "order-1", + } + assert [ + key + for span in (*sibling.get_finished_spans(), *operator.get_finished_spans()) + for key in span.attributes or {} + if key.startswith("litellm.metadata.") + ] == [] + + @staticmethod + def _team_export( + metadata: Mapping[str, str], + attributes: Mapping[str, str] = MappingProxyType({}), + limits: SpanLimits | None = None, + ) -> ReadableSpan: + team: Final = InMemorySpanExporter() + provider: Final = TracerProvider() if limits is None else TracerProvider(span_limits=limits) + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=lambda _: SimpleSpanProcessor(team))) + + def run(): + set_request_destinations((OtelDestination(endpoint="http://team.local", metadata_keys=tuple(metadata)),)) + set_request_metadata(MappingProxyType(dict(metadata))) + with get_tracer(provider, "litellm").start_as_current_span("chat gpt-4") as span: + span.set_attributes(dict(attributes)) + + in_fresh_context(run) + return team.get_finished_spans()[0] + + def test_destination_metadata_key_never_replaces_an_attribute_the_span_carries(self): + exported: Final = self._team_export( + {"requester_metadata.tenant_id": "from-caller"}, + {"litellm.metadata.tenant_id": "from-operator-allowlist"}, + ) + + assert (exported.attributes or {})["litellm.metadata.tenant_id"] == "from-operator-allowlist" + + def test_destination_metadata_obeys_the_span_value_length_limit(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "8") + exported: Final = self._team_export({"requester_metadata.tenant_id": "x" * 1_048_576}) + + assert (exported.attributes or {})["litellm.metadata.tenant_id"] == "x" * 8 + + def test_destination_metadata_obeys_a_providers_own_span_limits(self): + exported: Final = self._team_export( + {"requester_metadata.a": "abcdefgh", "requester_metadata.b": "2", "requester_metadata.c": "3"}, + {"litellm.team.id": "team-a", "litellm.metadata.correlation_id": "corr"}, + SpanLimits(max_span_attributes=3, max_span_attribute_length=4), + ) + + assert dict(exported.attributes or {}) == { + "litellm.team.id": "team", + "litellm.metadata.correlation_id": "corr", + "litellm.metadata.a": "abcd", + } + assert exported.dropped_attributes == 2 + + def test_destination_metadata_counts_against_the_view_after_content_is_redacted(self): + prompt: Final = {f"llm.input_messages.{index}.message.content": "hello" for index in range(4)} + exported: Final = self._team_export( + {"requester_metadata.correlation_id": "corr-1"}, prompt, SpanLimits(max_span_attributes=4) + ) + + assert dict(exported.attributes or {}) == {"litellm.metadata.correlation_id": "corr-1"} + assert exported.dropped_attributes == 0 + + def test_sampled_destination_keeps_the_metadata_of_the_request_that_ended_its_spans(self): + team: Final = InMemorySpanExporter() + provider: Final = TracerProvider() + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=lambda _: SimpleSpanProcessor(team))) + destination: Final = OtelDestination( + endpoint="http://team.local", + metadata_keys=("requester_metadata.tenant_id",), + success_sampling_rate=1.0, + error_sampling_rate=1.0, + ) + + def run(tenant: str): + set_request_destinations((destination,)) + set_request_metadata(MappingProxyType({"requester_metadata.tenant_id": tenant})) + emit(provider) + + in_fresh_context(run, "tenant-a") + in_fresh_context(run, "tenant-b") + + assert sorted(span.attributes["litellm.metadata.tenant_id"] for span in team.get_finished_spans()) == [ + "tenant-a", + "tenant-b", + ] + def test_team_exclusions_replace_empty_global_and_operator_keeps_all_database_spans(self): team, inherited, operator = InMemorySpanExporter(), InMemorySpanExporter(), InMemorySpanExporter() exporters: Final = {"http://team.local": team, "http://inherited.local": inherited} diff --git a/tests/unit/proxy/common_utils/test_callback_config_validation.py b/tests/unit/proxy/common_utils/test_callback_config_validation.py index ac1fe2d785e..19cac032c8a 100644 --- a/tests/unit/proxy/common_utils/test_callback_config_validation.py +++ b/tests/unit/proxy/common_utils/test_callback_config_validation.py @@ -345,7 +345,7 @@ def test_an_unsupported_capture_message_content_is_rejected_on_key_logging_metad @pytest.mark.usefixtures("otel_v2_on") -@pytest.mark.parametrize("callback_var", ["capture_message_content", "excluded_services"]) +@pytest.mark.parametrize("callback_var", ["capture_message_content", "excluded_services", "baggage_metadata_keys"]) def test_key_logging_metadata_rejects_non_string_destination_callback_vars(callback_var: str) -> None: error: Final = logging_metadata_config_error( { @@ -575,3 +575,72 @@ def test_a_non_string_credential_name_is_not_a_reference(name: object) -> None: credential_reference_rejection({"logging": [entry]}, None, is_proxy_admin=False, is_defined=lambda _: False) is None ) + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize("callback_name", ["langfuse_otel", "arize", "weave_otel", "newrelic"]) +def test_baggage_metadata_keys_is_accepted_on_every_otel_v2_destination(callback_name: str) -> None: + assert ( + callback_config_error( + callback_name, {"baggage_metadata_keys": "requester_metadata.tenant_id, requester_metadata.order.id"} + ) + is None + ) + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize("callback_name", ["langfuse", "datadog", "otel", "arize_phoenix", None]) +def test_baggage_metadata_keys_is_rejected_where_it_would_never_take_effect(callback_name: str | None) -> None: + error: Final = callback_config_error(callback_name, {"baggage_metadata_keys": "requester_metadata.tenant_id"}) + assert error is not None and "baggage_metadata_keys" in error and "newrelic" in error + + +@pytest.mark.usefixtures("otel_v2_on") +@pytest.mark.parametrize( + ("value", "reason"), + [ + ("tenant_id", "requester_metadata."), + ("requester_metadata.", "requester_metadata."), + ("requester_metadata.tenant_id,", "requester_metadata."), + ("user_api_key_user_id", "requester_metadata."), + ("requester_metadata.user_api_key_user_id", "proxy identity attribute"), + ("requester_metadata.user_api_key_hash", "proxy identity attribute"), + ("requester_metadata.requester_ip_address", "proxy identity attribute"), + ("requester_metadata.headers.x-internal-token", "would export request headers"), + ("requester_metadata.headers", "would export request headers"), + ("requester_metadata.Headers.authorization", "would export request headers"), + (",".join(f"requester_metadata.k{index}" for index in range(21)), "at most 20 keys"), + ], +) +def test_unusable_baggage_metadata_keys_are_rejected(value: str, reason: str) -> None: + error: Final = callback_config_error("newrelic", {"newrelic_api_key": "nr-key", "baggage_metadata_keys": value}) + assert error is not None and reason in error, error + + +def test_baggage_metadata_keys_is_rejected_while_otel_v2_is_off(monkeypatch: pytest.MonkeyPatch) -> None: + from litellm.integrations.otel.model.config import is_otel_v2_enabled + + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + is_otel_v2_enabled.cache_clear() + try: + error: Final = callback_config_error("newrelic", {"baggage_metadata_keys": "requester_metadata.tenant_id"}) + finally: + is_otel_v2_enabled.cache_clear() + assert error is not None and "baggage_metadata_keys" in error and "LITELLM_OTEL_V2" in error + + +@pytest.mark.parametrize( + ("stored", "rejected"), + [ + ("requester_metadata.b, requester_metadata.a", False), + ("requester_metadata.a", True), + ], +) +def test_entries_for_one_backend_share_one_baggage_metadata_keys_list(stored: str, rejected: bool) -> None: + error: Final = conflicting_callback_var_error( + "newrelic", + "baggage_metadata_keys", + {"baggage_metadata_keys": "requester_metadata.a,requester_metadata.b"}, + [("newrelic", {"baggage_metadata_keys": stored})], + ) + assert (error is not None) is rejected, error diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index f5f75e65c83..6daedc3bfbd 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -18283,6 +18283,7 @@ export interface paths { * An explicit value overrides the global setting for this destination; omitted follows the proxy's own callback for that backend, else the global setting. * "span_only" puts prompt and response content on the destination's spans, "no_content" leaves it out * - excluded_services: For OTel v2 callbacks, a comma-separated list of datastore names such as "redis,postgres", or "none" to exclude nothing. An explicit value replaces the global exclusions; omitted follows them + * - baggage_metadata_keys: For OTel v2 callbacks, a comma-separated list of request metadata keys such as "requester_metadata.correlation_id", exported as litellm.metadata. on this destination's spans only, on top of the global LITELLM_OTEL_BAGGAGE_METADATA_KEYS. At most 20 keys; names of proxy identity fields (user_api_key_*, requester_ip_address) and request headers (requester_metadata.headers.*) are rejected * - gcs_bucket_name: The name of the GCS bucket * - gcs_path_service_account: The path to the GCS service account * - langsmith_api_key: The API key for the Langsmith callback