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