feat(otel): resolve a request's trace destinations from its identity

litellm/integrations/otel/plumbing/context.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Mapping
 from contextvars import ContextVar, Token
-from typing import Final
+from typing import TYPE_CHECKING, Final
 
 from opentelemetry import baggage
 from opentelemetry.context import Context, get_current
@@ -18,6 +18,9 @@ from opentelemetry.trace.propagation.tracecontext import (
     TraceContextTextMapPropagator,
 )
 
+if TYPE_CHECKING:
+    from litellm.integrations.otel.model.destination import OtelDestination
+
 _PROPAGATOR: Final = TraceContextTextMapPropagator()
 
 # The request's root span — the FastAPI-owned SERVER span — captured ONCE when the
@@ -37,6 +40,20 @@ _PROPAGATOR: Final = TraceContextTextMapPropagator()
 # request task, so there is nothing to leak.
 _request_root_span: Final["ContextVar[Span | None]"] = ContextVar("litellm_otel_request_root_span", default=None)
 
+_request_destinations: 'ContextVar[tuple["OtelDestination", ...]]' = ContextVar(
+    "litellm_otel_request_destinations", default=()
+)
+
+
+def set_request_destinations(destinations: 'tuple["OtelDestination", ...]') -> None:
+    """Anchor the admin-resolved destinations for this request."""
+    _request_destinations.set(tuple(destinations))
+
+
+def request_destinations() -> 'tuple["OtelDestination", ...]':
+    """Destinations the request fans out to, or empty when none were resolved."""
+    return _request_destinations.get()
+
 
 def set_request_root_span(span: Span) -> None:
     """Anchor the request's root (server) span for explicit child parenting.

litellm/proxy/auth/user_api_key_auth.py

@@ -1009,6 +1009,51 @@ async def _resolve_jwt_to_virtual_key(
     return None
 
 
+async def _hoist_request_destinations(request: Request, user_api_key_dict: UserAPIKeyAuth) -> None:
+    """Resolve admin-owned OTEL destinations for this request and anchor them.
+
+    Runs after the auth builder, while we are still inside the request task, so
+    the ``ContextVar`` is visible to every ``SpanProcessor.on_end`` that fires
+    for spans this request opens. Stashes the same list on ``request.state`` so
+    ``_apply_admin_logging_exporters`` can reuse it without a second DB pass.
+
+    Best-effort: a resolver failure must not break the request. The contextvar
+    is left at its default (empty tuple), so the fan-out processor no-ops. Idempotent:
+    it fires early in the builder and again as an outer catch-all; the second call
+    skips once ``request.state`` holds the result (a failed first call leaves it unset).
+    """
+    if getattr(getattr(request, "state", None), "otel_destinations", None) is not None:
+        return
+    try:
+        from litellm.integrations.otel.model.destination import OtelDestination
+        from litellm.integrations.otel.plumbing.context import (
+            set_request_destinations,
+        )
+        from litellm.proxy.litellm_pre_call_utils import (
+            _resolve_logging_exporters,
+        )
+
+        destinations_raw, _backends = await _resolve_logging_exporters(user_api_key_dict)
+        destinations = tuple(
+            OtelDestination(
+                callback_name=item.get("callback_name"),
+                endpoint=item.get("endpoint", ""),
+                headers=item.get("headers") or {},
+                resource_attributes=item.get("resource_attributes") or {},
+                protocol=item.get("protocol"),
+            )
+            for item in destinations_raw
+            if isinstance(item, dict) and item.get("endpoint")
+        )
+        set_request_destinations(destinations)
+        try:
+            request.state.otel_destinations = destinations_raw
+        except Exception:  # noqa: BLE001  # request.state mirror is best-effort; the ContextVar is the source of truth
+            pass
+    except Exception as exc:  # noqa: BLE001  # destination hoist is best-effort telemetry setup; it must never fail auth
+        verbose_proxy_logger.debug("OTel V2: hoist destination resolution failed: %s", exc)
+
+
 def _ensure_parent_otel_span_on_request_state(request: Request) -> None:
     """Idempotently create the OTEL SERVER span and stash it on
     ``request.state.parent_otel_span``. Safe to call multiple times.
