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.<key> paths. The tenant fan-out
adds them as litellm.metadata.<key> 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.<name> 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

@@ -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.<key>`
+  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.<key>` 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.
 

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(

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,

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.<key>`` 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}<key> 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,

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.<key>`` paths exported as ``litellm.metadata.<key>`` 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.
         """

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"

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:

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

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

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
 
 

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,

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

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.<key> 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

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

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.<key>"),
+        ("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)

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}

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.<key>"),
+        ("requester_metadata.", "requester_metadata.<key>"),
+        ("requester_metadata.tenant_id,", "requester_metadata.<key>"),
+        ("user_api_key_user_id", "requester_metadata.<key>"),
+        ("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

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.<key> 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