mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
Merge pull request #40321 from BerriAI/litellm_backport_39654_rc_1_101_0
feat(otel): backport tenant trace destinations to rc/1.101.0 (#39654)
This commit is contained in:
yuneng-jiang
GitHub
commit
cbf0c76869
29 changed files with 4214 additions and 53 deletions
|
|
@ -325,6 +325,9 @@ ssl_certificate: Optional[str] = None
|
|||
user_url_validation: bool = True
|
||||
user_url_allowed_hosts: List[str] = []
|
||||
provider_url_destination_allowed_hosts: List[str] = []
|
||||
#: "override" (default) or "additive": whether a key or team destination replaces
|
||||
#: the operator's exporter for that backend or exports alongside it.
|
||||
otel_tenant_destination_mode: str | None = None
|
||||
ssl_ecdh_curve: Optional[str] = None # Set to 'X25519' to disable PQC and improve performance
|
||||
disable_streaming_logging: bool = False
|
||||
disable_token_counter: bool = False
|
||||
|
|
|
|||
|
|
@ -16,9 +16,11 @@ from opentelemetry.trace import (
|
|||
Span,
|
||||
Tracer,
|
||||
get_current_span,
|
||||
get_tracer_provider,
|
||||
set_span_in_context,
|
||||
use_span,
|
||||
)
|
||||
from opentelemetry.trace import TracerProvider as ApiTracerProvider
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
|
|
@ -63,6 +65,7 @@ from litellm.integrations.otel.plumbing.metrics import (
|
|||
create_genai_metrics,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.providers import (
|
||||
attach_tenant_fan_out,
|
||||
build_tracer_provider,
|
||||
get_event_logger,
|
||||
get_meter,
|
||||
|
|
@ -85,6 +88,7 @@ if TYPE_CHECKING:
|
|||
)
|
||||
|
||||
LITELLM_TRACER_NAME: Final = "litellm"
|
||||
_published_v2_provider: ApiTracerProvider | None = None
|
||||
|
||||
|
||||
def _span_error_from_exception(
|
||||
|
|
@ -180,7 +184,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)
|
||||
|
|
@ -195,6 +201,11 @@ class OpenTelemetryV2(CustomLogger):
|
|||
self._open_llm_calls: OrderedDict[str, _LLMCallSpan] = OrderedDict()
|
||||
self._init_otel_logger_on_litellm_proxy()
|
||||
|
||||
@property
|
||||
def tracer_provider(self) -> TracerProvider:
|
||||
"""The provider this logger emits through, read-only to its callers."""
|
||||
return self._tracer_provider
|
||||
|
||||
def _init_metrics(self, meter_provider: "MeterProvider | None") -> "GenAIMetricRecorder | None":
|
||||
"""Create the six GenAI histograms when metrics are enabled, else ``None``.
|
||||
|
||||
|
|
@ -863,12 +874,33 @@ def publish_global_otel_v2_provider(
|
|||
``opentelemetry.trace.set_tracer_provider``) are injected so the publish step is
|
||||
unit-testable without reading or mutating real global OTel state. Returns the
|
||||
logger whose provider was published.
|
||||
|
||||
The published provider is also the one that fans spans out to key/team
|
||||
destinations, because it is the only provider the whole request tree passes
|
||||
through; see :func:`attach_tenant_fan_out`. It is remembered for
|
||||
:func:`fan_out_provider` because neither the OTel global (``set_tracer_provider``
|
||||
keeps the first provider it was ever handed) nor
|
||||
``proxy_server.open_telemetry_logger`` (a legacy v1 logger can hold that slot)
|
||||
reliably leads back to it.
|
||||
"""
|
||||
global _published_v2_provider
|
||||
logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered)
|
||||
set_global_provider(logger._tracer_provider)
|
||||
attach_tenant_fan_out(logger.tracer_provider, *_v2_configs(in_memory_loggers, logger))
|
||||
set_global_provider(logger.tracer_provider)
|
||||
_published_v2_provider = logger.tracer_provider # rebind-ok: startup records the one provider carrying the fan-out
|
||||
return logger
|
||||
|
||||
|
||||
def _v2_configs(in_memory_loggers: Sequence[object], logger: "OpenTelemetryV2") -> tuple[OpenTelemetryV2Config, ...]:
|
||||
"""Every v2 logger's config, the published logger's first.
|
||||
|
||||
Each preset keeps its own provider and exporters, so the accounts the operator
|
||||
writes to are spread over all of them, not held by the published logger alone.
|
||||
"""
|
||||
others: Final = tuple(cb.config for cb in in_memory_loggers if isinstance(cb, OpenTelemetryV2) and cb is not logger)
|
||||
return (logger.config, *others)
|
||||
|
||||
|
||||
def _registered_v2_logger() -> "OpenTelemetryV2 | None":
|
||||
try:
|
||||
from litellm.proxy import proxy_server
|
||||
|
|
@ -904,6 +936,25 @@ def seed_request_identity(user_api_key_dict: object, model: str | None = None) -
|
|||
logger.seed_request_identity(user_api_key_dict, model=model)
|
||||
|
||||
|
||||
def fan_out_provider() -> ApiTracerProvider:
|
||||
"""The provider :func:`publish_global_otel_v2_provider` gave the tenant fan-out.
|
||||
|
||||
Read off the publish itself, not the OTel global and not the registered logger:
|
||||
the global keeps whichever provider claimed it first (auto-instrumentation, a
|
||||
legacy logger), and the registered slot can hold a v1 logger while the publish
|
||||
picked a v2 one from ``_in_memory_loggers``. Either detour lands on a provider
|
||||
with no fan-out and drops every destination at auth.
|
||||
"""
|
||||
published: Final = _published_v2_provider
|
||||
if published is not None:
|
||||
return published
|
||||
logger: Final = _registered_v2_logger()
|
||||
if logger is not None:
|
||||
attach_tenant_fan_out(logger.tracer_provider, logger.config)
|
||||
return logger.tracer_provider
|
||||
return get_tracer_provider()
|
||||
|
||||
|
||||
@contextmanager
|
||||
def phase_span(name: str) -> "Iterator[Span | None]":
|
||||
logger: Final = _registered_v2_logger()
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ from litellm.integrations.otel.model.payloads import (
|
|||
ServiceSpanData,
|
||||
ToolDefinition,
|
||||
)
|
||||
from litellm.integrations.otel.model.semconv import Error
|
||||
|
||||
# Attribute keys in the semconv-ai / Traceloop vocabulary.
|
||||
_LEGACY_SYSTEM: Final = "gen_ai.system"
|
||||
|
|
@ -36,7 +37,7 @@ _LEGACY_PRESENCE_PENALTY: Final = "llm.presence_penalty"
|
|||
_LEGACY_STOP_SEQUENCES: Final = "llm.chat.stop_sequences"
|
||||
_LEGACY_SERVICE: Final = "service"
|
||||
_LEGACY_CALL_TYPE: Final = "call_type"
|
||||
_LEGACY_ERROR: Final = "error"
|
||||
_LEGACY_ERROR: Final = Error.MESSAGE_LEGACY
|
||||
|
||||
|
||||
class LegacyMapper:
|
||||
|
|
|
|||
|
|
@ -253,7 +253,9 @@ class OpenTelemetryV2Config(BaseSettings):
|
|||
if self.endpoint and self.exporter == "console":
|
||||
self.exporter = "otlp_http"
|
||||
# When no explicit destinations are given, fold the single-destination
|
||||
# shorthand into one spec so the provider always has a destination.
|
||||
# shorthand into one spec so the provider always has a destination. A spec
|
||||
# with no fields set is how the presets tell "nothing configured" from an
|
||||
# operator who asked for the console by name.
|
||||
if not self.exporters:
|
||||
self.exporters = [
|
||||
ExporterSpec(
|
||||
|
|
@ -261,6 +263,8 @@ class OpenTelemetryV2Config(BaseSettings):
|
|||
endpoint=self.endpoint,
|
||||
headers=self.headers,
|
||||
)
|
||||
if not self.model_fields_set.isdisjoint(("exporter", "endpoint", "headers"))
|
||||
else ExporterSpec()
|
||||
]
|
||||
# Ensure ``genai`` is always present and first.
|
||||
names = list(self.mapper_names)
|
||||
|
|
|
|||
49
litellm/integrations/otel/model/destination.py
Normal file
49
litellm/integrations/otel/model/destination.py
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
"""The resolved OTLP destination a request's traces export to.
|
||||
|
||||
Backend-agnostic on purpose: every OTEL backend reduces to an endpoint plus auth
|
||||
headers. The per-backend field mapping lives in ``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, defaulting to the backend's own. Not derivable from the "
|
||||
"scheme: Arize's ``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, so one destination means one exporter."""
|
||||
return (
|
||||
self.endpoint,
|
||||
tuple(sorted(self.headers.items())),
|
||||
tuple(sorted(self.resource_attributes.items())),
|
||||
self.protocol,
|
||||
)
|
||||
|
||||
|
||||
NO_DESTINATIONS: Final[tuple[OtelDestination, ...]] = ()
|
||||
|
|
@ -204,6 +204,9 @@ class Error:
|
|||
|
||||
TYPE: Final = "error.type"
|
||||
MESSAGE: Final = "error.message"
|
||||
# The same text under the bare key the semconv-ai / Traceloop vocabulary uses
|
||||
# (see ``LegacyMapper``), so anything reading or redacting error text covers both.
|
||||
MESSAGE_LEGACY: Final = "error"
|
||||
|
||||
|
||||
class LiteLLMError:
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
"""Trace-context + Baggage helpers."""
|
||||
|
||||
import os
|
||||
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 +22,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 +308,65 @@ 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. Stateful MCP handlers set and reset it per message; the
|
||||
# request-task value otherwise dies with that task.
|
||||
_request_destinations: Final['ContextVar[tuple["OtelDestination", ...]]'] = ContextVar(
|
||||
"litellm_otel_request_destinations", default=()
|
||||
)
|
||||
|
||||
|
||||
def set_request_destinations(destinations: 'tuple["OtelDestination", ...]') -> "Token[tuple[OtelDestination, ...]]":
|
||||
"""Anchor the destinations this request exports to and return a reset token."""
|
||||
return _request_destinations.set(destinations)
|
||||
|
||||
|
||||
def reset_request_destinations(token: "Token[tuple[OtelDestination, ...]]") -> None:
|
||||
_request_destinations.reset(token)
|
||||
|
||||
|
||||
def request_destinations() -> 'tuple["OtelDestination", ...]':
|
||||
"""The destinations resolved for this request, empty outside a proxy request."""
|
||||
return _request_destinations.get()
|
||||
|
||||
|
||||
#: ``litellm_settings: otel_tenant_destination_mode`` and its env equivalent.
|
||||
ADDITIVE_DESTINATION_MODE: Final = "additive"
|
||||
OTEL_TENANT_DESTINATION_MODE_ENV: Final = "LITELLM_OTEL_TENANT_DESTINATION_MODE"
|
||||
|
||||
|
||||
def tenant_destinations_are_additive() -> bool:
|
||||
"""Whether a tenant destination exports alongside the operator's own exporter.
|
||||
|
||||
Override is the default: the tenant's traffic reaches the tenant's account and
|
||||
nowhere else. Operators running one org-wide backend across every team set this
|
||||
to ``additive`` so the same trace lands in both places.
|
||||
"""
|
||||
import litellm
|
||||
|
||||
configured: Final = litellm.otel_tenant_destination_mode or os.environ.get(OTEL_TENANT_DESTINATION_MODE_ENV)
|
||||
return isinstance(configured, str) and configured.strip().lower() == ADDITIVE_DESTINATION_MODE
|
||||
|
||||
|
||||
def destination_backends() -> frozenset[str]:
|
||||
"""Backends this request resolved a tenant destination for.
|
||||
|
||||
The fan-out already carries the whole trace to those destinations, so the
|
||||
per-request tracer route must never send a second copy, in either mode.
|
||||
"""
|
||||
return frozenset(d.callback_name for d in _request_destinations.get() if d.callback_name)
|
||||
|
||||
|
||||
def suppressed_backends() -> frozenset[str]:
|
||||
"""Backends whose operator-level exporters this request must NOT reach.
|
||||
|
||||
Empty under ``additive``, where the operator keeps its copy of every span.
|
||||
"""
|
||||
if tenant_destinations_are_additive():
|
||||
return frozenset()
|
||||
return destination_backends()
|
||||
|
|
|
|||
|
|
@ -1,9 +1,14 @@
|
|||
"""Provider / exporter factory + the Baggage span processor."""
|
||||
|
||||
from collections.abc import Callable, Iterable
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal
|
||||
|
||||
from opentelemetry import _logs, baggage, metrics
|
||||
from opentelemetry import _logs, baggage, metrics, trace
|
||||
from opentelemetry._events import EventLogger
|
||||
from opentelemetry._logs import LoggerProvider, NoOpLoggerProvider
|
||||
from opentelemetry.context import Context
|
||||
|
|
@ -19,7 +24,8 @@ 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 Event, ReadableSpan, SpanProcessor, TracerProvider
|
||||
from opentelemetry.sdk.trace import Span as SDKSpan
|
||||
from opentelemetry.sdk.trace.export import (
|
||||
BatchSpanProcessor,
|
||||
ConsoleSpanExporter,
|
||||
|
|
@ -29,18 +35,35 @@ from opentelemetry.sdk.trace.export import (
|
|||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
||||
InMemorySpanExporter,
|
||||
)
|
||||
from opentelemetry.trace import Span, SpanKind, Tracer
|
||||
from opentelemetry.trace import Span, SpanKind, Status, Tracer
|
||||
from opentelemetry.util.re import parse_env_headers
|
||||
from opentelemetry.util.types import Attributes, AttributeValue
|
||||
|
||||
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.semconv import (
|
||||
DB,
|
||||
MCP,
|
||||
Error,
|
||||
ExceptionEvent,
|
||||
GenAI,
|
||||
LiteLLM,
|
||||
LiteLLMError,
|
||||
Server,
|
||||
)
|
||||
from litellm.integrations.otel.model.spans import LiteLLMSpanKind
|
||||
from litellm.integrations.otel.plumbing.context import (
|
||||
request_destinations,
|
||||
suppressed_backends,
|
||||
)
|
||||
|
||||
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 +217,555 @@ 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
|
||||
|
||||
#: Workers closing shed destination processors, bounding the threads a tenant can
|
||||
#: create by cycling its destination config.
|
||||
_DRAIN_WORKERS: Final = 2
|
||||
|
||||
#: Shed processors waiting to be closed before the fan-out stops building new ones.
|
||||
#: Each still owns a batch thread until its close returns, and a collector that never
|
||||
#: answers makes every close take the exporter's full timeout, so past this many the
|
||||
#: operator's exporter keeps the span instead (see ``deliverable``).
|
||||
_MAX_PENDING_DRAINS: Final = 64
|
||||
|
||||
#: How long ``shutdown`` waits for spans already being forwarded, so teardown closes
|
||||
#: no processor under one. Bounded: an exporter that never returns must not hold the
|
||||
#: proxy open.
|
||||
_SHUTDOWN_DRAIN_SECONDS: Final = 5.0
|
||||
|
||||
#: An exporter's account: its normalized endpoint and the credentials it presents.
|
||||
_SinkKey = tuple[str, tuple[tuple[str, str], ...]]
|
||||
|
||||
#: Header names that spell one credential two ways. Arize's operator exporter sends
|
||||
#: ``space_id`` where a tenant destination sends ``arize-space-id``.
|
||||
_CREDENTIAL_ALIASES: Final = MappingProxyType({"arize_space_id": "space_id"})
|
||||
|
||||
|
||||
class _DrainPool:
|
||||
"""Closes shed destination processors off the span-export path.
|
||||
|
||||
``shutdown`` flushes over the network and is reached from ``on_end``, so closing
|
||||
one inline would let a single unreachable tenant collector stall every other
|
||||
tenant's spans behind it. A fixed set of workers rather than a thread per
|
||||
processor means a tenant cycling its destination config cannot spawn threads as
|
||||
fast as it can send requests; slow shutdowns queue behind each other.
|
||||
|
||||
The workers are daemons and belong to the fan-out that sheds the processors, so
|
||||
neither an unreachable collector nor a lazily built process-wide singleton can
|
||||
hold the proxy open on the way down.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
workers: int = _DRAIN_WORKERS,
|
||||
pending: "queue.Queue[SpanProcessor | None] | None" = None,
|
||||
capacity: int = _MAX_PENDING_DRAINS,
|
||||
) -> None:
|
||||
self._workers: Final = workers
|
||||
self._capacity: Final = capacity
|
||||
self._lock: Final = threading.Lock()
|
||||
self._closed = False
|
||||
self._backlog = 0 # guarded by ``_lock``: submitted processors whose close has not returned
|
||||
self._pending: Final[queue.Queue[SpanProcessor | None]] = pending if pending is not None else queue.Queue()
|
||||
self._threads: Final = tuple(
|
||||
threading.Thread(target=self._drain_until_closed, daemon=True, name="litellm-otel-destination-drain")
|
||||
for _ in range(workers)
|
||||
)
|
||||
for worker in self._threads:
|
||||
worker.start()
|
||||
|
||||
def submit(self, processor: SpanProcessor) -> None:
|
||||
"""Queue ``processor`` for closing, or hand it off once the pool is retired.
|
||||
|
||||
The check and the put share one lock. Reading a closed flag on its own leaves
|
||||
room for :meth:`close` to run in between, and the processor would land behind
|
||||
the sentinels every worker has already exited on.
|
||||
|
||||
Past close there is no worker left to take it, and the caller is whichever
|
||||
thread just ended a span, so closing it inline would park that thread on a
|
||||
network flush the shutdown deadline has already stopped waiting for. The extra
|
||||
thread is bounded by the same close: the fan-out stops handing processors out
|
||||
at that point, so only the ones already exporting when it happened arrive here.
|
||||
"""
|
||||
with self._lock:
|
||||
if not self._closed:
|
||||
self._backlog += 1
|
||||
self._pending.put(processor)
|
||||
return
|
||||
threading.Thread(
|
||||
target=_shutdown_quietly,
|
||||
args=(processor,),
|
||||
daemon=True,
|
||||
name="litellm-otel-destination-drain-straggler",
|
||||
).start()
|
||||
|
||||
def saturated(self) -> bool:
|
||||
"""Whether enough closes are outstanding that building another processor must wait.
|
||||
|
||||
The workers close in order and each close blocks for as long as its exporter
|
||||
does, so a collector that stopped answering would otherwise turn every new
|
||||
destination into one more batch thread parked behind them, for as long as the
|
||||
tenants keep rotating. Holding the count here rather than reading the queue
|
||||
keeps the two processors a worker is mid-close on in the total.
|
||||
"""
|
||||
with self._lock:
|
||||
return self._backlog >= self._capacity
|
||||
|
||||
def close(self, timeout: float | None = None) -> None:
|
||||
"""Retire the workers once they have closed everything already queued.
|
||||
|
||||
A proxy that rebuilds its telemetry builds another fan-out, so workers that
|
||||
outlive the one that started them are two more threads per reload, forever.
|
||||
|
||||
``timeout`` bounds how long the caller waits for that draining to finish. The
|
||||
workers are daemons, so whatever is still flushing when it expires is dropped
|
||||
by the interpreter rather than holding it open.
|
||||
"""
|
||||
with self._lock:
|
||||
if self._closed:
|
||||
return
|
||||
self._closed = True
|
||||
for _ in range(self._workers):
|
||||
self._pending.put(None)
|
||||
if timeout is None:
|
||||
return
|
||||
deadline: Final = time.monotonic() + timeout
|
||||
for worker in self._threads:
|
||||
worker.join(timeout=max(0.0, deadline - time.monotonic()))
|
||||
|
||||
def _drain_until_closed(self) -> None:
|
||||
while True:
|
||||
processor: SpanProcessor | None = self._pending.get() # rebind-ok: loop variable
|
||||
if processor is None:
|
||||
return
|
||||
_shutdown_quietly(processor)
|
||||
with self._lock:
|
||||
self._backlog -= 1
|
||||
|
||||
|
||||
_NO_ATTRIBUTES: Final[Mapping[str, AttributeValue]] = MappingProxyType({})
|
||||
_DB_SYSTEM_KEYS: Final = frozenset({DB.SYSTEM_NAME, DB.SYSTEM_LEGACY})
|
||||
# Keys on a database span that describe the proxy's own datastore: its host, its
|
||||
# port, and its schema.
|
||||
_DATASTORE_ENDPOINT_KEYS: Final = frozenset({Server.ADDRESS, Server.PORT, DB.NAMESPACE})
|
||||
# A span carrying one of these describes the tenant's own call (the model call, the
|
||||
# MCP call, the guardrail), so its error text is theirs to see. Every other span is
|
||||
# the proxy's own work, whose error text names the operator's infrastructure.
|
||||
_TENANT_OWNED_KEYS: Final = frozenset({GenAI.OPERATION_NAME, MCP.METHOD_NAME, LiteLLM.GUARDRAIL_NAME})
|
||||
_PROXY_ERROR_TEXT_KEYS: Final = frozenset({Error.MESSAGE, Error.MESSAGE_LEGACY})
|
||||
# A guardrail that never answered carries the exception it raised as its response,
|
||||
# which names the operator's guardrail endpoint. The second spelling is the legacy
|
||||
# status the request-level logger still maps.
|
||||
_GUARDRAIL_UNREACHABLE_STATUSES: Final = frozenset({"guardrail_failed_to_respond", "failure"})
|
||||
# Attribute prefixes the FastAPI instrumentor uses for headers the operator opted to
|
||||
# capture (``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_*``). The request
|
||||
# side carries the caller's bearer token verbatim.
|
||||
_CAPTURED_HEADER_PREFIXES: Final = ("http.request.header.", "http.response.header.")
|
||||
# The instrumentor stamps the request URL on the server span with its query string,
|
||||
# under the old convention and the new one, and litellm accepts a virtual key as a
|
||||
# ``?key=`` query parameter.
|
||||
_URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"})
|
||||
_URL_QUERY_KEY: Final = "url.query"
|
||||
|
||||
|
||||
class _TenantSpanView(ReadableSpan):
|
||||
"""A ``ReadableSpan`` view for one destination, leaving the operator's own span alone."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
inner: ReadableSpan,
|
||||
resource: Resource,
|
||||
attributes: Attributes,
|
||||
events: Sequence[Event],
|
||||
status: Status,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
name=inner.name,
|
||||
context=inner.context,
|
||||
parent=inner.parent,
|
||||
resource=resource,
|
||||
attributes=attributes,
|
||||
events=events,
|
||||
links=inner.links,
|
||||
kind=inner.kind,
|
||||
status=status,
|
||||
start_time=inner.start_time,
|
||||
end_time=inner.end_time,
|
||||
instrumentation_scope=inner.instrumentation_scope,
|
||||
)
|
||||
|
||||
|
||||
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return any(key in attributes for key in _DB_SYSTEM_KEYS)
|
||||
|
||||
|
||||
def _is_tenant_owned_span(attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return any(key in attributes for key in _TENANT_OWNED_KEYS)
|
||||
|
||||
|
||||
def _guardrail_unreachable(attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return attributes.get(LiteLLM.GUARDRAIL_STATUS) in _GUARDRAIL_UNREACHABLE_STATUSES
|
||||
|
||||
|
||||
def _tenant_visible(key: str, database: bool, owned: bool, unreachable_guardrail: bool) -> bool:
|
||||
if key.startswith(_CAPTURED_HEADER_PREFIXES) or key in (LiteLLMError.STACK_TRACE, _URL_QUERY_KEY):
|
||||
return False
|
||||
if database and key in _DATASTORE_ENDPOINT_KEYS:
|
||||
return False
|
||||
if unreachable_guardrail and key == LiteLLM.GUARDRAIL_RESPONSE:
|
||||
return False
|
||||
return owned or key not in _PROXY_ERROR_TEXT_KEYS
|
||||
|
||||
|
||||
def _without_query(key: str, value: AttributeValue) -> AttributeValue:
|
||||
if key not in _URL_KEYS or not isinstance(value, str):
|
||||
return value
|
||||
return value.partition("?")[0]
|
||||
|
||||
|
||||
def _same_attributes(kept: Mapping[str, AttributeValue], attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return len(kept) == len(attributes) and all(kept[key] is value for key, value in attributes.items())
|
||||
|
||||
|
||||
def _without_stack_trace(event: Event) -> Event:
|
||||
attributes: Final = event.attributes or _NO_ATTRIBUTES
|
||||
if ExceptionEvent.STACKTRACE not in attributes:
|
||||
return event
|
||||
return Event(
|
||||
name=event.name,
|
||||
attributes=MappingProxyType(
|
||||
{key: value for key, value in attributes.items() if key != ExceptionEvent.STACKTRACE}
|
||||
),
|
||||
timestamp=event.timestamp,
|
||||
)
|
||||
|
||||
|
||||
def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan:
|
||||
"""The view of ``span`` a tenant destination receives.
|
||||
|
||||
A span the tenant's own call produced keeps its error text. Every other span is
|
||||
the proxy's own work (the request root, auth, the database), and its error text,
|
||||
its events and its status description come off, since a Prisma failure there
|
||||
spells out the operator's Postgres endpoint. A database span loses that endpoint
|
||||
too, and a guardrail that failed to respond loses its response text, which is the
|
||||
exception it raised and names the operator's guardrail endpoint. Stack traces walk
|
||||
the operator's install and come off every span, as do the headers the operator
|
||||
captures on the server span, whose request side holds the caller's bearer token,
|
||||
and the query string of the request URL, which can hold the same key. The span
|
||||
itself stays, so the tenant still gets the whole trace tree.
|
||||
"""
|
||||
extra: Final = destination.resource_attributes
|
||||
attributes: Final = span.attributes or _NO_ATTRIBUTES
|
||||
database: Final = _is_database_span(attributes)
|
||||
owned: Final = _is_tenant_owned_span(attributes)
|
||||
unreachable: Final = _guardrail_unreachable(attributes)
|
||||
kept: Final = MappingProxyType(
|
||||
{
|
||||
key: _without_query(key, value)
|
||||
for key, value in attributes.items()
|
||||
if _tenant_visible(key, database, owned, unreachable)
|
||||
}
|
||||
)
|
||||
recorded: Final = span.events
|
||||
events: Final = tuple(_without_stack_trace(event) for event in recorded) if owned else ()
|
||||
unchanged: Final = owned and _same_attributes(kept, attributes) and all(a is b for a, b in zip(events, recorded))
|
||||
if not extra and unchanged:
|
||||
return span
|
||||
resource: Final = span.resource.merge(Resource(extra)) if extra else span.resource
|
||||
status: Final = span.status if owned else Status(span.status.status_code)
|
||||
return _TenantSpanView(span, resource, kept, events, status)
|
||||
|
||||
|
||||
class TenantFanOutSpanProcessor(SpanProcessor):
|
||||
"""Export every finished span to each destination this request resolved.
|
||||
|
||||
Destinations ride a request-scoped ``ContextVar`` set during auth, so concurrent
|
||||
requests stay isolated. The forwarded view keeps the original trace and parent
|
||||
ids, so the tenant gets the same tree the operator would have received.
|
||||
|
||||
Exactly one provider carries this processor, the one published as the OTel global
|
||||
(see :func:`attach_tenant_fan_out`). That provider is the only one every span
|
||||
passes through: the FastAPI server span, the auth span and the post-call database
|
||||
spans are emitted on the global, while a second v2 logger's provider sees only
|
||||
that logger's own gen-AI span. Attaching the fan-out per logger would hand a
|
||||
tenant a one-span trace whenever its backend is not the global one, and two
|
||||
copies of the model call whenever it is.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
processor_factory: 'Callable[["OtelDestination"], SpanProcessor | None] | None' = None,
|
||||
shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS,
|
||||
operator_sinks: frozenset[_SinkKey] = frozenset(),
|
||||
pending_drains: int = _MAX_PENDING_DRAINS,
|
||||
drain_pool: _DrainPool | None = None,
|
||||
) -> None:
|
||||
self._operator_sinks: Final = operator_sinks
|
||||
self._drain_seconds: Final = shutdown_drain_seconds
|
||||
self._lock: Final = threading.Condition()
|
||||
self._closed = False # guarded by ``_lock``: an unlocked read races the teardown it gates
|
||||
self._build: Final = processor_factory if processor_factory is not None else _destination_processor
|
||||
self._processors: OrderedDict[object, SpanProcessor] = OrderedDict() # mutable-ok: bounded LRU
|
||||
self._retired: OrderedDict[int, SpanProcessor] = OrderedDict() # mutable-ok: drains as exports finish
|
||||
self._exporting: dict[int, int] = {} # mutable-ok: per-processor in-flight export count
|
||||
self._drain: Final = drain_pool if drain_pool is not None else _DrainPool(capacity=pending_drains)
|
||||
|
||||
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
|
||||
return None
|
||||
|
||||
def on_end(self, span: ReadableSpan) -> None:
|
||||
suppressed: Final = suppressed_backends()
|
||||
for destination in request_destinations():
|
||||
if self._operator_already_writes(destination, suppressed):
|
||||
continue
|
||||
processor = self._acquire(destination) # rebind-ok: loop variable; pyright forbids Final in a loop
|
||||
if processor is None:
|
||||
continue
|
||||
try:
|
||||
processor.on_end(_for_destination(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)
|
||||
finally:
|
||||
self._release(processor)
|
||||
|
||||
def _operator_already_writes(self, destination: "OtelDestination", suppressed: frozenset[str]) -> bool:
|
||||
"""Whether the operator's own exporter is sending this span to the same account.
|
||||
|
||||
Only reachable under ``additive``, where nothing is suppressed: a team that
|
||||
names the operator's own project would otherwise have every span written
|
||||
there twice, once by the operator's exporter and once by the fan-out.
|
||||
"""
|
||||
return (
|
||||
destination.callback_name not in suppressed
|
||||
and _sink_key(destination.endpoint, destination.headers) in self._operator_sinks
|
||||
)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""Close every destination processor, once the spans in flight have landed.
|
||||
|
||||
``on_end`` runs on whichever thread ends a span and can reach this fan-out
|
||||
while the SDK is tearing the provider down, so closing blind would drop a
|
||||
trace mid-forward and would hand the next caller a fresh exporter nothing
|
||||
will ever close. Refusing new work and then waiting out the in-flight ones
|
||||
keeps both from happening. A straggler past the bound is retired instead of
|
||||
closed: the thread still exporting it closes it through the drain as soon as
|
||||
its export returns, so no span is dropped mid-forward.
|
||||
|
||||
Every close then goes to the drain rather than running here. Closing a
|
||||
destination processor flushes it over the network and the SDK joins its own
|
||||
worker with no timeout of its own, so one tenant collector that answers but
|
||||
never finishes a response would otherwise hold process teardown open for as
|
||||
long as it likes. The drain's workers are daemons, and the whole teardown
|
||||
shares one deadline.
|
||||
"""
|
||||
deadline: Final = time.monotonic() + self._drain_seconds
|
||||
with self._lock:
|
||||
self._closed = True
|
||||
self._lock.wait_for(lambda: not self._exporting, timeout=self._drain_seconds)
|
||||
live: Final = tuple((id(p), p) for p in (*self._processors.values(), *self._retired.values()))
|
||||
closing: Final = tuple(p for ident, p in live if ident not in self._exporting)
|
||||
self._processors.clear()
|
||||
self._retired = OrderedDict( # mutable-ok: the same bounded map, keeping only what is still exporting
|
||||
(ident, p) for ident, p in live if ident in self._exporting
|
||||
)
|
||||
for processor in closing:
|
||||
self._drain.submit(processor)
|
||||
self._drain.close(timeout=max(0.0, deadline - time.monotonic()))
|
||||
|
||||
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 (*self._processors.values(), *self._retired.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 deliverable(self, destinations: Iterable["OtelDestination"]) -> tuple["OtelDestination", ...]:
|
||||
"""The subset of ``destinations`` this fan-out can actually export to.
|
||||
|
||||
A destination whose exporter will not build (a protocol whose package is not
|
||||
installed, a malformed endpoint) has to be dropped before the request anchors
|
||||
it, not when its first span ends. By then the operator's own exporter has been
|
||||
told to hold that backend's spans back for this request, so dropping there
|
||||
loses the span outright instead of leaving it where it would have gone with no
|
||||
override at all.
|
||||
"""
|
||||
return tuple(destination for destination in destinations if self._buildable(destination))
|
||||
|
||||
def _buildable(self, destination: "OtelDestination") -> bool:
|
||||
"""Whether a processor for ``destination`` exists or can be built right now."""
|
||||
with self._lock:
|
||||
if self._closed:
|
||||
return False
|
||||
built: Final = self._cached_or_built_locked(destination, anchored=False)
|
||||
drained: Final = self._drainable_locked()
|
||||
for shed in drained:
|
||||
self._drain.submit(shed)
|
||||
return built is not None
|
||||
|
||||
def _acquire(self, destination: "OtelDestination") -> SpanProcessor | None:
|
||||
"""The processor for ``destination``, marked busy until ``_release``.
|
||||
|
||||
The build happens under the same lock that reads the cache, so a cold cache
|
||||
met by a burst of concurrent requests yields one exporter rather than one per
|
||||
thread with all but the winner shed. Building an exporter opens no connection,
|
||||
so the cost of holding the lock is a constructor, once per destination.
|
||||
"""
|
||||
with self._lock:
|
||||
if self._closed:
|
||||
return None
|
||||
processor: Final = self._cached_or_built_locked(destination, anchored=True)
|
||||
if processor is None:
|
||||
return None
|
||||
self._exporting[id(processor)] = self._exporting.get(id(processor), 0) + 1
|
||||
drained: Final = self._drainable_locked()
|
||||
for shed in drained:
|
||||
self._drain.submit(shed)
|
||||
return processor
|
||||
|
||||
def _cached_or_built_locked(self, destination: "OtelDestination", *, anchored: bool) -> SpanProcessor | None:
|
||||
"""The cached processor for ``destination``, or a new one if the drain can take it.
|
||||
|
||||
Every build past the cache cap sheds one processor into the drain, so while the
|
||||
shed ones are stuck closing against a collector that stopped answering, a
|
||||
destination that is not yet anchored is refused rather than parked behind them:
|
||||
``deliverable`` then leaves its spans with the operator's exporter until the
|
||||
drain catches up. One the request already anchored is rebuilt regardless. The
|
||||
operator's exporter has stood down for it, so refusing here would drop the span,
|
||||
and other tenants' auths can evict it in the meantime, with that eviction being
|
||||
what tips the drain over. Eviction holds while the drain is saturated, so such a
|
||||
rebuild costs the cache one entry rather than shedding another processor, and
|
||||
the total stays at one per destination in flight.
|
||||
"""
|
||||
key: Final = destination.cache_key()
|
||||
if (cached := self._processors.get(key)) is not None:
|
||||
self._processors.move_to_end(key)
|
||||
self._retire_overflow_locked()
|
||||
return cached
|
||||
if not anchored and self._drain.saturated():
|
||||
verbose_logger.debug("OTel V2 fan-out: drain saturated, not building for %s", destination.endpoint)
|
||||
return None
|
||||
return self._build_locked(destination, key)
|
||||
|
||||
def _build_locked(self, destination: "OtelDestination", key: object) -> SpanProcessor | None:
|
||||
built: Final = self._build(destination)
|
||||
if built is None:
|
||||
return None
|
||||
self._processors[key] = built
|
||||
self._retire_overflow_locked()
|
||||
return built
|
||||
|
||||
def _release(self, processor: SpanProcessor) -> None:
|
||||
with self._lock:
|
||||
remaining: Final = self._exporting.get(id(processor), 1) - 1
|
||||
if remaining > 0:
|
||||
self._exporting[id(processor)] = remaining
|
||||
else:
|
||||
self._exporting.pop(id(processor), None)
|
||||
if not self._exporting:
|
||||
self._lock.notify_all()
|
||||
drained: Final = self._drainable_locked()
|
||||
for retired in drained:
|
||||
self._drain.submit(retired)
|
||||
|
||||
def _retire_overflow_locked(self) -> None:
|
||||
"""Move the LRU processor out of the cache once it is past the cap, drain permitting.
|
||||
|
||||
Eviction is what feeds the drain, and a destination a request already anchored
|
||||
is rebuilt on its next span, which would shed another one. While the shed ones
|
||||
are stuck closing against a collector that stopped answering, evicting would
|
||||
churn the cache at one more processor, and one more batch thread, per span.
|
||||
Holding above the cap instead keeps the total at one processor per destination
|
||||
in flight, since ``deliverable`` anchors no new destination while the drain is
|
||||
saturated. Once it has room again, every hit and build trims one entry.
|
||||
"""
|
||||
if len(self._processors) <= _MAX_CACHED_DESTINATION_PROCESSORS or self._drain.saturated():
|
||||
return
|
||||
_, evicted = self._processors.popitem(last=False)
|
||||
self._retired[id(evicted)] = evicted
|
||||
|
||||
def _drainable_locked(self) -> tuple[SpanProcessor, ...]:
|
||||
"""Retired processors no thread is exporting through, removed from the list.
|
||||
|
||||
``on_end`` holds a processor across an export, so closing an evicted one there
|
||||
drops the span it is holding. A retiree is out of the cache and can never be
|
||||
handed out again, so once its export count reaches zero it stays there.
|
||||
"""
|
||||
idle: Final = tuple(key for key in self._retired if self._exporting.get(key, 0) == 0)
|
||||
return tuple(self._retired.pop(key) for key in idle)
|
||||
|
||||
|
||||
def _destination_processor(destination: "OtelDestination") -> SpanProcessor | None:
|
||||
"""A batching OTLP processor aimed at ``destination``, or ``None`` if unbuildable.
|
||||
|
||||
A protocol that resolves to a headerless exporter is unbuildable too: the
|
||||
console fallback would swallow the tenant's credentials and print its spans to
|
||||
the proxy's stdout while the operator's exporter stands down for them.
|
||||
"""
|
||||
kind: Final = destination.protocol or "otlp_http"
|
||||
if exporter_transport(kind) == "headerless":
|
||||
verbose_logger.debug("OTel V2 fan-out: no OTLP transport for protocol %r at %s", kind, destination.endpoint)
|
||||
return None
|
||||
try:
|
||||
spec: Final = ExporterSpec(
|
||||
kind=kind,
|
||||
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.
|
||||
|
||||
Wrapping is the only place this works: ``SynchronousMultiSpanProcessor.on_end``
|
||||
ignores return values, so a sibling processor can never veto the export.
|
||||
|
||||
Under ``additive`` mode nothing is suppressed, so the wrapper passes every span
|
||||
straight through and the operator keeps its copy.
|
||||
"""
|
||||
|
||||
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 suppressed_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 +1009,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 +1018,13 @@ 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`` wraps each owned exporter so a request that pointed that
|
||||
backend at a key's or team's own account skips it. Every v2 logger's provider
|
||||
wants it, since any of them may own the overridden backend; delivering to the
|
||||
tenant is a separate job, done once by :func:`attach_tenant_fan_out`. 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,15 +1041,107 @@ def build_tracer_provider(
|
|||
if spec.requires_headers and not spec.headers:
|
||||
continue
|
||||
exp = _exporter_from_spec(spec)
|
||||
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(
|
||||
_processor_for(
|
||||
exp,
|
||||
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
|
||||
)
|
||||
_OverriddenBackendFilter(processor, owner) if tenant_overrides and owner is not None else processor
|
||||
)
|
||||
return provider
|
||||
|
||||
|
||||
_FAN_OUT_ATTACH_LOCK: Final = threading.Lock()
|
||||
|
||||
|
||||
def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Config) -> None:
|
||||
"""Give ``provider`` the fan-out that delivers spans to key/team destinations.
|
||||
|
||||
Called on the one provider published as the OTel global, and idempotent so a
|
||||
second publish (a test, a re-initialized proxy) cannot double-export. Concurrent
|
||||
first calls (requests racing to anchor before any publish) serialize on one lock
|
||||
so exactly one fan-out lands. ``configs`` name the operator's own exporters, one
|
||||
config per v2 logger since each keeps its own provider and still writes its
|
||||
account, so an additive destination pointing at any of them is delivered once
|
||||
rather than twice.
|
||||
"""
|
||||
with _FAN_OUT_ATTACH_LOCK:
|
||||
if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)):
|
||||
return
|
||||
provider.add_span_processor(TenantFanOutSpanProcessor(operator_sinks=operator_sink_keys(*configs)))
|
||||
|
||||
|
||||
def deliverable_destinations(
|
||||
destinations: Iterable["OtelDestination"],
|
||||
provider: trace.TracerProvider | None = None,
|
||||
) -> tuple["OtelDestination", ...]:
|
||||
"""The destinations a request can anchor, given what is published to carry them.
|
||||
|
||||
Anchoring a destination is what tells the operator's own exporter to stand down
|
||||
for that backend, so one nothing can deliver has to be dropped here: with no
|
||||
fan-out attached, or with an exporter that will not build, the request keeps
|
||||
exactly the routing it would have had without any override.
|
||||
"""
|
||||
fan_out: Final = next(
|
||||
(
|
||||
processor
|
||||
for processor in _attached_processors(provider if provider is not None else trace.get_tracer_provider())
|
||||
if isinstance(processor, TenantFanOutSpanProcessor)
|
||||
),
|
||||
None,
|
||||
)
|
||||
return fan_out.deliverable(destinations) if fan_out is not None else ()
|
||||
|
||||
|
||||
def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]:
|
||||
"""The accounts the operator's own exporters write to, in destination terms.
|
||||
|
||||
Every v2 logger's config counts, since each logger exports through its own
|
||||
provider. An exporter with no endpoint of its own resolves one from the
|
||||
environment at export time, so it has no comparable identity and is left out,
|
||||
and so is one that never reaches the wire: a console kind ignores the endpoint,
|
||||
and a header-gated spec with no credentials is skipped when the provider is built.
|
||||
"""
|
||||
return frozenset(
|
||||
key
|
||||
for config in configs
|
||||
for spec in config.exporters
|
||||
if _exports_to_the_wire(spec) and (key := _sink_key(spec.endpoint, parse_headers(spec.headers))) is not None
|
||||
)
|
||||
|
||||
|
||||
def _exports_to_the_wire(spec: ExporterSpec) -> bool:
|
||||
"""Whether ``build_tracer_provider`` gives ``spec`` an exporter that sends OTLP."""
|
||||
return exporter_transport(spec.kind) != "headerless" and not (spec.requires_headers and not spec.headers)
|
||||
|
||||
|
||||
def _sink_key(endpoint: str | None, headers: Mapping[str, str]) -> "_SinkKey | None":
|
||||
"""The account an exporter writes to, or ``None`` when it has no fixed one.
|
||||
|
||||
Normalized on the three counts that make one account look like two: the operator's
|
||||
spec carries the signal path a tenant destination leaves for the exporter to
|
||||
append, header names survive one round trip lowercased and the other not, and one
|
||||
credential answers to more than one name (see :data:`_CREDENTIAL_ALIASES`).
|
||||
"""
|
||||
normalized: Final = _otlp_traces_endpoint(endpoint)
|
||||
if normalized is None:
|
||||
return None
|
||||
return (normalized, tuple(sorted((_credential_name(name), value) for name, value in headers.items())))
|
||||
|
||||
|
||||
def _credential_name(header: str) -> str:
|
||||
"""The credential a header carries, under whichever name the backend spells it."""
|
||||
normalized: Final = header.strip().lower().replace("-", "_")
|
||||
return _CREDENTIAL_ALIASES.get(normalized, normalized)
|
||||
|
||||
|
||||
def _attached_processors(provider: trace.TracerProvider) -> "tuple[SpanProcessor, ...]":
|
||||
"""The processors already on ``provider``, or empty when the SDK hides them."""
|
||||
multi: Final = getattr(provider, "_active_span_processor", None)
|
||||
return tuple(getattr(multi, "_span_processors", ()))
|
||||
|
||||
|
||||
def get_tracer(provider: TracerProvider, name: str = "litellm") -> Tracer:
|
||||
# Stamp the instrumentation scope with the LiteLLM package version so every
|
||||
# emitted span carries a deterministic ``scope.version`` (the standard OTel
|
||||
|
|
|
|||
|
|
@ -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 destination_backends
|
||||
from litellm.integrations.otel.plumbing.providers import (
|
||||
build_tracer_provider,
|
||||
exporter_transport,
|
||||
|
|
@ -231,10 +232,21 @@ 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.
|
||||
"""
|
||||
# A backend with a destination is delivered by the fan-out processor, which
|
||||
# carries the whole trace and already carries this tenant's credentials and
|
||||
# service name. 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.
|
||||
if self._callback_name is not None and self._callback_name in destination_backends():
|
||||
return TenantRoute(tracer=default, detached=False)
|
||||
credential_headers: Final = 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:
|
||||
tenant_account: Final = bool(credential_headers) or bool(project_headers)
|
||||
# A service name on its own only relabels the operator's own backend, so moving
|
||||
# the span to a second provider for it while some other backend has a
|
||||
# destination would drop the model call out of the trace the fan-out delivers.
|
||||
# The destination stamps the same service name itself.
|
||||
if not tenant_account and (service_name is None or destination_backends()):
|
||||
return TenantRoute(tracer=default, detached=False)
|
||||
# A fixed per-integration region endpoint (New Relic us/eu), never a
|
||||
# caller-supplied host; ``None`` keeps the preset's own endpoint.
|
||||
|
|
@ -255,7 +267,7 @@ class TenantTracerCache:
|
|||
_shutdown_provider(evicted)
|
||||
return TenantRoute(
|
||||
tracer=get_tracer(provider, self._tracer_name),
|
||||
detached=bool(project_headers) or bool(credential_headers),
|
||||
detached=tenant_account,
|
||||
provider=provider,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -26,10 +26,12 @@ class _ArizeSettings(BaseSettings):
|
|||
def arize_preset(
|
||||
*,
|
||||
config_overrides: OpenTelemetryV2Config | None = None,
|
||||
allow_missing_credentials: bool = False,
|
||||
) -> OpenTelemetryV2Config:
|
||||
base: Final = config_overrides or OpenTelemetryV2Config()
|
||||
mappers: Final = ensure_mappers(base.mapper_names, "openinference")
|
||||
arize_cfg: Final = _V1ArizeLogger.get_arize_config()
|
||||
headers: Final = _arize_headers(arize_cfg)
|
||||
base: Final = config_overrides or OpenTelemetryV2Config()
|
||||
return base.model_copy(
|
||||
update={
|
||||
"exporters": [
|
||||
|
|
@ -41,7 +43,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 {}),
|
||||
|
|
|
|||
|
|
@ -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 and
|
||||
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: ...
|
||||
|
|
|
|||
152
litellm/integrations/otel/presets/destinations.py
Normal file
152
litellm/integrations/otel/presets/destinations.py
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
"""Map a key's or team's callback vars to the OTLP destination its traces export to.
|
||||
|
||||
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 and transport need a per-backend rule.
|
||||
"""
|
||||
|
||||
import os
|
||||
from collections.abc import Callable, Mapping
|
||||
from functools import lru_cache
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host
|
||||
from litellm.types.utils import StandardCallbackDynamicParams
|
||||
|
||||
#: An endpoint plus the OTLP transport to reach it with, or ``None`` when the backend
|
||||
#: names no destination. The transport is ``None`` where the backend has only one.
|
||||
_Destination = tuple[str, str | None]
|
||||
|
||||
|
||||
@lru_cache(maxsize=128)
|
||||
def _warn_host_not_allowlisted(host: str) -> None:
|
||||
"""Cached so one misconfigured team logs once rather than once per request."""
|
||||
verbose_logger.warning(
|
||||
"OTel V2: not exporting to key/team Langfuse host '%s'. Add it to "
|
||||
"litellm_settings.provider_url_destination_allowed_hosts to permit it",
|
||||
host,
|
||||
)
|
||||
|
||||
|
||||
def _langfuse_destination(params: StandardCallbackDynamicParams) -> "_Destination | None":
|
||||
"""The tenant's own Langfuse host, else the operator's, else Langfuse US cloud.
|
||||
|
||||
A host the tenant named has to be allowlisted by the operator, the same way a
|
||||
URL-valued ``model`` is: anyone who can mint a key can write it, and it becomes an
|
||||
endpoint the proxy posts the request's whole trace to, carrying the tenant's own
|
||||
credentials. The operator's own ``LANGFUSE_HOST`` is not checked, since an internal
|
||||
collector there is a deployment choice.
|
||||
"""
|
||||
from litellm.integrations.langfuse.langfuse_otel import (
|
||||
LANGFUSE_CLOUD_US_ENDPOINT,
|
||||
LangfuseOtelLogger,
|
||||
)
|
||||
|
||||
tenant_host: Final = params.get("langfuse_host") or None
|
||||
host: Final = tenant_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, None)
|
||||
normalized: Final = host if host.startswith("http") else f"https://{host}"
|
||||
endpoint: Final = f"{normalized.rstrip('/')}/api/public/otel"
|
||||
if tenant_host is None:
|
||||
return (endpoint, None)
|
||||
if not is_url_destination_allowed_by_host(endpoint, litellm.provider_url_destination_allowed_hosts):
|
||||
_warn_host_not_allowlisted(host)
|
||||
return None
|
||||
return (endpoint, None)
|
||||
|
||||
|
||||
def _arize_destination(params: StandardCallbackDynamicParams) -> "_Destination | None":
|
||||
from litellm.integrations.arize.arize import ArizeLogger
|
||||
|
||||
config: Final = ArizeLogger.get_arize_config()
|
||||
return (config.endpoint, config.protocol)
|
||||
|
||||
|
||||
def _weave_destination(params: StandardCallbackDynamicParams) -> "_Destination | None":
|
||||
from litellm.integrations.weave.weave_otel import weave_otel_endpoint
|
||||
|
||||
return (weave_otel_endpoint(os.environ.get("WANDB_HOST")), None)
|
||||
|
||||
|
||||
def _newrelic_destination(params: StandardCallbackDynamicParams) -> "_Destination | None":
|
||||
from litellm.integrations.otel.presets.newrelic import newrelic_dynamic_endpoint
|
||||
|
||||
endpoint: Final = newrelic_dynamic_endpoint(params)
|
||||
return (endpoint, None) if endpoint else None
|
||||
|
||||
|
||||
#: Callback name -> destination 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.
|
||||
_DESTINATION_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], "_Destination | None"]]] = (
|
||||
MappingProxyType(
|
||||
{
|
||||
"langfuse_otel": _langfuse_destination,
|
||||
"arize": _arize_destination,
|
||||
"weave_otel": _weave_destination,
|
||||
"newrelic": _newrelic_destination,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
#: Headers a destination must carry to authenticate. Several dynamic-header builders
|
||||
#: gate each credential independently, so a half-configured backend yields a non-empty
|
||||
#: but unusable header set; accepting it would suppress the operator's own exporter and
|
||||
#: send the request's whole trace where it cannot be stored.
|
||||
_REQUIRED_HEADERS_BY_CALLBACK: Final[Mapping[str, frozenset[str]]] = MappingProxyType(
|
||||
{
|
||||
"langfuse_otel": frozenset({"Authorization"}),
|
||||
"arize": frozenset({"arize-space-id", "api_key"}),
|
||||
"weave_otel": frozenset({"Authorization", "project_id"}),
|
||||
"newrelic": frozenset({"api-key"}),
|
||||
}
|
||||
)
|
||||
|
||||
_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(_DESTINATION_BY_CALLBACK) & frozenset(DYNAMIC_HEADERS_BY_CALLBACK)
|
||||
|
||||
|
||||
def destination_for(
|
||||
callback_name: str,
|
||||
params: StandardCallbackDynamicParams,
|
||||
service_name: str | None = None,
|
||||
) -> 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. ``service_name`` is the key's or
|
||||
team's ``otel_service_name``, which the per-request tracer route applies when the
|
||||
backend is not overridden and the destination has to apply once it is.
|
||||
"""
|
||||
from litellm.integrations.otel.presets import DYNAMIC_HEADERS_BY_CALLBACK
|
||||
|
||||
header_builder: Final = DYNAMIC_HEADERS_BY_CALLBACK.get(callback_name)
|
||||
destination_builder: Final = _DESTINATION_BY_CALLBACK.get(callback_name)
|
||||
if header_builder is None or destination_builder is None:
|
||||
return None
|
||||
headers: Final = header_builder(params)
|
||||
if not headers or not _REQUIRED_HEADERS_BY_CALLBACK[callback_name] <= frozenset(headers):
|
||||
return None
|
||||
resolved: Final = destination_builder(params)
|
||||
if resolved is None:
|
||||
return None
|
||||
endpoint, protocol = resolved
|
||||
return OtelDestination(
|
||||
endpoint=endpoint,
|
||||
headers=MappingProxyType(dict(headers)), # mutable-ok: MappingProxyType needs a concrete mapping to wrap
|
||||
resource_attributes=MappingProxyType({"service.name": service_name}) if service_name else _NO_ATTRS,
|
||||
callback_name=callback_name,
|
||||
protocol=protocol,
|
||||
)
|
||||
|
|
@ -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,
|
||||
}
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@
|
|||
from collections.abc import Iterable
|
||||
from typing import Final
|
||||
|
||||
from litellm.integrations.otel.model.config import ExporterOwner, 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 +17,32 @@ 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 not is_unconfigured_placeholder(spec)),
|
||||
ExporterSpec(owner=owner, requires_headers=True),
|
||||
)
|
||||
|
||||
|
||||
def is_unconfigured_placeholder(spec: "ExporterSpec") -> bool:
|
||||
"""Whether ``spec`` is the one ``_normalize`` folds in when nothing was configured.
|
||||
|
||||
No field set is what says the operator asked for nothing: an exporter they did
|
||||
configure survives, even ``OTEL_EXPORTER=console`` whose value matches the default,
|
||||
and so does the gated spec this module appends, which would otherwise eat itself
|
||||
when one preset layers onto another.
|
||||
"""
|
||||
return not spec.model_fields_set
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -117,6 +117,14 @@ def _get_weave_authorization_header(api_key: str) -> str:
|
|||
return f"Basic {auth_header}"
|
||||
|
||||
|
||||
def weave_otel_endpoint(host: str | None) -> str:
|
||||
"""The OTLP traces endpoint for a self-managed ``host``, else Weave cloud."""
|
||||
if not host:
|
||||
return WEAVE_BASE_URL + WEAVE_OTEL_ENDPOINT
|
||||
normalized: Final = host if host.startswith("http") else f"https://{host}"
|
||||
return normalized.rstrip("/") + WEAVE_OTEL_ENDPOINT
|
||||
|
||||
|
||||
def get_weave_otel_config() -> WeaveOtelConfig:
|
||||
"""
|
||||
Retrieves the Weave OpenTelemetry configuration based on environment variables.
|
||||
|
|
@ -134,7 +142,6 @@ def get_weave_otel_config() -> WeaveOtelConfig:
|
|||
"""
|
||||
api_key: Final = os.getenv("WANDB_API_KEY")
|
||||
project_id: Final = os.getenv("WANDB_PROJECT_ID")
|
||||
host = os.getenv("WANDB_HOST")
|
||||
|
||||
if not api_key:
|
||||
raise ValueError("WANDB_API_KEY must be set for Weave OpenTelemetry integration.")
|
||||
|
|
@ -144,15 +151,8 @@ def get_weave_otel_config() -> WeaveOtelConfig:
|
|||
"WANDB_PROJECT_ID must be set for Weave OpenTelemetry integration. Format: <entity>/<project_name>"
|
||||
)
|
||||
|
||||
if host:
|
||||
if not host.startswith("http"):
|
||||
host = "https://" + host
|
||||
# Self-managed instances use a different path
|
||||
endpoint = host.rstrip("/") + WEAVE_OTEL_ENDPOINT
|
||||
verbose_logger.debug("Using Weave OTEL endpoint from host: %s", endpoint)
|
||||
else:
|
||||
endpoint = WEAVE_BASE_URL + WEAVE_OTEL_ENDPOINT
|
||||
verbose_logger.debug("Using Weave cloud endpoint: %s", endpoint)
|
||||
endpoint: Final = weave_otel_endpoint(os.getenv("WANDB_HOST"))
|
||||
verbose_logger.debug("Using Weave OTEL endpoint: %s", endpoint)
|
||||
|
||||
# Weave uses Basic auth with format: api:<WANDB_API_KEY>
|
||||
auth_header: Final = _get_weave_authorization_header(api_key=api_key)
|
||||
|
|
|
|||
|
|
@ -200,6 +200,7 @@ if TYPE_CHECKING:
|
|||
from mcp.types import EmbeddedResource, ImageContent, TextContent
|
||||
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2
|
||||
from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config
|
||||
from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig
|
||||
try:
|
||||
from litellm_enterprise.enterprise_callbacks.callback_controls import (
|
||||
|
|
@ -4800,31 +4801,83 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
|
|||
|
||||
Returns ``None`` when V2 is off OR when there's no preset registered for
|
||||
``callback_name`` — callers should then fall through to the legacy path.
|
||||
|
||||
A preset that needs operator credentials it cannot find is allowed to build
|
||||
only when this request has a key/team destination for that backend and another
|
||||
V2 logger is already registered to carry the fan-out. The resulting logger keeps
|
||||
only its credential-gated exporter, while the registered logger owns operator
|
||||
delivery. Without that carrier, a preset that raises or that ends up with nothing
|
||||
but its gated exporter and the default console placeholder returns ``None``, so the
|
||||
caller falls through to the legacy path exactly as before V2 landed.
|
||||
"""
|
||||
from litellm.integrations.otel.model.config import is_otel_v2_enabled
|
||||
|
||||
if not is_otel_v2_enabled():
|
||||
return None
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2, build_otel_v2_logger
|
||||
from litellm.integrations.otel.plumbing.context import destination_backends
|
||||
from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
|
||||
|
||||
preset_fn: Final = PRESET_BY_CALLBACK.get(callback_name)
|
||||
if preset_fn is None:
|
||||
return None
|
||||
serves_a_destination: Final = callback_name in destination_backends()
|
||||
has_v2_logger: Final = any(isinstance(callback, OpenTelemetryV2) for callback in _in_memory_loggers)
|
||||
carried: Final = serves_a_destination and has_v2_logger
|
||||
for callback in _in_memory_loggers:
|
||||
if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name:
|
||||
if (
|
||||
isinstance(callback, OpenTelemetryV2)
|
||||
and getattr(callback, "callback_name", None) == callback_name
|
||||
and (serves_a_destination or not _exports_nowhere(callback.config))
|
||||
):
|
||||
return callback
|
||||
try:
|
||||
config: Final = preset_fn()
|
||||
built: Final = preset_fn(allow_missing_credentials=carried)
|
||||
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
|
||||
gated: Final = _is_credential_gated(built)
|
||||
if gated and not carried and not _has_operator_exporter(built):
|
||||
return None
|
||||
config: Final = _only_the_gated_exporter(built) if gated and carried else built
|
||||
if _exports_nowhere(config):
|
||||
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
|
||||
|
||||
|
||||
def _exports_nowhere(config: "OpenTelemetryV2Config") -> bool:
|
||||
"""Whether every exporter in ``config`` is waiting on credentials it never got."""
|
||||
return all(_is_gated(spec) for spec in config.exporters)
|
||||
|
||||
|
||||
def _is_credential_gated(config: "OpenTelemetryV2Config") -> bool:
|
||||
"""Whether the preset built without the operator's own credentials for its backend."""
|
||||
return any(_is_gated(spec) for spec in config.exporters)
|
||||
|
||||
|
||||
def _has_operator_exporter(config: "OpenTelemetryV2Config") -> bool:
|
||||
"""Whether the operator configured somewhere real to export, beyond the default console placeholder."""
|
||||
from litellm.integrations.otel.presets.utils import is_unconfigured_placeholder
|
||||
|
||||
return any(not _is_gated(spec) and not is_unconfigured_placeholder(spec) for spec in config.exporters)
|
||||
|
||||
|
||||
def _only_the_gated_exporter(config: "OpenTelemetryV2Config") -> "OpenTelemetryV2Config":
|
||||
return config.model_copy(
|
||||
update={"exporters": [spec for spec in config.exporters if _is_gated(spec)]} # mutable-ok: model_copy update
|
||||
)
|
||||
|
||||
|
||||
def _is_gated(spec: "ExporterSpec") -> bool:
|
||||
return spec.requires_headers and not spec.headers
|
||||
|
||||
|
||||
def _maybe_auto_initialize_arize_phoenix(_in_memory_loggers: list[CustomLogger]) -> None:
|
||||
"""
|
||||
Auto-initialize ArizePhoenixLogger when Phoenix env vars are detected.
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from typing import TYPE_CHECKING, Any, Final, Protocol
|
|||
|
||||
import httpx
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import AnyUrl, ConfigDict
|
||||
from pydantic import AnyUrl, ConfigDict, TypeAdapter, ValidationError
|
||||
from starlette.requests import Request as StarletteRequest
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.types import Message, Receive, Scope, Send
|
||||
|
|
@ -107,9 +107,9 @@ _MAX_STATEFUL_SESSIONS_PER_OWNER: Final = 100
|
|||
# prevents an authenticated client from forcing the proxy to buffer an
|
||||
# arbitrarily large body just to make a routing decision.
|
||||
_MCP_ROUTING_PEEK_MAX_BYTES: Final = 4096
|
||||
# ASGI scope key holding the tracing span of the request carrying an MCP
|
||||
# message, written on the request task and read back by the message handler.
|
||||
# ASGI scope keys carrying OTel request state into a stateful MCP message handler.
|
||||
_MCP_TRANSPORT_SPAN_SCOPE_KEY: Final = "litellm_otel_transport_span"
|
||||
_MCP_DESTINATIONS_SCOPE_KEY: Final = "litellm_otel_request_destinations"
|
||||
|
||||
|
||||
def _invalidate_byok_cred_cache(user_id: str, server_id: str) -> None:
|
||||
|
|
@ -327,18 +327,17 @@ def _otel_publish_transport_span_on_scope(scope: Scope) -> None:
|
|||
scope[_MCP_TRANSPORT_SPAN_SCOPE_KEY] = span
|
||||
|
||||
|
||||
def _otel_transport_span_from_message(req_ctx: object) -> object:
|
||||
"""The tracing span of the HTTP request that carried this MCP message.
|
||||
|
||||
Read off that request's ASGI scope, reached through the ``Request`` the
|
||||
streamable-HTTP transport attaches to each message, so it is this message's
|
||||
transport and not whichever request happens to have touched the session last.
|
||||
Returns whatever the scope holds; the otel plumbing validates it."""
|
||||
def _otel_value_from_message_scope(req_ctx: object, key: str) -> object:
|
||||
request: Final = getattr(req_ctx, "request", None)
|
||||
scope: Final = getattr(request, "scope", None)
|
||||
if not isinstance(scope, Mapping):
|
||||
return None
|
||||
return scope.get(_MCP_TRANSPORT_SPAN_SCOPE_KEY)
|
||||
return scope.get(key)
|
||||
|
||||
|
||||
def _otel_transport_span_from_message(req_ctx: object) -> object:
|
||||
"""The tracing span of the HTTP request that carried this MCP message."""
|
||||
return _otel_value_from_message_scope(req_ctx, _MCP_TRANSPORT_SPAN_SCOPE_KEY)
|
||||
|
||||
|
||||
def _otel_set_mcp_transport_span(span: object) -> object:
|
||||
|
|
@ -371,6 +370,44 @@ def _otel_reset_mcp_transport_span(token: object) -> None:
|
|||
return
|
||||
|
||||
|
||||
def _otel_publish_request_destinations_on_scope(scope: Scope) -> None:
|
||||
try:
|
||||
from litellm.integrations.otel.plumbing.context import request_destinations
|
||||
|
||||
scope[_MCP_DESTINATIONS_SCOPE_KEY] = request_destinations()
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
|
||||
def _otel_set_mcp_request_destinations(req_ctx: object) -> object:
|
||||
destinations: Final = _otel_value_from_message_scope(req_ctx, _MCP_DESTINATIONS_SCOPE_KEY)
|
||||
if not isinstance(destinations, tuple):
|
||||
return None
|
||||
try:
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.integrations.otel.plumbing.context import set_request_destinations
|
||||
|
||||
destination_adapter: Final[TypeAdapter[tuple[OtelDestination, ...]]] = TypeAdapter(
|
||||
tuple[OtelDestination, ...],
|
||||
config=ConfigDict(revalidate_instances="always"),
|
||||
)
|
||||
validated_destinations: Final = destination_adapter.validate_python(destinations, strict=True)
|
||||
return set_request_destinations(validated_destinations)
|
||||
except (ImportError, ValidationError):
|
||||
return None
|
||||
|
||||
|
||||
def _otel_reset_mcp_request_destinations(token: object) -> None:
|
||||
if token is None:
|
||||
return
|
||||
try:
|
||||
from litellm.integrations.otel.plumbing.context import reset_request_destinations
|
||||
|
||||
reset_request_destinations(token)
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
|
||||
def _proxy_exception_to_http_exception(exc: ProxyException) -> HTTPException:
|
||||
"""Map a ``ProxyException`` to an ``HTTPException`` that preserves its real
|
||||
status code and headers.
|
||||
|
|
@ -762,10 +799,12 @@ if MCP_AVAILABLE:
|
|||
_session_reset_token = active_mcp_session_var.set(req_ctx.session)
|
||||
_trace_token = None
|
||||
_transport_token = None
|
||||
_destinations_token = None
|
||||
|
||||
try:
|
||||
_trace_token = _otel_set_mcp_trace_carrier(_mcp_meta_trace_carrier(req_ctx))
|
||||
_transport_token = _otel_set_mcp_transport_span(_otel_transport_span_from_message(req_ctx))
|
||||
_destinations_token = _otel_set_mcp_request_destinations(req_ctx)
|
||||
# Get user authentication from context variable
|
||||
(
|
||||
user_api_key_auth,
|
||||
|
|
@ -822,6 +861,7 @@ if MCP_AVAILABLE:
|
|||
# This prevents the HTTP stream from failing and allows the client to get a response
|
||||
return []
|
||||
finally:
|
||||
_otel_reset_mcp_request_destinations(_destinations_token)
|
||||
_otel_reset_mcp_transport_span(_transport_token)
|
||||
_otel_reset_mcp_trace_carrier(_trace_token)
|
||||
if _session_reset_token is not None:
|
||||
|
|
@ -1006,10 +1046,12 @@ if MCP_AVAILABLE:
|
|||
_session_reset_token = active_mcp_session_var.set(req_ctx.session)
|
||||
_trace_token = None
|
||||
_transport_token = None
|
||||
_destinations_token = None
|
||||
|
||||
try:
|
||||
_trace_token = _otel_set_mcp_trace_carrier(_mcp_meta_trace_carrier(req_ctx))
|
||||
_transport_token = _otel_set_mcp_transport_span(_otel_transport_span_from_message(req_ctx))
|
||||
_destinations_token = _otel_set_mcp_request_destinations(req_ctx)
|
||||
# Validate arguments
|
||||
(
|
||||
user_api_key_auth,
|
||||
|
|
@ -1147,6 +1189,7 @@ if MCP_AVAILABLE:
|
|||
|
||||
return response
|
||||
finally:
|
||||
_otel_reset_mcp_request_destinations(_destinations_token)
|
||||
_otel_reset_mcp_transport_span(_transport_token)
|
||||
_otel_reset_mcp_trace_carrier(_trace_token)
|
||||
if _session_reset_token is not None:
|
||||
|
|
@ -4397,6 +4440,7 @@ if MCP_AVAILABLE:
|
|||
|
||||
async def _dispatch() -> None:
|
||||
_otel_publish_transport_span_on_scope(scope)
|
||||
_otel_publish_request_destinations_on_scope(scope)
|
||||
auth_user: Final = _set_or_update_auth_context(
|
||||
user_api_key_auth=user_api_key_auth,
|
||||
mcp_auth_header=mcp_auth_header,
|
||||
|
|
|
|||
|
|
@ -2845,6 +2845,43 @@ async def _authorize_authenticated_request(
|
|||
|
||||
|
||||
@tracer.wrap()
|
||||
def _seed_request_destinations(user_api_key_dict: UserAPIKeyAuth, request: Request | None = None) -> 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.
|
||||
|
||||
``request`` carries the headers, so a backend this request disabled with
|
||||
``x-litellm-disable-callbacks`` resolves to no destination.
|
||||
|
||||
Only destinations the published fan-out can build are anchored. Anchoring one is
|
||||
what tells the operator's exporter to hold that backend's spans back under
|
||||
``override``, so an unbuildable one would leave the span with nowhere to go.
|
||||
|
||||
The ``postgres`` spans under ``auth`` close before this runs, because they are the
|
||||
reads that resolve the identity being read here. They never reach the tenant's
|
||||
account, and they are never withheld from the operator's backend, whichever mode
|
||||
is set.
|
||||
"""
|
||||
try:
|
||||
from litellm.integrations.otel.logger import fan_out_provider
|
||||
from litellm.integrations.otel.plumbing.context import set_request_destinations
|
||||
from litellm.integrations.otel.plumbing.providers import deliverable_destinations
|
||||
from litellm.proxy.litellm_pre_call_utils import (
|
||||
resolve_tenant_otel_destinations,
|
||||
)
|
||||
|
||||
set_request_destinations(
|
||||
deliverable_destinations(
|
||||
resolve_tenant_otel_destinations(user_api_key_dict, _safe_get_request_headers(request)),
|
||||
fan_out_provider(),
|
||||
)
|
||||
)
|
||||
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 +2928,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, request)
|
||||
|
||||
# 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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
@ -26,6 +27,7 @@ from litellm.constants import (
|
|||
SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
|
||||
SESSION_ID_GENERATED_METADATA_KEY,
|
||||
SESSION_ID_OMITTED_METADATA_KEY,
|
||||
X_LITELLM_DISABLE_CALLBACKS,
|
||||
)
|
||||
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
|
||||
from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
|
||||
|
|
@ -157,6 +159,7 @@ from litellm.types.utils import (
|
|||
CustomPricingLiteLLMParams,
|
||||
LlmProviders,
|
||||
ProviderSpecificHeader,
|
||||
StandardCallbackDynamicParams,
|
||||
StandardLoggingUserAPIKeyMetadata,
|
||||
SupportedCacheControls,
|
||||
)
|
||||
|
|
@ -170,6 +173,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 +978,142 @@ 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()
|
||||
|
||||
|
||||
_NO_REQUEST_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
|
||||
|
||||
|
||||
def _dynamically_disabled_backends(
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
request_headers: Mapping[str, str] | None,
|
||||
) -> frozenset[str]:
|
||||
"""The callbacks this request turned off, read the way dispatch reads them.
|
||||
|
||||
Same sources, precedence, and premium gate ``EnterpriseCallbackControls`` applies
|
||||
before it skips a callback: the ``x-litellm-disable-callbacks`` header wins over the
|
||||
key's stored list, team settings are not a source, and a non-premium proxy honours
|
||||
neither. A destination has to agree with that decision, or a backend the key turned
|
||||
off would still be exported to, now through the fan-out instead of the callback.
|
||||
"""
|
||||
from litellm.proxy.proxy_server import premium_user
|
||||
|
||||
if litellm.allow_dynamic_callback_disabling is not True or not premium_user:
|
||||
return frozenset()
|
||||
header: Final = (request_headers if request_headers is not None else _NO_REQUEST_HEADERS).get(
|
||||
X_LITELLM_DISABLE_CALLBACKS
|
||||
)
|
||||
if header is not None:
|
||||
return frozenset(name.strip().lower() for name in header.split(","))
|
||||
metadata: Final = user_api_key_dict.metadata
|
||||
disabled: Final = metadata.get("litellm_disabled_callbacks") if metadata else None
|
||||
if not isinstance(disabled, list):
|
||||
return frozenset()
|
||||
return frozenset(name.lower() for name in disabled if isinstance(name, str))
|
||||
|
||||
|
||||
def resolve_tenant_otel_destinations(
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
request_headers: Mapping[str, str] | None = None,
|
||||
) -> "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. An empty key-level list counts as configured, since that
|
||||
is what disabling a key's callbacks writes. 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.
|
||||
|
||||
Two entries naming the same backend merge their ``callback_vars`` last-wins, the
|
||||
way ``convert_key_logging_metadata_to_callback`` merges them, so the destination
|
||||
and the per-request tracer routing cannot read one config two ways.
|
||||
|
||||
A ``failure``-only entry is skipped: a destination is resolved during auth, before
|
||||
the request has an outcome, so honouring the filter would mean holding every span
|
||||
back until the call finishes. Those entries keep today's behaviour instead, where
|
||||
the tenant's credentials reach the backend through per-request tracer routing and
|
||||
the operator's exporter is left alone.
|
||||
|
||||
A backend the request disabled dynamically, through the key's
|
||||
``litellm_disabled_callbacks`` or the ``x-litellm-disable-callbacks`` header in
|
||||
``request_headers``, resolves to no destination, so the fan-out never carries the
|
||||
request tree to that account and the operator's exporter is never suppressed for
|
||||
it. That leaves the request exactly where it stood before destinations existed:
|
||||
the OTel V2 logger itself is not on the disable list's class registry, so its own
|
||||
span still routes to the tenant's credentials the way it did then.
|
||||
"""
|
||||
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 ()
|
||||
key_entries: Final = KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings(user_api_key_dict)
|
||||
entries: Final = (
|
||||
key_entries
|
||||
if key_entries is not None
|
||||
else KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(user_api_key_dict)
|
||||
)
|
||||
if not entries:
|
||||
return ()
|
||||
disabled: Final = _dynamically_disabled_backends(user_api_key_dict, request_headers)
|
||||
callbacks: Final = tuple(
|
||||
callback
|
||||
for item in entries
|
||||
if (callback := _get_validated_callback_metadata(item=item, source="otel-destination")) is not None
|
||||
if callback.callback_type != "failure"
|
||||
if callback.callback_name.lower() not in disabled
|
||||
)
|
||||
return tuple(
|
||||
destination
|
||||
for name in dict.fromkeys(callback.callback_name for callback in callbacks)
|
||||
if (
|
||||
destination := destination_for(
|
||||
name,
|
||||
_tenant_otel_params(
|
||||
MappingProxyType(
|
||||
{
|
||||
var: value
|
||||
for callback in callbacks
|
||||
if callback.callback_name == name
|
||||
for var, value in callback.callback_vars.items()
|
||||
}
|
||||
)
|
||||
),
|
||||
_tenant_service_name(user_api_key_dict),
|
||||
)
|
||||
)
|
||||
is not None
|
||||
)
|
||||
|
||||
|
||||
def _tenant_service_name(user_api_key_dict: UserAPIKeyAuth) -> str | None:
|
||||
"""The ``service.name`` this key or team configured, the key winning over its team.
|
||||
|
||||
Same fields and same precedence the request-metadata build applies, read straight
|
||||
off the auth object because destinations resolve during auth, before that metadata
|
||||
is assembled.
|
||||
"""
|
||||
sources: Final = (user_api_key_dict.metadata, user_api_key_dict.team_metadata)
|
||||
return next(
|
||||
(
|
||||
stripped
|
||||
for source in sources
|
||||
if source
|
||||
for field in OTEL_SERVICE_NAME_METADATA_KEYS
|
||||
if isinstance(value := source.get(field), str) and (stripped := value.strip())
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def clean_headers(
|
||||
headers: Headers,
|
||||
litellm_key_header_name: str | None = None,
|
||||
|
|
|
|||
2719
tests/test_litellm/integrations/otel/test_otel_v2_destinations.py
Normal file
2719
tests/test_litellm/integrations/otel/test_otel_v2_destinations.py
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -2041,7 +2041,7 @@ def test_select_global_otel_v2_logger_builds_one_when_none_registered():
|
|||
assert isinstance(chosen, OpenTelemetryV2)
|
||||
|
||||
|
||||
def test_publish_global_otel_v2_provider_sets_selected_logger_provider():
|
||||
def test_publish_global_otel_v2_provider_sets_selected_logger_provider(monkeypatch):
|
||||
"""The startup publish must set the OTel global provider to the *selected*
|
||||
logger's provider (the preset logger that owns every exporter), so the FastAPI
|
||||
server span and the gen-ai spans share one provider and one trace.
|
||||
|
|
@ -2051,8 +2051,10 @@ def test_publish_global_otel_v2_provider_sets_selected_logger_provider():
|
|||
test would otherwise miss: that the published provider is the selected logger's,
|
||||
not some other.
|
||||
"""
|
||||
from litellm.integrations.otel import logger as otel_logger
|
||||
from litellm.integrations.otel.logger import publish_global_otel_v2_provider
|
||||
|
||||
monkeypatch.setattr(otel_logger, "_published_v2_provider", None)
|
||||
cfg = OpenTelemetryV2Config(exporter="in_memory")
|
||||
tp = providers.build_tracer_provider(cfg)
|
||||
preset_logger = OpenTelemetryV2(
|
||||
|
|
|
|||
|
|
@ -5953,15 +5953,17 @@ def test_prompt_hooks_skip_prompt_managers_when_no_prompt_id(logging_obj, tmp_pa
|
|||
for hook in [cb for cb in litellm.callbacks if isinstance(cb, VectorStorePreCallHook)]:
|
||||
litellm.logging_callback_manager.remove_callback_from_list_by_object(litellm.callbacks, hook)
|
||||
def test_newrelic_dispatch_prefers_otel_v2_when_flag_on(monkeypatch):
|
||||
"""With LITELLM_OTEL_V2 on, the "newrelic" callback builds the OTel v2
|
||||
logger (per-team credential routing); with the flag off (default) it keeps
|
||||
the legacy agent-based logger, so existing deployments are untouched."""
|
||||
"""With LITELLM_OTEL_V2 on and operator credentials present, the "newrelic"
|
||||
callback builds the OTel v2 logger (per-team credential routing); with the
|
||||
flag off (default) it keeps the legacy agent-based logger, so existing
|
||||
deployments are untouched."""
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2
|
||||
from litellm.integrations.otel.model.config import is_otel_v2_enabled
|
||||
from litellm.litellm_core_utils import litellm_logging as logging_module
|
||||
|
||||
logging_module._in_memory_loggers.clear()
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
monkeypatch.setenv("NEW_RELIC_LICENSE_KEY", "test-license-key")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
try:
|
||||
v2_logger = logging_module._init_custom_logger_compatible_class(
|
||||
|
|
@ -6016,6 +6018,7 @@ def test_get_custom_logger_compatible_class_finds_v2_newrelic(monkeypatch):
|
|||
|
||||
logging_module._in_memory_loggers.clear()
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
monkeypatch.setenv("NEW_RELIC_LICENSE_KEY", "test-license-key")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
try:
|
||||
created = logging_module._init_custom_logger_compatible_class(
|
||||
|
|
|
|||
|
|
@ -7319,6 +7319,75 @@ class TestMCPMetaTraceCarrier:
|
|||
assert _mcp_meta_trace_carrier(SimpleNamespace(meta=only_progress)) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stateful_mcp_tool_call_uses_current_requests_otel_destinations() -> None:
|
||||
from types import SimpleNamespace
|
||||
|
||||
from mcp.server.lowlevel.server import request_ctx
|
||||
from mcp.shared.context import RequestContext
|
||||
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.integrations.otel.plumbing.context import (
|
||||
request_destinations,
|
||||
reset_request_destinations,
|
||||
set_request_destinations,
|
||||
)
|
||||
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
|
||||
global_mcp_server_manager,
|
||||
)
|
||||
from litellm.proxy._experimental.mcp_server.server import (
|
||||
_MCP_DESTINATIONS_SCOPE_KEY,
|
||||
mcp_server_tool_call,
|
||||
set_auth_context,
|
||||
)
|
||||
from litellm.proxy._experimental.mcp_server.tool_registry import (
|
||||
global_mcp_tool_registry,
|
||||
)
|
||||
|
||||
initialized_destination = OtelDestination(endpoint="https://initialize.example", callback_name="langfuse_otel")
|
||||
current_destination = OtelDestination(endpoint="https://current.example", callback_name="arize")
|
||||
server = MCPServer(
|
||||
server_id="otel-context-test",
|
||||
name="otelcontext",
|
||||
transport=MCPTransport.http,
|
||||
allow_all_keys=True,
|
||||
)
|
||||
|
||||
async def observe_destinations() -> str:
|
||||
assert request_destinations() == (current_destination,)
|
||||
return "ok"
|
||||
|
||||
global_mcp_server_manager.registry[server.server_id] = server
|
||||
global_mcp_server_manager.tool_name_to_mcp_server_name_mapping["otelcontext-observe"] = server.name
|
||||
global_mcp_tool_registry.register_tool(
|
||||
name="otelcontext-observe",
|
||||
description="Observe request destinations",
|
||||
input_schema={"type": "object"},
|
||||
handler=observe_destinations,
|
||||
)
|
||||
set_auth_context(None, raw_headers={})
|
||||
destinations_token = set_request_destinations((initialized_destination,))
|
||||
scope = {_MCP_DESTINATIONS_SCOPE_KEY: (current_destination,)}
|
||||
current_request_context = RequestContext(
|
||||
request_id=1,
|
||||
meta=None,
|
||||
session=SimpleNamespace(),
|
||||
lifespan_context=None,
|
||||
request=SimpleNamespace(scope=scope),
|
||||
)
|
||||
request_token = request_ctx.set(current_request_context)
|
||||
try:
|
||||
result = await mcp_server_tool_call("otelcontext-observe", {})
|
||||
assert result.isError is False
|
||||
assert request_destinations() == (initialized_destination,)
|
||||
finally:
|
||||
request_ctx.reset(request_token)
|
||||
reset_request_destinations(destinations_token)
|
||||
global_mcp_tool_registry.tools.pop("otelcontext-observe", None)
|
||||
global_mcp_server_manager.registry.pop(server.server_id, None)
|
||||
global_mcp_server_manager.tool_name_to_mcp_server_name_mapping.pop("otelcontext-observe", None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_allowed_mcp_servers_includes_active_servers_submitted_by_user():
|
||||
"""BYOM submitters can see approved servers they submitted without allow_all_keys."""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue