diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index a76ee61ef9c..7e3ba0bc42e 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -63,6 +63,7 @@ if TYPE_CHECKING: from litellm.integrations.otel.model.destination import OtelDestination from litellm.proxy._types import UserAPIKeyAuth from litellm.types.utils import ( + StandardCallbackDynamicParams, StandardLoggingGuardrailInformation, StandardLoggingPayload, ) @@ -255,7 +256,9 @@ class OpenTelemetryV2(CustomLogger): start_time_ns=start_time_ns, tracer=tracer, ) - for tracer in self._tenant_tracers.tracers_for(self.tracer, self._destinations_for_backend(call)) + for tracer in self._tenant_tracers.genai_tracers_for( + self.tracer, self._destinations_for_backend(call), call.dynamic_params + ) ) self._open_llm_calls[call_id] = _LLMCallSpan(spans=spans, start_time_ns=start_time_ns) # Evict the oldest open call if over budget; a call that opens but never closes would @@ -411,6 +414,7 @@ class OpenTelemetryV2(CustomLogger): to_ns(start_time), to_ns(end_time), call.time_to_first_chunk_seconds, + call.dynamic_params, ) end_time_ns = to_ns(end_time) @@ -435,6 +439,7 @@ class OpenTelemetryV2(CustomLogger): carrier.start_time_ns, end_time_ns, call.time_to_first_chunk_seconds, + call.dynamic_params, ) def _mark_closed(self, call_id: str | None) -> None: @@ -452,6 +457,7 @@ class OpenTelemetryV2(CustomLogger): start_time_ns: int | None, end_time_ns: int | None, time_to_first_chunk_seconds: float | None = None, + dynamic_params: "StandardCallbackDynamicParams | None" = None, ) -> Span | None: """Emit an LLM-call span outside the ``pre_call`` boundary. @@ -470,7 +476,7 @@ class OpenTelemetryV2(CustomLogger): parent_context=parent_ctx, start_time_ns=start_time_ns, end_time_ns=end_time_ns, - tracers=self._tenant_tracers.tracers_for(self.tracer, destinations), + tracers=self._tenant_tracers.genai_tracers_for(self.tracer, destinations, dynamic_params), ) # ====================================================================== # diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index 4c34acf91a5..6f7bde2c832 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -28,7 +28,7 @@ from litellm.integrations.otel.model.utils import as_str, to_seconds from litellm.integrations.otel.plumbing.context import request_destinations if TYPE_CHECKING: - from litellm.types.utils import StandardLoggingPayload + from litellm.types.utils import StandardCallbackDynamicParams, StandardLoggingPayload @dataclass(frozen=True) @@ -141,6 +141,9 @@ class LLMCallEvent: # The success/failure payload; ``None`` at ``pre_call`` or if the call closed with no payload. payload: "StandardLoggingPayload | None" otel_destinations: tuple[OtelDestination, ...] + # The request's ``standard_callback_dynamic_params`` (team/key OTLP credentials), or ``None`` + # when the call isn't scoped; routes the gen-AI span to a credential-scoped tracer. + dynamic_params: "StandardCallbackDynamicParams | None" # True for synthetic proxy-gate logs (auth/rate-limit rejections): no upstream call, so no span. is_no_upstream_call: bool # Best-effort ``"{operation} {model}"`` name at ``pre_call``; only matters for a leaked span (renamed at close). @@ -157,6 +160,7 @@ class LLMCallEvent: call_id=_call_id(payload, kwargs), payload=payload, otel_destinations=request_destinations(), + dynamic_params=kwargs.get("standard_callback_dynamic_params"), is_no_upstream_call=bool(kwargs.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL)), provisional_span_name=f"{operation.value} {model}".strip(), time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs), diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 63057bf6876..37cbea69145 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -3,9 +3,16 @@ ``TenantTracerCache`` routes the gen-AI LLM-call span, building per-tenant clone ``TracerProvider``s that export to the request's admin-owned destinations plus the configured/global exporter. ``TenantFanOutSpanProcessor`` (at the bottom) forwards the -proxy-internal spans (server, auth, DB, cost) to every destination. Both read the request's -destinations from the same server-only contextvar, so a caller can neither redirect a trace -nor spawn providers. +proxy-internal spans (server, auth, DB, cost) to every destination. The destinations both +read from a server-only contextvar, so a caller can neither redirect a trace nor spawn +providers through them. + +Separately, ``genai_tracers_for`` also routes the gen-AI span by the request's +``standard_callback_dynamic_params`` (team/key OTLP credentials), restoring the per-request +credential routing that predates the admin-destination refactor: it rewrites only the owned +exporter's headers, bounded by the same LRU. Those params are partly caller-influenced, so +this path can spawn a per-request provider; closing that for request-body-supplied credentials +(vs admin-configured team ``callback_vars``) is tracked separately. The gen-AI path (``tracers_for``) groups destinations by their backend-required Resource attributes and builds one provider per group, because a span carries exactly one Resource and @@ -15,6 +22,7 @@ tagged span instead of a last-wins merge. Empty destinations -> the logger's def import threading from collections import OrderedDict +from typing import TYPE_CHECKING from opentelemetry.context import Context from opentelemetry.sdk.resources import Resource @@ -29,6 +37,10 @@ from litellm.integrations.otel.plumbing.providers import ( build_tracer_provider, get_tracer, ) +from litellm.integrations.otel.presets import dynamic_otlp_headers + +if TYPE_CHECKING: + from litellm.types.utils import StandardCallbackDynamicParams _NON_OTLP_KINDS = ("console", "in_memory", "inmemory", "memory") @@ -77,21 +89,96 @@ class TenantTracerCache: _, evicted = self._providers.popitem(last=False) _shutdown_in_background(evicted) - def tracers_for(self, default: Tracer, destinations: "tuple[OtelDestination, ...]") -> "tuple[Tracer, ...]": + def tracers_for( + self, + default: Tracer, + destinations: "tuple[OtelDestination, ...]", + *, + include_base_on_first: bool = True, + ) -> "tuple[Tracer, ...]": """The tracers for this request's gen-AI span, one per distinct Resource group. A backend like Arize selects its project from the Resource, so destinations are grouped by ``destination_resource_attrs`` and the caller emits the span once per tracer. The configured/global exporters ride the FIRST group only, so the global receives the span once. Empty ``destinations`` -> the logger's default tracer (deny). + + ``include_base_on_first`` is set to ``False`` by ``genai_tracers_for`` when a + credential-scoped tracer already carries the configured/global exporters, so the + destination groups don't also export the span to the global collector a second time. """ if not destinations: return (default,) return tuple( - self._tracer_for_group(resource_key, group, include_base=index == 0) + self._tracer_for_group(resource_key, group, include_base=include_base_on_first and index == 0) for index, (resource_key, group) in enumerate(self._group_by_resource(destinations)) ) + def genai_tracers_for( + self, + default: Tracer, + destinations: "tuple[OtelDestination, ...]", + dynamic_params: "StandardCallbackDynamicParams | None", + ) -> "tuple[Tracer, ...]": + """The gen-AI span's tracers, layering per-request credential routing over the + admin-destination fan-out. + + When the request carries this backend's team/key OTLP credentials + (``standard_callback_dynamic_params``), the global export rides a credential-scoped + provider (the configured exporters with this backend's own exporter rewritten to those + credentials), and the admin-destination groups omit the base exporters so the span + reaches the global collector exactly once. Without dynamic credentials this is the plain + destination fan-out (empty destinations -> the default tracer). + """ + headers = dynamic_otlp_headers(self._callback_name, dynamic_params) + if not headers: + return self.tracers_for(default, destinations) + dynamic = self._credential_scoped_tracer(headers) + if not destinations: + return (dynamic,) + return (dynamic, *self.tracers_for(default, destinations, include_base_on_first=False)) + + def dynamic_tracer_for(self, default: Tracer, dynamic_params: "StandardCallbackDynamicParams | None") -> Tracer: + """The credential-scoped tracer when the request carries this backend's team/key OTLP + credentials, else ``default``. Distinct from ``tracer_for`` (admin destinations); this + is the per-request path restored for parity with the pre-v2-refactor behavior.""" + headers = dynamic_otlp_headers(self._callback_name, dynamic_params) + if not headers: + return default + return self._credential_scoped_tracer(headers) + + def _credential_scoped_tracer(self, headers: "dict[str, str]") -> Tracer: + """A cached provider that keeps the configured exporters and rewrites only this + backend's owned exporter's headers to ``headers`` (the per-request credentials).""" + cache_key: tuple[object, ...] = ("dynamic", tuple(sorted(headers.items()))) + provider = self._providers.get(cache_key) + if provider is not None: + self._providers.move_to_end(cache_key) + else: + provider = build_tracer_provider(self._config_with_headers(headers)) + self._providers[cache_key] = provider + self._evict_if_full() + return get_tracer(provider, self._tracer_name) + + def _config_with_headers(self, headers: "dict[str, str]") -> OpenTelemetryV2Config: + """Clone the config, stamping ``headers`` onto this backend's own exporter only. + + ``headers`` are the per-request credentials of ``self._callback_name``, so they apply + only to the exporter that integration contributed (``spec.owner``). A request carrying + one tenant's Arize key must never rewrite a co-configured Langfuse or self-hosted + collector exporter, which would leak that key to a different backend. + """ + header_str = ",".join(f"{key}={value}" for key, value in headers.items()) + exporters = [ + ( + spec.model_copy(update={"headers": header_str}) + if spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS + else spec + ) + for spec in self._config.exporters + ] + return self._config.model_copy(update={"exporters": exporters}) + def _group_by_resource( self, destinations: "tuple[OtelDestination, ...]" ) -> "tuple[tuple[tuple[tuple[str, str], ...], tuple[OtelDestination, ...]], ...]": diff --git a/litellm/integrations/otel/presets/__init__.py b/litellm/integrations/otel/presets/__init__.py index 635e0480b77..ed41babe464 100644 --- a/litellm/integrations/otel/presets/__init__.py +++ b/litellm/integrations/otel/presets/__init__.py @@ -7,22 +7,55 @@ maps a callback name (``"arize"``, ``"langfuse_otel"``, ...) to its preset so the factory in ``litellm_logging`` can resolve a name and build a single ``OpenTelemetryV2`` instance from the result. -Per-key/team routing does not live here. A trace destination is admin-owned -infrastructure config, resolved server-side from a named credential into an -``OtelDestination`` (see ``litellm.integrations.otel.presets.destinations`` and -``plumbing.routing``); nothing in this package reads vendor credentials or a -host off a request. +Admin-owned trace destinations are resolved server-side from a named credential +into an ``OtelDestination`` (see ``litellm.integrations.otel.presets.destinations`` +and ``plumbing.routing``). Separately, per-request team/key OTLP credentials from +``standard_callback_dynamic_params`` route the gen-AI span to a credential-scoped +tracer for the integrations that support it; ``dynamic_otlp_headers`` below builds +those per-request headers. """ +from typing import Callable + from litellm.integrations.otel.presets.agentops import agentops_preset -from litellm.integrations.otel.presets.arize import arize_preset +from litellm.integrations.otel.presets.arize import arize_dynamic_headers, arize_preset from litellm.integrations.otel.presets.base import Preset from litellm.integrations.otel.presets.generic import generic_preset -from litellm.integrations.otel.presets.langfuse import langfuse_preset +from litellm.integrations.otel.presets.langfuse import ( + langfuse_dynamic_headers, + langfuse_preset, +) from litellm.integrations.otel.presets.langtrace import langtrace_preset from litellm.integrations.otel.presets.levo import levo_preset from litellm.integrations.otel.presets.phoenix import phoenix_preset -from litellm.integrations.otel.presets.weave import weave_preset +from litellm.integrations.otel.presets.weave import weave_dynamic_headers, weave_preset +from litellm.types.utils import StandardCallbackDynamicParams + +#: Callback name -> per-request OTLP header builder (team/key multi-tenant +#: routing). Only integrations that support dynamic credentials appear here; +#: Arize-Phoenix/Langtrace/Levo/AgentOps/generic don't, so they use the logger's +#: default tracer. +DYNAMIC_HEADERS_BY_CALLBACK: dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]] = { + "arize": arize_dynamic_headers, + "langfuse_otel": langfuse_dynamic_headers, + "weave_otel": weave_dynamic_headers, +} + + +def dynamic_otlp_headers( + callback_name: str | None, + dynamic_params: "StandardCallbackDynamicParams | None", +) -> dict[str, str] | None: + """Per-request OTLP headers for ``callback_name``, or ``None`` if N/A. + + ``None`` means "no per-request routing" -- the caller uses its default tracer. + """ + builder = DYNAMIC_HEADERS_BY_CALLBACK.get(callback_name or "") + if builder is None or not dynamic_params: + return None + headers = builder(dynamic_params) + return headers or None + #: Callback name → preset. The ``Preset`` annotation makes mypy verify every #: registered value matches the preset interface. @@ -39,10 +72,12 @@ PRESET_BY_CALLBACK: dict[str, Preset] = { __all__ = [ + "DYNAMIC_HEADERS_BY_CALLBACK", "PRESET_BY_CALLBACK", "Preset", "agentops_preset", "arize_preset", + "dynamic_otlp_headers", "generic_preset", "langfuse_preset", "langtrace_preset", diff --git a/litellm/integrations/otel/presets/arize.py b/litellm/integrations/otel/presets/arize.py index 622ba2f90da..a4b0d7d5bf0 100644 --- a/litellm/integrations/otel/presets/arize.py +++ b/litellm/integrations/otel/presets/arize.py @@ -10,6 +10,7 @@ from litellm.integrations.otel.model.config import ( OpenTelemetryV2Config, ) from litellm.integrations.otel.presets.utils import ensure_mappers +from litellm.types.utils import StandardCallbackDynamicParams class _ArizeSettings(BaseSettings): @@ -59,3 +60,16 @@ def _arize_headers(arize_cfg) -> str | None: # credentials are configured. return _ArizeSettings().otlp_traces_headers return ",".join(pieces) + + +def arize_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]: + """Per-request Arize OTLP headers from team/key dynamic params.""" + headers: dict[str, str] = {} + # ``arize_space_key`` is the suggested param and wins over ``arize_space_id``. + space = params.get("arize_space_key") or params.get("arize_space_id") + if space: + headers["arize-space-id"] = space + api_key = params.get("arize_api_key") + if api_key: + headers["api_key"] = api_key + return headers diff --git a/litellm/integrations/otel/presets/langfuse.py b/litellm/integrations/otel/presets/langfuse.py index 9fd27f0187e..6c128fc658f 100644 --- a/litellm/integrations/otel/presets/langfuse.py +++ b/litellm/integrations/otel/presets/langfuse.py @@ -9,6 +9,20 @@ from litellm.integrations.otel.model.config import ( OpenTelemetryV2Config, ) from litellm.integrations.otel.presets.utils import ensure_mappers +from litellm.types.utils import StandardCallbackDynamicParams + + +def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]: + """Per-request Langfuse OTLP headers from team/key dynamic params.""" + public_key = params.get("langfuse_public_key") + secret_key = params.get("langfuse_secret_key") + if public_key and secret_key: + return { + "Authorization": _V1Langfuse._get_langfuse_authorization_header( + public_key=public_key, secret_key=secret_key + ) + } + return {} def langfuse_preset( diff --git a/litellm/integrations/otel/presets/weave.py b/litellm/integrations/otel/presets/weave.py index 1be6603dae6..587994a7c2c 100644 --- a/litellm/integrations/otel/presets/weave.py +++ b/litellm/integrations/otel/presets/weave.py @@ -6,7 +6,23 @@ from litellm.integrations.otel.model.config import ( OpenTelemetryV2Config, ) from litellm.integrations.otel.presets.utils import ensure_mappers -from litellm.integrations.weave.weave_otel import get_weave_otel_config +from litellm.integrations.weave.weave_otel import ( + _get_weave_authorization_header, + get_weave_otel_config, +) +from litellm.types.utils import StandardCallbackDynamicParams + + +def weave_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]: + """Per-request Weave OTLP headers from team/key dynamic params.""" + headers: dict[str, str] = {} + api_key = params.get("wandb_api_key") + if api_key: + headers["Authorization"] = _get_weave_authorization_header(api_key=api_key) + project_id = params.get("weave_project_id") + if project_id: + headers["project_id"] = project_id + return headers def weave_preset( diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py index f00a11266c1..986725de9ad 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py @@ -1,11 +1,13 @@ -"""Per-tenant tracer routing on admin-owned OTEL destinations, with fan-out. +"""Per-tenant tracer routing on admin-owned OTEL destinations, with fan-out, plus the +per-request credential routing layered over it. -A request's identity chain is assigned a set of admin-owned exporters; the v2 logger -fans the trace out to all of them (plus the configured/global exporter), and never -routes on request-supplied vendor credentials. These tests lock the contract: the -request cannot route a trace, each destination's endpoint follows its resolved host -(cross-host fix), the configured exporters are kept (global also receives), and a -logger only exports the destinations tagged with its own backend. +A request's identity chain is assigned a set of admin-owned exporters; the v2 logger fans +the trace out to all of them (plus the configured/global exporter). Separately, a request's +``standard_callback_dynamic_params`` (team/key OTLP credentials) route the gen-AI span to a +credential-scoped tracer for the integrations that support it (V1 parity). These tests lock +both: each destination's endpoint follows its resolved host (cross-host fix), the configured +exporters are kept (global also receives), a logger only exports the destinations tagged with +its own backend, and request credentials rewrite only their own backend's exporter headers. """ import os @@ -319,38 +321,82 @@ def test_clone_provider_emits_genai_span_with_destination_resource(): assert resource_attrs.get("arize.project.name") == "team-b-proj" -# --- security: request credentials never route a trace --------------------- # +# --- request credentials route the gen-AI span (V1 parity) ------------------ # -@pytest.mark.parametrize( - "request_creds", - [ - { - "langfuse_public_key": "pk-attacker", - "langfuse_secret_key": "sk-attacker", - "langfuse_host": "https://attacker.example", - }, - {"arize_api_key": "K-attacker", "arize_space_id": "S-attacker"}, - {"wandb_api_key": "w-attacker", "weave_endpoint": "https://attacker/otel"}, - ], -) -def test_request_credentials_are_inert_on_v2(request_creds): - """Any backend's credentials in the request's dynamic params (no admin - destinations) produce no per-tenant routing.""" - event = LLMCallEvent.from_dict( - { - "standard_callback_dynamic_params": request_creds, - "call_type": "acompletion", - "model": "gpt-4o", - } +_DYNAMIC_CREDS = [ + ("langfuse_otel", {"langfuse_public_key": "pk-teamA", "langfuse_secret_key": "sk-teamA"}, "Authorization="), + ("arize", {"arize_space_id": "space-teamA", "arize_api_key": "key-teamA"}, "arize-space-id=space-teamA"), + ("weave_otel", {"wandb_api_key": "wandb-teamA", "weave_project_id": "proj-teamA"}, "project_id=proj-teamA"), +] + + +@pytest.mark.parametrize("backend, creds, owned_fragment", _DYNAMIC_CREDS) +def test_request_credentials_rewrite_only_the_owned_exporter(backend, creds, owned_fragment): + """A request's team/key OTLP credentials rewrite the headers of THIS backend's own + exporter and nothing else, so the gen-AI span exports to that team's account while a + co-configured backend's exporter is untouched (no cross-backend key leak).""" + from litellm.integrations.otel.presets import dynamic_otlp_headers + + cache = _cache( + backend, + exporters=[ + ExporterSpec(kind="otlp_http", endpoint="http://collector/otel", headers="Authorization=Basic GLOBAL", owner=backend), + ExporterSpec(kind="otlp_http", endpoint="http://other/otel", headers="x=coconfigured", owner="levo"), + ], + ) + headers = dynamic_otlp_headers(backend, creds) + assert headers, f"{backend} must support dynamic credentials" + scoped = cache._config_with_headers(headers) + assert owned_fragment in (scoped.exporters[0].headers or "") + assert "GLOBAL" not in (scoped.exporters[0].headers or "") # the global header was rewritten away + assert scoped.exporters[1].headers == "x=coconfigured" # a different backend's exporter is untouched + + +@pytest.mark.parametrize("backend, creds, owned_fragment", _DYNAMIC_CREDS) +def test_genai_tracers_for_spawns_credential_scoped_provider(backend, creds, owned_fragment): + """genai_tracers_for with request creds and no admin destination returns exactly the + credential-scoped tracer (not the default) and caches its provider under a dynamic key.""" + cache = _cache( + backend, + exporters=[ExporterSpec(kind="otlp_http", endpoint="http://c/otel", headers="Authorization=Basic GLOBAL", owner=backend)], ) - assert event.otel_destinations == () - cache = _cache("langfuse_otel") default = NoOpTracer() - assert cache.tracer_for(default, event.otel_destinations) is default + tracers = cache.genai_tracers_for(default, (), creds) + assert len(tracers) == 1 + assert tracers[0] is not default + assert len(cache._providers) == 1 + assert next(iter(cache._providers))[0] == "dynamic" # namespaced key, never aliases a destination group + + +def test_genai_tracers_for_without_creds_is_plain_default(): + """No dynamic creds and no destinations -> the logger's default tracer, no provider spawned.""" + cache = _cache( + "langfuse_otel", + exporters=[ExporterSpec(kind="otlp_http", endpoint="http://c/otel", headers="", owner="langfuse_otel")], + ) + default = NoOpTracer() + assert cache.genai_tracers_for(default, (), None) == (default,) assert cache._providers == {} +def test_genai_tracers_compose_credential_scoped_plus_destination(): + """Request creds AND an admin destination: the span exports via the credential-scoped + tracer (which carries the global exporter) plus one per-destination tracer that omits the + base exporters, so the global collector receives the span exactly once.""" + cache = _cache( + "langfuse_otel", + exporters=[ExporterSpec(kind="otlp_http", endpoint="http://global/otel", headers="Authorization=Basic GLOBAL", owner="langfuse_otel")], + ) + creds = {"langfuse_public_key": "pk-teamA", "langfuse_secret_key": "sk-teamA"} + destinations = (_dest("https://cloud.langfuse.com/api/public/otel", auth="Basic ADMIN"),) + tracers = cache.genai_tracers_for(NoOpTracer(), destinations, creds) + assert len(tracers) == 2 # credential-scoped (global) + one destination group + # the destination group provider carries no base exporter (base rides the dynamic tracer) + dest_only = cache._config_with_destinations(destinations, include_base_exporters=False) + assert all(spec.endpoint != "http://global/otel" for spec in dest_only.exporters) + + def test_admin_destinations_route(): event = _event( [