From bcba86e426fedea716c703d5c1fa4c7a65e479b7 Mon Sep 17 00:00:00 2001 From: Yucheng He Date: Thu, 3 Sep 2026 14:42:07 -0700 Subject: [PATCH] feat(otel v2): send a key's or team's whole trace to its own destination A key or team that configures its own Langfuse, Arize, Weave or New Relic credentials used to get a single detached span in its account while the rest of the request trace stayed on the operator's backend, so neither side held a complete trace. Resolve the destination during auth, forward every span of the request to it, and hold the same request back from the operator's exporter for that backend, so the tenant gets the tree the operator would have seen and the operator gets nothing for that request. Also let a credential-mandatory preset build without the operator's own env credentials. Without that, a proxy whose teams each bring their own account fell back to the legacy integration and never ran a line of the v2 path. --- litellm/integrations/otel/logger.py | 4 +- .../integrations/otel/model/destination.py | 55 +++ litellm/integrations/otel/plumbing/context.py | 36 +- .../integrations/otel/plumbing/providers.py | 203 +++++++++- litellm/integrations/otel/plumbing/routing.py | 10 +- litellm/integrations/otel/presets/agentops.py | 1 + litellm/integrations/otel/presets/arize.py | 23 +- litellm/integrations/otel/presets/base.py | 14 +- .../integrations/otel/presets/destinations.py | 118 ++++++ litellm/integrations/otel/presets/langfuse.py | 23 +- .../integrations/otel/presets/langtrace.py | 1 + litellm/integrations/otel/presets/levo.py | 1 + litellm/integrations/otel/presets/newrelic.py | 1 + litellm/integrations/otel/presets/phoenix.py | 1 + litellm/integrations/otel/presets/utils.py | 23 ++ litellm/integrations/otel/presets/weave.py | 21 +- litellm/litellm_core_utils/litellm_logging.py | 7 +- litellm/proxy/auth/user_api_key_auth.py | 23 ++ litellm/proxy/litellm_pre_call_utils.py | 48 +++ .../otel/test_otel_v2_destinations.py | 360 ++++++++++++++++++ 20 files changed, 952 insertions(+), 21 deletions(-) create mode 100644 litellm/integrations/otel/model/destination.py create mode 100644 litellm/integrations/otel/presets/destinations.py create mode 100644 tests/test_litellm/integrations/otel/test_otel_v2_destinations.py diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 5519896a961..ec3c3eaa004 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -180,7 +180,9 @@ class OpenTelemetryV2(CustomLogger): self.config: OpenTelemetryV2Config = config or OpenTelemetryV2Config(**kwargs) self.callback_name = callback_name self._tracer_provider: TracerProvider = ( - tracer_provider if tracer_provider is not None else build_tracer_provider(self.config) + tracer_provider + if tracer_provider is not None + else build_tracer_provider(self.config, tenant_overrides=True) ) self.tracer: Tracer = get_tracer(self._tracer_provider, LITELLM_TRACER_NAME) self._metrics_recorder = self._init_metrics(meter_provider) diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py new file mode 100644 index 00000000000..b88a240c966 --- /dev/null +++ b/litellm/integrations/otel/model/destination.py @@ -0,0 +1,55 @@ +"""The resolved OTLP destination a request's traces export to. + +A destination is a backend-agnostic target: an endpoint plus the auth headers the +exporter sends. The proxy builds one per backend from the key or team logging +config resolved at auth, and the fan-out span processor exports the request's +spans through it. Every OTEL backend reduces to this shape; the per-backend field +mapping lives in ``litellm.integrations.otel.presets.destinations``. +""" + +from collections.abc import Mapping +from typing import Final +from urllib.parse import quote + +from pydantic import BaseModel, ConfigDict, Field + + +class OtelDestination(BaseModel): + model_config = ConfigDict(frozen=True) + + endpoint: str + headers: Mapping[str, str] = Field(default_factory=dict) + resource_attributes: Mapping[str, str] = Field(default_factory=dict) + callback_name: str | None = None + protocol: str | None = Field( + default=None, + description=( + "OTLP transport for this endpoint (``otlp_http`` / ``otlp_grpc``). The " + "backend's intrinsic default is used when unset. A backend whose own cloud " + "endpoint is gRPC can still be pointed at an HTTP collector, which the " + "scheme alone cannot express: Arize's own ``https://otlp.arize.com/v1`` is gRPC." + ), + ) + + def header_string(self) -> str: + """Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects. + + Values are percent-encoded because ``providers.parse_headers`` decodes them + with the SDK's W3C-Baggage parser: a value carrying a ``,`` or ``=`` (a + Langfuse project name, a base64 Authorization payload ending in ``==``) + would otherwise be split into bogus pairs on the way back out. + """ + return ",".join(f"{key}={quote(value, safe='')}" for key, value in self.headers.items()) + + def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]: + """Identity for processor reuse: two requests naming the same destination + must share one exporter rather than minting a connection pool each.""" + return ( + self.endpoint, + tuple(sorted(self.headers.items())), + tuple(sorted(self.resource_attributes.items())), + self.protocol, + ) + + +NO_DESTINATIONS: Final[tuple[OtelDestination, ...]] = () diff --git a/litellm/integrations/otel/plumbing/context.py b/litellm/integrations/otel/plumbing/context.py index aa7cc8e2afd..3d8d8b53388 100644 --- a/litellm/integrations/otel/plumbing/context.py +++ b/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 @@ -21,6 +21,9 @@ from opentelemetry.trace.propagation.tracecontext import ( from litellm.integrations.otel.model.semconv import HTTP +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 @@ -304,3 +307,34 @@ def extract_traceparent(headers: Mapping[str, str]) -> Context | None: return None carrier: Final = {str(key).lower(): value for key, value in headers.items()} return _PROPAGATOR.extract(carrier) + + +# The OTLP destinations this request's key or team pointed its traces at, resolved +# once during auth. A ``ContextVar`` for the same reason the root span above is one: +# it rides the request task's context into the ``asyncio.create_task`` children that +# close the LLM span, and it is visible to every ``SpanProcessor.on_end`` that fires +# on the request task. Never reset -- it dies with the task. +_request_destinations: Final['ContextVar[tuple["OtelDestination", ...]]'] = ContextVar( + "litellm_otel_request_destinations", default=() +) + + +def set_request_destinations(destinations: 'tuple["OtelDestination", ...]') -> None: + """Anchor the destinations this request exports to.""" + _request_destinations.set(destinations) + + +def request_destinations() -> 'tuple["OtelDestination", ...]': + """The destinations resolved for this request, empty outside a proxy request.""" + return _request_destinations.get() + + +def overridden_backends() -> frozenset[str]: + """Backends whose global exporters this request must NOT reach. + + A team destination is an override, not an addition: once the request resolved a + destination for a backend, that backend's operator-level exporters are suppressed + for every span of the request, so the tenant's traffic reaches the tenant's + account and nowhere else. + """ + return frozenset(d.callback_name for d in _request_destinations.get() if d.callback_name) diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index fb74ff85e5b..00198234f54 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -1,5 +1,7 @@ """Provider / exporter factory + the Baggage span processor.""" +import threading +from collections import OrderedDict from collections.abc import Callable, Iterable from typing import TYPE_CHECKING, Any, Final, Literal @@ -20,6 +22,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 ReadableSpan, SpanProcessor, TracerProvider +from opentelemetry.sdk.trace import Span as SDKSpan from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, ConsoleSpanExporter, @@ -32,15 +35,22 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( from opentelemetry.trace import Span, SpanKind, Tracer from opentelemetry.util.re import parse_env_headers +from litellm._logging import verbose_logger from litellm._version import version as litellm_version from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config from litellm.integrations.otel.model.semconv import LiteLLM from litellm.integrations.otel.model.spans import LiteLLMSpanKind +from litellm.integrations.otel.plumbing.context import ( + overridden_backends, + request_destinations, +) if TYPE_CHECKING: from opentelemetry.metrics import Meter from opentelemetry.sdk.metrics.export import MetricReader + from litellm.integrations.otel.model.destination import OtelDestination + _SPAN_KIND_BY_ROLE_KIND: Final[dict[LiteLLMSpanKind, SpanKind]] = { LiteLLMSpanKind.SERVER: SpanKind.SERVER, LiteLLMSpanKind.CLIENT: SpanKind.CLIENT, @@ -194,6 +204,178 @@ def _processor_for(exporter: SpanExporter, use_simple: bool | None) -> SpanProce return SimpleSpanProcessor(exporter) if use_simple else BatchSpanProcessor(exporter) +#: Distinct tenant destinations whose exporters stay alive. Each holds a connection +#: pool and a batch thread, so the cache is bounded and evicts least-recently-used. +_MAX_CACHED_DESTINATION_PROCESSORS: Final = 32 + + +class _ResourceWrappedReadableSpan(ReadableSpan): + """A ``ReadableSpan`` view with an overridden Resource, leaving the original alone.""" + + def __init__(self, inner: ReadableSpan, resource: Resource) -> None: + super().__init__( + name=inner.name, + context=inner.context, + parent=inner.parent, + resource=resource, + attributes=inner.attributes, + events=inner.events, + links=inner.links, + kind=inner.kind, + status=inner.status, + start_time=inner.start_time, + end_time=inner.end_time, + instrumentation_scope=inner.instrumentation_scope, + ) + + +def _with_destination_resource(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan: + extra: Final = destination.resource_attributes + if not extra: + return span + merged: Final = Resource.create( + {**dict(span.resource.attributes), **dict(extra)} # mutable-ok: the OTel SDK takes a concrete attribute mapping + ) + return _ResourceWrappedReadableSpan(span, merged) + + +class TenantFanOutSpanProcessor(SpanProcessor): + """Export every finished span to each destination this request resolved. + + Destinations ride a request-scoped ``ContextVar`` set during auth, so the + processor keeps no per-request state and concurrent requests stay isolated. + Every span is forwarded, the gen-AI span included: the tenant's account gets the + tree the operator's would have received, still parented, because the forwarded + view keeps the original span's trace and parent ids. + """ + + def __init__( + self, + processor_factory: 'Callable[["OtelDestination"], SpanProcessor | None] | None' = None, + ) -> None: + self._lock: Final = threading.Lock() + self._build: Final = processor_factory if processor_factory is not None else _destination_processor + self._processors: OrderedDict[object, SpanProcessor] = OrderedDict() # mutable-ok: bounded LRU + + def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: + return None + + def on_end(self, span: ReadableSpan) -> None: + for destination in request_destinations(): + processor = self._processor_for(destination) # rebind-ok: loop variable; pyright forbids Final in a loop + if processor is None: + continue + try: + processor.on_end(_with_destination_resource(span, destination)) + 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) + + def shutdown(self) -> None: + # Snapshot first: ``on_end`` mutates the cache on whichever thread ends a span + # and can run concurrently with this SDK-driven shutdown, so iterating the live + # mapping risks a "mutated during iteration" the per-item except cannot catch. + for processor in self._snapshot(): + try: + processor.shutdown() + except Exception as exc: # noqa: BLE001 # one processor's shutdown must not abort the rest + verbose_logger.debug("OTel V2 fan-out: processor shutdown failed: %s", exc) + with self._lock: + self._processors.clear() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + results: Final = tuple(self._flush_one(processor, timeout_millis) for processor in self._snapshot()) + return all(results) + + def _snapshot(self) -> tuple[SpanProcessor, ...]: + with self._lock: + return tuple(self._processors.values()) + + @staticmethod + def _flush_one(processor: SpanProcessor, timeout_millis: int) -> bool: + try: + return processor.force_flush(timeout_millis) + except Exception: # noqa: BLE001 # one exporter's flush failure must not fail the whole flush + return False + + def _processor_for(self, destination: "OtelDestination") -> SpanProcessor | None: + key: Final = destination.cache_key() + with self._lock: + cached: Final = self._processors.get(key) + if cached is not None: + self._processors.move_to_end(key) + return cached + built: Final = self._build(destination) + if built is None: + return None + with self._lock: + existing: Final = self._processors.get(key) + if existing is not None: + # Another thread won the race; drop ours rather than leak its thread. + _shutdown_quietly(built) + return existing + self._processors[key] = built + evicted: Final = ( + self._processors.popitem(last=False)[1] + if len(self._processors) > _MAX_CACHED_DESTINATION_PROCESSORS + else None + ) + if evicted is not None: + _shutdown_quietly(evicted) + return built + + +def _destination_processor(destination: "OtelDestination") -> SpanProcessor | None: + """A batching OTLP processor aimed at ``destination``, or ``None`` if unbuildable.""" + try: + spec: Final = ExporterSpec( + kind=destination.protocol or "otlp_http", + endpoint=destination.endpoint, + headers=destination.header_string(), + owner=None, + ) + return _processor_for(_exporter_from_spec(spec), use_simple=False) + except Exception as exc: # noqa: BLE001 # a malformed destination must not break the request or the other destinations + verbose_logger.debug("OTel V2 fan-out: no processor for %s: %s", destination.endpoint, exc) + return None + + +def _shutdown_quietly(processor: SpanProcessor) -> None: + try: + processor.shutdown() + except Exception as exc: # noqa: BLE001 # defensive: shedding a spare processor must not raise + verbose_logger.debug("OTel V2 fan-out: discarding processor failed: %s", exc) + + +class _OverriddenBackendFilter(SpanProcessor): + """Hold a span back from ``owner``'s operator-level exporter when the request + pointed ``owner`` at a tenant's own account. + + A team destination is an override rather than an addition, and a ``SpanProcessor`` + cannot veto its siblings (``SynchronousMultiSpanProcessor.on_end`` ignores return + values), so suppression has to wrap the exporter's own processor. Dropping here + also keeps the span out of ``BatchSpanProcessor``'s bounded queue instead of + filling it with spans that will never ship. + """ + + def __init__(self, inner: SpanProcessor, owner: str) -> None: + self._inner: Final = inner + self._owner: Final = owner + + def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: + self._inner.on_start(span, parent_context) + + def on_end(self, span: ReadableSpan) -> None: + if self._owner in overridden_backends(): + return + self._inner.on_end(span) + + def shutdown(self) -> None: + self._inner.shutdown() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return self._inner.force_flush(timeout_millis) + + def build_span_exporter(config: OpenTelemetryV2Config) -> SpanExporter: """Build a single exporter from the top-level config fields. @@ -437,6 +619,7 @@ def build_tracer_provider( exporter: SpanExporter | None = None, baggage_processor: SpanProcessor | None = None, use_simple_processor: bool | None = None, + tenant_overrides: bool = False, ) -> TracerProvider: """Build the shared :class:`TracerProvider`. @@ -445,6 +628,12 @@ def build_tracer_provider( ``config.exporters`` entry — this is what fans spans out to multiple backends. ``exporter`` and ``use_simple_processor`` are explicit overrides: pass a single exporter to attach exactly that one (used by tests). + + ``tenant_overrides`` belongs to the operator-level provider alone: it wraps each + owned exporter so a request that pointed that backend at a key's or team's own + account skips it, and adds the fan-out processor that delivers to that account + instead. The per-tenant providers this same function builds must leave it off, + or they would filter out the very spans they exist to carry. """ provider: Final = TracerProvider(resource=build_resource(config)) if baggage_processor is None: @@ -461,12 +650,16 @@ def build_tracer_provider( if spec.requires_headers and not spec.headers: continue exp = _exporter_from_spec(spec) - provider.add_span_processor( - _processor_for( - exp, - (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), - ) + processor = _processor_for( + exp, + (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), ) + owner = spec.owner.value if spec.owner is not None else None + provider.add_span_processor( + _OverriddenBackendFilter(processor, owner) if tenant_overrides and owner is not None else processor + ) + if tenant_overrides: + provider.add_span_processor(TenantFanOutSpanProcessor()) return provider diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 227e18f3663..710c51d4942 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -25,6 +25,7 @@ from opentelemetry.trace import Tracer from litellm._logging import verbose_logger from litellm.constants import OTEL_SERVICE_NAME_METADATA_KEYS from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config +from litellm.integrations.otel.plumbing.context import overridden_backends from litellm.integrations.otel.plumbing.providers import ( build_tracer_provider, exporter_transport, @@ -231,7 +232,14 @@ class TenantTracerCache: concurrent overflow eviction can't shut it down between selection and the caller's span start. The caller must ``release`` it exactly once. """ - credential_headers: Final = self._credential_headers(dynamic_params) + # An overridden backend is delivered by the fan-out processor, which carries the + # whole trace. Routing here too would detach this span onto a second provider, + # so the tenant would get the request tree plus a stray one-span trace. + credential_headers: Final = ( + _NO_HEADERS + if self._callback_name is not None and self._callback_name in overridden_backends() + else self._credential_headers(dynamic_params) + ) project_headers: Final = self._project_headers(auth_metadata) service_name: Final = tenant_service_name(auth_metadata) if not credential_headers and not project_headers and service_name is None: diff --git a/litellm/integrations/otel/presets/agentops.py b/litellm/integrations/otel/presets/agentops.py index f45b1cd3cff..965213f2ee4 100644 --- a/litellm/integrations/otel/presets/agentops.py +++ b/litellm/integrations/otel/presets/agentops.py @@ -39,6 +39,7 @@ class _AgentOpsSettings(BaseSettings): def agentops_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: """Build the AgentOps config without any network I/O. diff --git a/litellm/integrations/otel/presets/arize.py b/litellm/integrations/otel/presets/arize.py index ee0de675657..63222856b2a 100644 --- a/litellm/integrations/otel/presets/arize.py +++ b/litellm/integrations/otel/presets/arize.py @@ -11,7 +11,10 @@ from litellm.integrations.otel.model.config import ( ExporterSpec, OpenTelemetryV2Config, ) -from litellm.integrations.otel.presets.utils import ensure_mappers +from litellm.integrations.otel.presets.utils import ( + credential_gated_exporters, + ensure_mappers, +) from litellm.types.utils import StandardCallbackDynamicParams @@ -26,10 +29,22 @@ class _ArizeSettings(BaseSettings): def arize_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: - arize_cfg: Final = _V1ArizeLogger.get_arize_config() - headers: Final = _arize_headers(arize_cfg) base: Final = config_overrides or OpenTelemetryV2Config() + mappers: Final = ensure_mappers(base.mapper_names, "openinference") + try: + arize_cfg: Final = _V1ArizeLogger.get_arize_config() + except Exception: + if not allow_missing_credentials: + raise + return base.model_copy( + update={ # mutable-ok: pydantic model_copy takes a plain update mapping + "exporters": credential_gated_exporters(base.exporters, ExporterOwner.ARIZE_AX), + "mapper_names": mappers, + } + ) + headers: Final = _arize_headers(arize_cfg) return base.model_copy( update={ "exporters": [ @@ -41,7 +56,7 @@ def arize_preset( owner=ExporterOwner.ARIZE_AX, ), ], - "mapper_names": ensure_mappers(base.mapper_names, "openinference"), + "mapper_names": mappers, "resource_attributes": { **base.resource_attributes, **({"model_id": arize_cfg.project_name} if arize_cfg.project_name else {}), diff --git a/litellm/integrations/otel/presets/base.py b/litellm/integrations/otel/presets/base.py index 3b9991f86a4..3b7264abd88 100644 --- a/litellm/integrations/otel/presets/base.py +++ b/litellm/integrations/otel/presets/base.py @@ -18,6 +18,18 @@ class Preset(Protocol): ``config_overrides`` lets one preset layer onto another's config (or onto test-supplied defaults); the factory calls presets with no arguments. + + ``allow_missing_credentials`` lets a credential-mandatory backend (langfuse / + arize / weave) degrade to an exporter-less, mapper-only config instead of + raising when the operator set no env credentials of their own. That is a real + deployment: every team brings its own account and the operator keeps none, and + without it the whole V2 path silently falls back to the legacy integration, so + no team destination is ever reached. Credential-optional backends ignore it. """ - def __call__(self, *, config_overrides: OpenTelemetryV2Config | None = None) -> OpenTelemetryV2Config: ... + def __call__( + self, + *, + config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, + ) -> OpenTelemetryV2Config: ... diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py new file mode 100644 index 00000000000..966d8f4e1f4 --- /dev/null +++ b/litellm/integrations/otel/presets/destinations.py @@ -0,0 +1,118 @@ +"""Map a key's or team's callback vars to the OTLP destination its traces export to. + +The auth path resolves one destination per backend the caller configured, and the +fan-out span processor exports the whole request through it. Header building is +delegated to each preset's existing ``*_dynamic_headers`` builder, so a destination +authenticates exactly the way the per-request tracer route already did; only the +endpoint needs a per-backend rule, because a backend's host is either fixed, taken +from a region table, or named by the tenant alongside its own key pair. +""" + +import os +from collections.abc import Callable, Mapping +from types import MappingProxyType +from typing import Final + +from litellm.integrations.otel.model.destination import OtelDestination +from litellm.types.utils import StandardCallbackDynamicParams + +#: gRPC is Arize's own transport; an explicitly named HTTP collector overrides it. +_ARIZE_GRPC_ENDPOINT: Final = "https://otlp.arize.com/v1" + + +def _langfuse_endpoint(params: StandardCallbackDynamicParams) -> str | None: + """The tenant's own Langfuse host, else the operator's, else Langfuse US cloud. + + Falling back to the operator's host is safe and is what V1 does: the tenant's + own key pair still selects its own project, and a self-hosted deployment where + every team lives on one Langfuse server is the common shape. + """ + from litellm.integrations.langfuse.langfuse_otel import ( + LANGFUSE_CLOUD_US_ENDPOINT, + LangfuseOtelLogger, + ) + + host: Final = params.get("langfuse_host") or LangfuseOtelLogger._get_langfuse_otel_host() # pyright: ignore[reportPrivateUsage] # reuse the backend's own env host resolver rather than duplicating it + if not host: + return LANGFUSE_CLOUD_US_ENDPOINT + normalized: Final = host if host.startswith("http") else f"https://{host}" + return f"{normalized.rstrip('/')}/api/public/otel" + + +def _arize_endpoint(params: StandardCallbackDynamicParams) -> str | None: + return os.environ.get("ARIZE_ENDPOINT") or _ARIZE_GRPC_ENDPOINT + + +def _arize_protocol(params: StandardCallbackDynamicParams) -> str | None: + return "otlp_http" if os.environ.get("ARIZE_HTTP_ENDPOINT") and not os.environ.get("ARIZE_ENDPOINT") else None + + +def _weave_endpoint(params: StandardCallbackDynamicParams) -> str | None: + from litellm.integrations.weave.weave_otel import WEAVE_BASE_URL, WEAVE_OTEL_ENDPOINT + + return WEAVE_BASE_URL + WEAVE_OTEL_ENDPOINT + + +def _newrelic_endpoint(params: StandardCallbackDynamicParams) -> str | None: + from litellm.integrations.otel.presets.newrelic import newrelic_dynamic_endpoint + + return newrelic_dynamic_endpoint(params) + + +#: Callback name -> endpoint resolver. A backend is destination-capable exactly +#: when it appears here AND in ``DYNAMIC_HEADERS_BY_CALLBACK``: without a header +#: builder the destination would carry no tenant credentials, and the exporter +#: would post the tenant's traffic to the operator's account. +_ENDPOINT_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], str | None]]] = MappingProxyType( + { + "langfuse_otel": _langfuse_endpoint, + "arize": _arize_endpoint, + "weave_otel": _weave_endpoint, + "newrelic": _newrelic_endpoint, + } +) + +_PROTOCOL_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], str | None]]] = MappingProxyType( + { + "arize": _arize_protocol, + } +) + +_NO_ATTRS: Final[Mapping[str, str]] = MappingProxyType({}) + + +def destination_capable_backends() -> frozenset[str]: + """Backends a key or team can point at its own account.""" + from litellm.integrations.otel.presets import DYNAMIC_HEADERS_BY_CALLBACK + + return frozenset(_ENDPOINT_BY_CALLBACK) & frozenset(DYNAMIC_HEADERS_BY_CALLBACK) + + +def destination_for(callback_name: str, params: StandardCallbackDynamicParams) -> OtelDestination | None: + """The destination ``params`` names for ``callback_name``, or ``None``. + + ``None`` means the caller configured nothing usable for this backend, so the + request keeps the operator's global exporters. A partial config (a host with + no key pair) resolves to ``None`` rather than to the operator's endpoint with + the tenant's host, which would post the operator's credentials elsewhere. + """ + from litellm.integrations.otel.presets import DYNAMIC_HEADERS_BY_CALLBACK + + header_builder: Final = DYNAMIC_HEADERS_BY_CALLBACK.get(callback_name) + endpoint_builder: Final = _ENDPOINT_BY_CALLBACK.get(callback_name) + if header_builder is None or endpoint_builder is None: + return None + headers: Final = header_builder(params) + if not headers: + return None + endpoint: Final = endpoint_builder(params) + if not endpoint: + return None + protocol_builder: Final = _PROTOCOL_BY_CALLBACK.get(callback_name) + return OtelDestination( + endpoint=endpoint, + headers=MappingProxyType(dict(headers)), + resource_attributes=_NO_ATTRS, + callback_name=callback_name, + protocol=protocol_builder(params) if protocol_builder is not None else None, + ) diff --git a/litellm/integrations/otel/presets/langfuse.py b/litellm/integrations/otel/presets/langfuse.py index c2f64422eff..9149e0c0d94 100644 --- a/litellm/integrations/otel/presets/langfuse.py +++ b/litellm/integrations/otel/presets/langfuse.py @@ -10,17 +10,32 @@ from litellm.integrations.otel.model.config import ( ExporterSpec, OpenTelemetryV2Config, ) -from litellm.integrations.otel.presets.utils import ensure_mappers +from litellm.integrations.otel.presets.utils import ( + credential_gated_exporters, + ensure_mappers, +) from litellm.types.utils import StandardCallbackDynamicParams def langfuse_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: - cfg: Final = _V1Langfuse.get_langfuse_otel_config() - kind: Final = cfg.exporter if isinstance(cfg.exporter, str) else "otlp_http" base: Final = config_overrides or OpenTelemetryV2Config() + mappers: Final = ensure_mappers(base.mapper_names, "langfuse") + try: + cfg: Final = _V1Langfuse.get_langfuse_otel_config() + except Exception: + if not allow_missing_credentials: + raise + return base.model_copy( + update={ # mutable-ok: pydantic model_copy takes a plain update mapping + "exporters": credential_gated_exporters(base.exporters, ExporterOwner.LANGFUSE_OTEL), + "mapper_names": mappers, + } + ) + kind: Final = cfg.exporter if isinstance(cfg.exporter, str) else "otlp_http" return base.model_copy( update={ "exporters": [ @@ -32,7 +47,7 @@ def langfuse_preset( owner=ExporterOwner.LANGFUSE_OTEL, ), ], - "mapper_names": ensure_mappers(base.mapper_names, "langfuse"), + "mapper_names": mappers, } ) diff --git a/litellm/integrations/otel/presets/langtrace.py b/litellm/integrations/otel/presets/langtrace.py index c88e4715ab0..2312575f04a 100644 --- a/litellm/integrations/otel/presets/langtrace.py +++ b/litellm/integrations/otel/presets/langtrace.py @@ -9,6 +9,7 @@ from litellm.integrations.otel.presets.utils import ensure_mappers def langtrace_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: """Compose the Langtrace mapper on top of the customer's OTLP destination. diff --git a/litellm/integrations/otel/presets/levo.py b/litellm/integrations/otel/presets/levo.py index 41b3758cf3e..c1580cf7a5b 100644 --- a/litellm/integrations/otel/presets/levo.py +++ b/litellm/integrations/otel/presets/levo.py @@ -13,6 +13,7 @@ from litellm.integrations.otel.model.config import ( def levo_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: cfg: Final = _V1Levo.get_levo_config() base: Final = config_overrides or OpenTelemetryV2Config() diff --git a/litellm/integrations/otel/presets/newrelic.py b/litellm/integrations/otel/presets/newrelic.py index 4660a707355..771b3f643c8 100644 --- a/litellm/integrations/otel/presets/newrelic.py +++ b/litellm/integrations/otel/presets/newrelic.py @@ -44,6 +44,7 @@ class _NewRelicSettings(BaseSettings): def newrelic_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: settings: Final = _NewRelicSettings() base: Final = config_overrides or OpenTelemetryV2Config() diff --git a/litellm/integrations/otel/presets/phoenix.py b/litellm/integrations/otel/presets/phoenix.py index eef407b6c1b..f4f34ee7525 100644 --- a/litellm/integrations/otel/presets/phoenix.py +++ b/litellm/integrations/otel/presets/phoenix.py @@ -60,6 +60,7 @@ def phoenix_project_headers(auth_metadata: Mapping[str, str] | None) -> Mapping[ def phoenix_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: cfg: Final = _V1Phoenix.get_arize_phoenix_config() headers: Final = cfg.otlp_auth_headers if hasattr(cfg, "otlp_auth_headers") else None diff --git a/litellm/integrations/otel/presets/utils.py b/litellm/integrations/otel/presets/utils.py index 328569d3daf..ef6e80ff33e 100644 --- a/litellm/integrations/otel/presets/utils.py +++ b/litellm/integrations/otel/presets/utils.py @@ -3,6 +3,11 @@ from collections.abc import Iterable from typing import Final +from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec + +#: What ``OpenTelemetryV2Config._normalize`` folds in when no destination is configured. +_DEFAULT_SHORTHAND_EXPORTER: Final = ExporterSpec() + def ensure_mappers(mapper_names: Iterable[str], *names: str) -> list[str]: """Return ``mapper_names`` with each of ``names`` appended if not already present. @@ -15,3 +20,21 @@ def ensure_mappers(mapper_names: Iterable[str], *names: str) -> list[str]: if name not in result: result.append(name) return result + + +def credential_gated_exporters( + exporters: "Iterable[ExporterSpec]", owner: "ExporterOwner" +) -> "tuple[ExporterSpec, ...]": + """``exporters`` with the operator's destination replaced by a header-gated one. + + Used when a credential-mandatory backend is asked to build without the operator's + own credentials, so only key/team destinations receive spans. Two things have to + happen for that to mean "export nowhere": the placeholder console spec that + ``OpenTelemetryV2Config`` folds in for an empty exporter list is dropped, or every + span would be printed to stdout, and the gated spec keeps the owner so the + override filter still recognises which backend this provider speaks for. + """ + return ( + *(spec for spec in exporters if spec != _DEFAULT_SHORTHAND_EXPORTER), + ExporterSpec(owner=owner, requires_headers=True), + ) diff --git a/litellm/integrations/otel/presets/weave.py b/litellm/integrations/otel/presets/weave.py index 51d0ad01093..644cd39ad36 100644 --- a/litellm/integrations/otel/presets/weave.py +++ b/litellm/integrations/otel/presets/weave.py @@ -7,7 +7,10 @@ from litellm.integrations.otel.model.config import ( ExporterSpec, OpenTelemetryV2Config, ) -from litellm.integrations.otel.presets.utils import ensure_mappers +from litellm.integrations.otel.presets.utils import ( + credential_gated_exporters, + ensure_mappers, +) from litellm.integrations.weave.weave_otel import ( _get_weave_authorization_header, get_weave_otel_config, @@ -18,9 +21,21 @@ from litellm.types.utils import StandardCallbackDynamicParams def weave_preset( *, config_overrides: OpenTelemetryV2Config | None = None, + allow_missing_credentials: bool = False, ) -> OpenTelemetryV2Config: - weave_cfg: Final = get_weave_otel_config() base: Final = config_overrides or OpenTelemetryV2Config() + mappers: Final = ensure_mappers(base.mapper_names, "openinference", "weave") + try: + weave_cfg: Final = get_weave_otel_config() + except Exception: + if not allow_missing_credentials: + raise + return base.model_copy( + update={ # mutable-ok: pydantic model_copy takes a plain update mapping + "exporters": credential_gated_exporters(base.exporters, ExporterOwner.WEAVE_OTEL), + "mapper_names": mappers, + } + ) return base.model_copy( update={ "exporters": [ @@ -33,7 +48,7 @@ def weave_preset( ), ], # Weave consumes OpenInference + a small Weave-specific overlay. - "mapper_names": ensure_mappers(base.mapper_names, "openinference", "weave"), + "mapper_names": mappers, } ) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 989777e9412..c7e8678be68 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -4815,11 +4815,16 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name: return callback try: - config: Final = preset_fn() + config: Final = preset_fn(allow_missing_credentials=True) except Exception: # If env vars are missing or the preset raises, defer to the legacy path # so customers get the same error story they had before V2 landed. return None + if all(spec.requires_headers and not spec.headers for spec in config.exporters): + verbose_logger.warning( + "OTel V2: no operator credentials for '%s'; only key/team destinations will receive its traces", + callback_name, + ) v2_logger: Final = build_otel_v2_logger(config=config, callback_name=callback_name) _in_memory_loggers.append(v2_logger) return v2_logger diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 93293db24c6..4a0257311d1 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -2845,6 +2845,28 @@ async def _authorize_authenticated_request( @tracer.wrap() +def _seed_request_destinations(user_api_key_dict: UserAPIKeyAuth) -> None: + """Anchor the OTLP destinations this key or team overrides its traces to. + + Called inside the ``auth`` phase span so that span reaches the tenant's account + as well, and on the request task so the ``ContextVar`` is inherited by the logging + tasks that close the LLM span. Best-effort: trace routing must never fail auth. + + The two ``postgres`` spans under ``auth`` close before this runs, because they are + the reads that resolve the identity being read here, so they keep going to the + operator's backend alone. + """ + try: + from litellm.integrations.otel.plumbing.context import set_request_destinations + from litellm.proxy.litellm_pre_call_utils import ( + resolve_tenant_otel_destinations, + ) + + set_request_destinations(resolve_tenant_otel_destinations(user_api_key_dict)) + except Exception as exc: # noqa: BLE001 # telemetry routing is best-effort and must never break authentication + verbose_proxy_logger.debug("OTel V2: tenant destination resolution failed: %s", exc) + + async def user_api_key_auth( request: Request, api_key: str = fastapi.Security(api_key_header), @@ -2891,6 +2913,7 @@ async def user_api_key_auth( raise body_parse_exception raise user_api_key_auth_obj.budget_reservation = None + _seed_request_destinations(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 diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index d026c5510e6..6069ed9a8c6 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -10,6 +10,7 @@ from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, cast from fastapi import HTTPException, Request +from pydantic import TypeAdapter from pydantic import ValidationError as PydanticValidationError from starlette.datastructures import Headers @@ -157,6 +158,7 @@ from litellm.types.utils import ( CustomPricingLiteLLMParams, LlmProviders, ProviderSpecificHeader, + StandardCallbackDynamicParams, StandardLoggingUserAPIKeyMetadata, SupportedCacheControls, ) @@ -170,6 +172,7 @@ _ENABLE_TEAM_STALE_ALIAS_BYPASS: bool | None = None if TYPE_CHECKING: + from litellm.integrations.otel.model.destination import OtelDestination from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig from litellm.types.proxy.policy_engine import Policy, PolicyMatchContext @@ -974,6 +977,51 @@ def _get_dynamic_logging_metadata( return callback_settings_obj +_TENANT_OTEL_PARAMS: Final = TypeAdapter(StandardCallbackDynamicParams) + + +def _tenant_otel_params(callback_vars: Mapping[str, str]) -> StandardCallbackDynamicParams: + try: + return _TENANT_OTEL_PARAMS.validate_python(callback_vars) + except PydanticValidationError: + return StandardCallbackDynamicParams() + + +def resolve_tenant_otel_destinations( + user_api_key_dict: UserAPIKeyAuth, +) -> "tuple[OtelDestination, ...]": + """The OTLP destinations this request's key or team config overrides its traces to. + + Key settings win over team settings outright, the same precedence + ``_get_dynamic_logging_metadata`` applies, so one caller never exports the same + backend to two accounts. Returns empty when OTEL V2 is off, when neither level + named a destination-capable backend, or when the config is incomplete, and the + request then keeps the operator's own exporters. + """ + from litellm.integrations.otel.model.config import is_otel_v2_enabled + from litellm.integrations.otel.presets.destinations import destination_for + + if not is_otel_v2_enabled(): + return () + entries: Final = KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings( + user_api_key_dict + ) or KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(user_api_key_dict) + if not entries: + return () + resolved: Final = tuple( + destination + for item in entries + if (callback := _get_validated_callback_metadata(item=item, source="otel-destination")) is not None + if (destination := destination_for(callback.callback_name, _tenant_otel_params(callback.callback_vars))) + is not None + ) + return tuple( + destination + for index, destination in enumerate(resolved) + if destination.callback_name not in tuple(earlier.callback_name for earlier in resolved[:index]) + ) + + def clean_headers( headers: Headers, litellm_key_header_name: str | None = None, diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py new file mode 100644 index 00000000000..ee67160533f --- /dev/null +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -0,0 +1,360 @@ +"""Key/team OTLP destinations override the operator's exporters for that backend.""" + +import contextvars +from collections.abc import Mapping + +import pytest +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from litellm.integrations.otel.model.config import ( + ExporterOwner, + ExporterSpec, + OpenTelemetryV2Config, + is_otel_v2_enabled, +) +from litellm.integrations.otel.model.destination import OtelDestination +from litellm.integrations.otel.plumbing.context import ( + overridden_backends, + request_destinations, + set_request_destinations, +) +from litellm.integrations.otel.plumbing.providers import ( + TenantFanOutSpanProcessor, + _OverriddenBackendFilter, + build_tracer_provider, +) +from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer +from litellm.integrations.otel.presets.destinations import ( + destination_capable_backends, + destination_for, +) +from litellm.integrations.otel.presets.langfuse import langfuse_preset +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.litellm_pre_call_utils import resolve_tenant_otel_destinations + +LANGFUSE_DEST = OtelDestination( + endpoint="http://tenant.local/api/public/otel", + headers={"Authorization": "Basic dGVuYW50"}, + callback_name="langfuse_otel", +) + + +def in_fresh_context(fn, *args): + """Run ``fn`` in its own context so one test's destinations never leak.""" + return contextvars.copy_context().run(fn, *args) + + +def emit(provider: TracerProvider, name: str = "chat gpt-4") -> None: + with get_tracer(provider, "litellm").start_as_current_span(name): + pass + + +def wired_provider(dest_exporter: InMemorySpanExporter, global_exporter: InMemorySpanExporter) -> TracerProvider: + """The operator's provider: one owned exporter plus the tenant fan-out.""" + provider = TracerProvider() + provider.add_span_processor(_OverriddenBackendFilter(SimpleSpanProcessor(global_exporter), "langfuse_otel")) + provider.add_span_processor( + TenantFanOutSpanProcessor(processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter)) + ) + return provider + + +class TestOverrideSuppression: + def test_operator_exporter_keeps_the_span_when_no_destination_is_resolved(self): + global_exporter, dest_exporter = InMemorySpanExporter(), InMemorySpanExporter() + provider = wired_provider(dest_exporter, global_exporter) + + in_fresh_context(emit, provider) + + assert [s.name for s in global_exporter.get_finished_spans()] == ["chat gpt-4"] + assert dest_exporter.get_finished_spans() == () + + def test_operator_exporter_is_skipped_once_the_backend_is_overridden(self): + global_exporter, dest_exporter = InMemorySpanExporter(), InMemorySpanExporter() + provider = wired_provider(dest_exporter, global_exporter) + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + emit(provider) + + in_fresh_context(run) + + assert global_exporter.get_finished_spans() == () + assert [s.name for s in dest_exporter.get_finished_spans()] == ["chat gpt-4"] + + def test_a_backend_the_request_did_not_override_still_exports(self): + arize_exporter, dest_exporter = InMemorySpanExporter(), InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(_OverriddenBackendFilter(SimpleSpanProcessor(arize_exporter), "arize")) + provider.add_span_processor( + TenantFanOutSpanProcessor(processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter)) + ) + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + emit(provider) + + in_fresh_context(run) + + assert [s.name for s in arize_exporter.get_finished_spans()] == ["chat gpt-4"] + + +class TestFanOut: + def test_every_span_of_the_request_reaches_the_destination_in_one_trace(self): + """The whole tree, gen-AI span included, parented as the operator would see it.""" + dest_exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor( + TenantFanOutSpanProcessor(processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter)) + ) + tracer = get_tracer(provider, "litellm") + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + with tracer.start_as_current_span("POST /v1/chat/completions"): + with tracer.start_as_current_span("auth /v1/chat/completions"): + pass + with tracer.start_as_current_span("chat gpt-4"): + pass + + in_fresh_context(run) + + spans = dest_exporter.get_finished_spans() + by_name = {s.name: s for s in spans} + assert set(by_name) == {"POST /v1/chat/completions", "auth /v1/chat/completions", "chat gpt-4"} + root = by_name["POST /v1/chat/completions"] + assert len({s.context.trace_id for s in spans}) == 1, "the tenant must receive one connected trace" + for child in ("auth /v1/chat/completions", "chat gpt-4"): + assert by_name[child].parent.span_id == root.context.span_id + + def test_two_destinations_each_receive_their_own_copy(self): + first, second = InMemorySpanExporter(), InMemorySpanExporter() + by_endpoint = {"http://a.local": first, "http://b.local": second} + provider = TracerProvider() + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda d: SimpleSpanProcessor(by_endpoint[d.endpoint]), + ) + ) + + def run(): + set_request_destinations( + ( + OtelDestination(endpoint="http://a.local", callback_name="langfuse_otel"), + OtelDestination(endpoint="http://b.local", callback_name="arize"), + ) + ) + emit(provider) + + in_fresh_context(run) + + assert [s.name for s in first.get_finished_spans()] == ["chat gpt-4"] + assert [s.name for s in second.get_finished_spans()] == ["chat gpt-4"] + + def test_a_destination_that_cannot_build_a_processor_is_skipped_quietly(self): + """An unbuildable destination must not cost the caller its request.""" + attempts = [] + reached_the_end = [] + + def factory(destination): + attempts.append(destination.endpoint) + + provider = TracerProvider() + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=factory)) + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + emit(provider) + reached_the_end.append(True) + + in_fresh_context(run) + + assert attempts == [LANGFUSE_DEST.endpoint] + assert reached_the_end == [True] + + def test_one_processor_is_reused_across_spans_of_the_same_destination(self): + built = [] + + def factory(_destination): + processor = SimpleSpanProcessor(InMemorySpanExporter()) + built.append(processor) + return processor + + provider = TracerProvider() + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=factory)) + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + emit(provider, "one") + emit(provider, "two") + + in_fresh_context(run) + + assert len(built) == 1 + + +class TestProviderWiring: + def test_build_tracer_provider_only_filters_when_asked(self): + config = OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.LANGFUSE_OTEL)]) + operator = build_tracer_provider(config, tenant_overrides=True) + tenant = build_tracer_provider(config) + + def kinds(provider): + return [type(p).__name__ for p in provider._active_span_processor._span_processors] + + assert "_OverriddenBackendFilter" in kinds(operator) + assert "TenantFanOutSpanProcessor" in kinds(operator) + assert "_OverriddenBackendFilter" not in kinds(tenant), "a per-tenant provider must not filter itself out" + assert "TenantFanOutSpanProcessor" not in kinds(tenant) + + +class TestRouting: + def test_an_overridden_backend_is_not_detached_onto_a_second_provider(self): + config = OpenTelemetryV2Config( + exporters=[ExporterSpec(kind="otlp_http", endpoint="http://op.local", owner=ExporterOwner.LANGFUSE_OTEL)] + ) + cache = TenantTracerCache(config, "langfuse_otel", "litellm") + default = get_tracer(TracerProvider(), "litellm") + params = {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"} + + assert cache.route_for(default, params).detached is True + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + return cache.route_for(default, params) + + route = in_fresh_context(run) + assert route.detached is False + assert route.tracer is default + assert route.provider is None + + +class TestDestinationResolution: + def test_a_langfuse_key_pair_and_host_become_a_destination(self, monkeypatch): + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + auth = UserAPIKeyAuth( + team_metadata={ + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": { + "langfuse_public_key": "pk-team", + "langfuse_secret_key": "sk-team", + "langfuse_host": "http://team.local", + }, + } + ] + } + ) + + destinations = resolve_tenant_otel_destinations(auth) + + assert [d.endpoint for d in destinations] == ["http://team.local/api/public/otel"] + assert destinations[0].callback_name == "langfuse_otel" + + def test_the_key_wins_over_the_team_for_the_same_backend(self, monkeypatch): + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + + def entry(host: str) -> Mapping[str, object]: + return { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": { + "langfuse_public_key": "pk", + "langfuse_secret_key": "sk", + "langfuse_host": host, + }, + } + + auth = UserAPIKeyAuth( + metadata={"logging": [entry("http://key.local")]}, + team_metadata={"logging": [entry("http://team.local")]}, + ) + + assert [d.endpoint for d in resolve_tenant_otel_destinations(auth)] == ["http://key.local/api/public/otel"] + + def test_nothing_resolves_while_otel_v2_is_off(self, monkeypatch): + monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) + is_otel_v2_enabled.cache_clear() + auth = UserAPIKeyAuth( + team_metadata={ + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, + } + ] + } + ) + + assert resolve_tenant_otel_destinations(auth) == () + + def test_a_host_without_its_key_pair_resolves_to_nothing(self): + assert destination_for("langfuse_otel", {"langfuse_host": "http://team.local"}) is None + + def test_a_backend_with_no_dynamic_credentials_has_no_destination(self): + assert "arize_phoenix" not in destination_capable_backends() + assert destination_for("arize_phoenix", {"arize_api_key": "k"}) is None + + def test_the_destination_header_string_survives_the_exporter_round_trip(self): + from litellm.integrations.otel.plumbing.providers import parse_headers + + destination = destination_for( + "langfuse_otel", + {"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_host": "http://x"}, + ) + assert parse_headers(destination.header_string())["authorization"] == destination.headers["Authorization"] + + +class TestPresetDegradation: + def test_a_credential_less_langfuse_exports_nowhere_instead_of_to_the_console(self, monkeypatch, capsys): + """``_normalize`` folds a console exporter in for an empty list, which would + print every span on a proxy whose teams bring their own credentials.""" + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) + + config = langfuse_preset(allow_missing_credentials=True) + provider = build_tracer_provider(config, tenant_overrides=True) + capsys.readouterr() + in_fresh_context(emit, provider) + provider.force_flush() + + assert capsys.readouterr().out == "" + assert "langfuse" in config.mapper_names + + def test_langfuse_still_raises_for_a_global_callback_with_no_credentials(self, monkeypatch): + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) + + with pytest.raises(ValueError, match="LANGFUSE_PUBLIC_KEY"): + langfuse_preset() + + def test_a_credential_less_proxy_still_builds_the_v2_logger(self, monkeypatch): + from litellm.litellm_core_utils.litellm_logging import _maybe_construct_otel_v2 + + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) + + is_otel_v2_enabled.cache_clear() + logger = _maybe_construct_otel_v2("langfuse_otel", []) + is_otel_v2_enabled.cache_clear() + + assert logger is not None, "team-only deployments must not fall back to the legacy integration" + assert all(spec.requires_headers and not spec.headers for spec in logger.config.exporters) + + +class TestContextIsolation: + def test_destinations_do_not_leak_between_requests(self): + def first(): + set_request_destinations((LANGFUSE_DEST,)) + return overridden_backends() + + assert in_fresh_context(first) == frozenset({"langfuse_otel"}) + assert in_fresh_context(request_destinations) == ()