@@ -1405,6 +1450,8 @@ async def _user_api_key_auth_builder(
                             valid_token = auto_registered
                             api_key = valid_token.token or ""
 
+                    await _hoist_request_destinations(request, valid_token)
+
                     # Check if model has zero cost - if so, skip all budget checks
                     model = _get_model_from_request_context(
                         request_data=request_data,
@@ -1742,6 +1789,8 @@ async def _user_api_key_auth_builder(
         user_obj: LiteLLM_UserTable | None = None
         valid_token_dict: dict = {}
         if valid_token is not None:
+            valid_token.parent_otel_span = parent_otel_span
+            await _hoist_request_destinations(request, valid_token)
             # Got Valid Token from Cache, DB
             # Run checks for
             # 1. If token can call model
@@ -2646,6 +2695,8 @@ async def user_api_key_auth(
             raise
         user_api_key_auth_obj.budget_reservation = None
 
+        await _hoist_request_destinations(request, user_api_key_auth_obj)
+
         # A body that never parsed is authenticated (so the trace carries identity
         # and this ``auth`` span) but not authorized: there is no model to check it
         # against, and budget reservation would increment live spend counters that

litellm/proxy/litellm_pre_call_utils.py

@@ -4,7 +4,7 @@ import json
 import re
 import time
 from collections import OrderedDict
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final
 
@@ -148,8 +148,11 @@ _ENABLE_TEAM_STALE_ALIAS_BYPASS: bool | None = None
 
 
 if TYPE_CHECKING:
+    from litellm.integrations.otel.model.destination import OtelDestination
+    from litellm.models.credentials import CredentialItem
     from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig
     from litellm.types.proxy.policy_engine import PolicyMatchContext
+    from litellm.types.utils import OtelDestinationParams
 
     ProxyConfig = _ProxyConfig
 else:
@@ -709,6 +712,160 @@ class KeyAndTeamLoggingSettings:
         return None
 
 
+async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None:
+    """The org this request belongs to, falling back to the team's org when the token
+    carries none. Team keys frequently have no ``org_id`` on the token, so without this
+    an org-scoped destination would be invisible at request time even though the write
+    gate (which loads the team) accepted it. Mirrors the fallback in ``_check_org_budget``.
+    """
+    if user_api_key_dict.org_id is not None:
+        return user_api_key_dict.org_id
+    team_id = user_api_key_dict.team_id
+    if team_id is None:
+        return None
+    from litellm.proxy import proxy_server
+    from litellm.proxy.auth.auth_checks import get_team_object
+
+    if proxy_server.prisma_client is None:
+        return None
+    try:
+        team_obj = await get_team_object(
+            team_id=team_id,
+            prisma_client=proxy_server.prisma_client,
+            user_api_key_cache=proxy_server.user_api_key_cache,
+            parent_otel_span=getattr(user_api_key_dict, "parent_otel_span", None),
+        )
+    except HTTPException:
+        return None
+    return getattr(team_obj, "organization_id", None)
+
+
+async def _resolve_logging_exporters(
+    user_api_key_dict: UserAPIKeyAuth,
+) -> "tuple[tuple[OtelDestinationParams, ...], tuple[str, ...]]":
+    """Resolve the destinations this request fans out to, as (destinations, backends).
+
+    ``credential_info.access`` is the sole routing determinant: a destination is
+    selected when its ``access`` grants the caller's team/org. Empty access grants no
+    one, so an empty-access destination never fires (proxy-wide requires
+    ``access.global``). Each survivor is built via ``destination_for_credential`` and deduped on
+    (endpoint, headers, resource attributes). Returns ([], []) when nothing is selected
+    (default-deny).
+    """
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.proxy.management_endpoints.logging_exporter_access import (
+        access_grants,
+        destination_for_credential,
+        identity_scope,
+        parse_credential_info,
+    )
+
+    # Admin-owned destinations are an OTEL v2 feature; the LITELLM_OTEL_V2 flag is the
+    # sole activation gate. With the flag off, registering a destination resolves to
+    # nothing (no backend is activated for the request) until the admin sets the flag.
+    if not is_otel_v2_enabled():
+        return (), ()
+
+    if not any(
+        (info := parse_credential_info(credential.credential_info)) is not None and info.credential_type == "logging"
+        for credential in litellm.credential_list
+    ):
+        return (), ()
+
+    team_id = user_api_key_dict.team_id
+    org_id = await _effective_org_id(user_api_key_dict)
+    team_ids, org_ids = identity_scope(team_id, org_id)
+
+    def _selected(credential: "CredentialItem") -> bool:
+        info = parse_credential_info(credential.credential_info)
+        if info is None or info.credential_type != "logging":
+            return False
+        return access_grants(info.access, team_ids, org_ids)
+
+    built = tuple(
+        result
+        for credential in litellm.credential_list
+        if _selected(credential)
+        if (result := destination_for_credential(credential)) is not None
+    )
+    deduped = {
+        (
+            destination.endpoint,
+            tuple(sorted(destination.headers.items())),
+            tuple(sorted(destination.resource_attributes.items())),
+        ): (
+            backend,
+            destination,
+        )
+        for backend, destination in built
+    }
+    destinations: tuple[OtelDestinationParams, ...] = tuple(
+        {
+            "callback_name": backend,
+            "endpoint": destination.endpoint,
+            "headers": destination.headers,
+            "resource_attributes": destination.resource_attributes,
+            "protocol": destination.protocol,
+        }
+        for backend, destination in deduped.values()
+    )
+    backends = tuple(dict.fromkeys(backend for backend, _ in deduped.values()))
+    return destinations, backends
+
+
+def _request_destination_from_raw(item: object) -> "OtelDestination | None":
+    from litellm.integrations.otel.model.destination import OtelDestination
+
+    if isinstance(item, OtelDestination):
+        return item
+    if not isinstance(item, dict) or not item.get("endpoint"):
+        return None
+    try:
+        return OtelDestination.model_validate(item)
+    except PydanticValidationError:
+        return None
+
+
+def _set_request_otel_destinations(destinations: Sequence[object]) -> None:
+    from litellm.integrations.otel.plumbing.context import set_request_destinations
+
+    set_request_destinations(
+        tuple(destination for item in destinations if (destination := _request_destination_from_raw(item)) is not None)
+    )
+
+
+async def _apply_admin_logging_exporters(
+    user_api_key_dict: UserAPIKeyAuth,
+    cached_destinations: "Sequence[object] | None" = None,
+) -> None:
+    """Anchor the resolved fan-out destinations on the request context.
+
+    The destinations are set on a server-only ContextVar (never on ``data``), so
+    they are neither request-shaped nor reachable by the provider body; the OTEL v2
+    router and the fan-out processor both read them from that ContextVar. Default-deny
+    means an identity no destination's access grants gets no per-tenant destination here.
+
+    ``cached_destinations`` -- when ``user_api_key_auth`` already resolved the
+    destinations on this request (the FastAPI path), reuse the result instead of
+    running the resolver a second time. The SDK path passes ``None`` and the
+    resolver runs here.
+
+    An empty resolution is published too, matching ``_hoist_request_destinations``.
+    Returning early instead would leave a previous message's destinations standing on
+    a ContextVar this request never overwrites: a stateful MCP session runs every
+    message on the task its ``initialize`` spawned, so a revoked grant would keep
+    exporting for the life of that session.
+    """
+    if cached_destinations is not None:
+        destinations = tuple(cached_destinations)
+    else:
+        try:
+            destinations, _backends = await _resolve_logging_exporters(user_api_key_dict)
+        except Exception:  # noqa: BLE001  # best-effort telemetry setup must never break the request
+            return
+    _set_request_otel_destinations(destinations)
+
+
 def _get_dynamic_logging_metadata(
     user_api_key_dict: UserAPIKeyAuth, proxy_config: ProxyConfig
 ) -> TeamCallbackMetadata | None:
@@ -2004,6 +2161,11 @@ async def add_litellm_data_to_request(
     )
 
     # Team Callbacks controls
+    data.pop("otel_destinations", None)
+    for _carrier_key in (_metadata_variable_name, "litellm_metadata"):
+        carrier = data.get(_carrier_key)
+        if isinstance(carrier, dict):
+            carrier.pop("otel_destinations", None)
     callback_settings_obj: Final = _get_dynamic_logging_metadata(
         user_api_key_dict=user_api_key_dict, proxy_config=proxy_config
     )
@@ -2012,13 +2174,15 @@ async def add_litellm_data_to_request(
         data["failure_callback"] = callback_settings_obj.failure_callback
 
         if callback_settings_obj.callback_vars is not None:
-            # unpack callback_vars in data
             for k, v in callback_settings_obj.callback_vars.items():
                 data[k] = v
             # Callbacks that must not honour request-supplied credentials read this
             # proxy-owned field instead of the raw request kwargs.
             data[TRUSTED_CALLBACK_VARS_FIELD] = callback_settings_obj.callback_vars
 
+    cached = getattr(getattr(request, "state", None), "otel_destinations", None)
+    await _apply_admin_logging_exporters(user_api_key_dict, cached_destinations=cached)
+
     # Add disabled callbacks from key metadata
     if user_api_key_dict.metadata and "litellm_disabled_callbacks" in user_api_key_dict.metadata:
         disabled_callbacks: Final = user_api_key_dict.metadata["litellm_disabled_callbacks"]

litellm/types/utils.py

@@ -3199,6 +3199,22 @@ OPENAI_RESPONSE_HEADERS: Final = [
 ]
 
 
+class OtelDestinationParams(TypedDict, total=False):
+    """A resolved, admin-owned OTLP destination carried server-side only.
+
+    Populated by the proxy from the exporters assigned to a request's identity
+    chain; never read from a request body or metadata. The v2 logger validates and
+    exports through it. ``callback_name`` is the OTEL backend this destination
+    belongs to, so fan-out routes each destination to the right backend's logger.
+    """
+
+    callback_name: str
+    endpoint: str
+    headers: Mapping[str, str]
+    resource_attributes: Mapping[str, str]
+    protocol: str | None
+
+
 class StandardCallbackDynamicParams(TypedDict, total=False):
     # Langfuse dynamic params
     langfuse_public_key: str | None

tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py

@@ -80,6 +80,59 @@ def test_resolves_plain_values_from_metadata():
     assert params.get("langfuse_host") == "https://test.langfuse.com"
 
 
+def test_otel_destinations_not_carried_on_dynamic_params():
+    """OTEL destination routing no longer travels through dynamic params or request
+    data at all (Y6): the admin-resolved destinations are anchored on a server-only
+    ContextVar and the v2 router reads them from there. Even a value under the internal
+    ``litellm_metadata`` key is NOT surfaced on the dynamic params, so the
+    request-carried carrier stays removed. Re-introducing the carrier fails this."""
+    destinations = [
+        {
+            "callback_name": "langfuse_otel",
+            "endpoint": "https://cloud.langfuse.com/api/public/otel",
+            "headers": {"Authorization": "Basic ADMIN"},
+        }
+    ]
+
+    params = initialize_standard_callback_dynamic_params(
+        {"litellm_metadata": {"otel_destinations": destinations}}
+    )
+
+    assert params.get("otel_destinations") is None
+
+
+def test_otel_destinations_top_level_kwarg_is_ignored():
+    """A top-level ``otel_destinations`` kwarg is intentionally NOT read. The proxy
+    stashes admin-resolved destinations under ``litellm_metadata`` to keep unknown
+    keys out of the body forwarded to the provider; reading the top-level key would
+    re-open that surface and is therefore ignored."""
+    params = initialize_standard_callback_dynamic_params(
+        {"otel_destinations": [{"callback_name": "langfuse_otel"}]}
+    )
+    assert params.get("otel_destinations") is None
+
+
+def test_otel_destinations_never_read_from_request_metadata():
+    """A request body/metadata must not be able to inject OTEL destinations:
+    otel_destinations is deliberately absent from the request-read whitelist, so a
+    value nested in metadata is ignored. Guards the trust boundary."""
+    kwargs = {
+        "metadata": {
+            "otel_destinations": [
+                {
+                    "callback_name": "langfuse_otel",
+                    "endpoint": "https://attacker.example/api/public/otel",
+                    "headers": {"Authorization": "Basic ATTACKER"},
+                }
+            ]
+        }
+    }
+
+    params = initialize_standard_callback_dynamic_params(kwargs)
+
+    assert params.get("otel_destinations") is None
+
+
 def test_litellm_params_metadata_overrides_metadata():
     kwargs = {
         "metadata": {

tests/test_litellm/proxy/auth/test_user_api_key_auth.py

@@ -5012,6 +5012,63 @@ async def test_builder_succeeds_when_db_lookup_returns_valid_token():
     mock_return.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_builder_hoists_destinations_before_post_lookup_auth_checks():
+    valid_token = UserAPIKeyAuth(api_key="sk-db-lookup-test", token="hashed-valid")
+    get_key_object = AsyncMock(return_value=valid_token)
+
+    async def _assert_hoisted_first(*args, **kwargs):
+        assert mock_hoist.await_count == 1
+
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth._return_user_api_key_auth_obj",
+            new_callable=AsyncMock,
+            return_value=valid_token,
+        ),
+        patch(
+            "litellm.proxy.auth.user_api_key_auth._hoist_request_destinations",
+            new_callable=AsyncMock,
+        ) as mock_hoist,
+        patch(
+            "litellm.proxy.auth.user_api_key_auth._enforce_key_and_fallback_model_access",
+            new_callable=AsyncMock,
+            side_effect=_assert_hoisted_first,
+        ) as mock_enforce,
+    ):
+        result = await _run_builder_with_key_lookup(get_key_object)
+
+    assert result is valid_token
+    mock_hoist.assert_awaited_once()
+    mock_enforce.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_hoist_destinations_resolver_failure_never_breaks_auth():
+    """Destination resolution is best-effort telemetry setup: if the resolver raises,
+    _hoist_request_destinations must swallow it and leave the ContextVar at its empty
+    default so auth proceeds and the fan-out processor no-ops. A raise here would take
+    down every request."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from litellm.integrations.otel.plumbing.context import request_destinations
+    from litellm.proxy.auth.user_api_key_auth import _hoist_request_destinations
+
+    request = MagicMock()
+    request.state = MagicMock()
+    valid_token = UserAPIKeyAuth(api_key="sk-x", token="hashed")
+
+    with patch(
+        "litellm.proxy.litellm_pre_call_utils._resolve_logging_exporters",
+        new_callable=AsyncMock,
+        side_effect=RuntimeError("resolver blew up"),
+    ):
+        # must not raise
+        await _hoist_request_destinations(request, valid_token)
+
+    assert request_destinations() == ()
+
+
 def _mint_cli_session_token(monkeypatch, *, user_id="cli-admin"):
     """Mint a CLI session token for a PROXY_ADMIN user so auth resolves on the
     admin early-return path (no prisma/common_checks needed)."""
@@ -5967,3 +6024,26 @@ async def test_unlicensed_jwt_auth_is_forbidden_not_unauthorized():
 
     assert error.code == "403"
     assert "enterprise" in error.message.lower()
+@pytest.mark.asyncio
+async def test_hoist_request_destinations_idempotent(monkeypatch):
+    """The hoist fires early in the auth builder and again as an outer catch-all; the
+    second call must not re-run the resolver once request.state holds the result."""
+    import litellm.proxy.litellm_pre_call_utils as pcu
+    from litellm.proxy.auth.user_api_key_auth import _hoist_request_destinations
+
+    calls = {"n": 0}
+
+    async def fake_resolve(_uapk):
+        calls["n"] += 1
+        return ((), ())
+
+    monkeypatch.setattr(pcu, "_resolve_logging_exporters", fake_resolve)
+    request = MagicMock()
+    request.state = SimpleNamespace()
+    uapk = UserAPIKeyAuth(api_key="x", token="x")
+
+    await _hoist_request_destinations(request, uapk)
+    await _hoist_request_destinations(request, uapk)
+
+    assert calls["n"] == 1
+    assert request.state.otel_destinations == ()