mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-13 08:52:02 +00:00
feat(otel): route Phoenix traces to per-key/team projects under otel v2
The v2 arize_phoenix preset read PHOENIX_PROJECT_NAME once at startup into a static resource attribute, silently dropping the per-key/team project routing v1 supported. Route it via Phoenix's x-project-name OTLP/HTTP header instead: the env var stays the global default, and a phoenix_project_name (or phoenix_project_name_override) in key/team metadata sends that key's traces to the named project. The project comes only from user_api_key_auth_metadata (server-set at auth), never from client request metadata or StandardCallbackDynamicParams, since choosing the telemetry destination is a data-exfiltration primitive. The header is appended to the exporter's static headers rather than replacing them, so the preset's Authorization survives, and it is gated to OTLP/HTTP exporters because Phoenix only reads it on /v1/traces. Also unban the bare phoenix_project_name fields from the request-body gate: the proxy integrations ignore them (only user_api_key_auth_metadata routes, and that stays banned), so rejecting them just broke SDK-style callers.
This commit is contained in:
mubashir1osmani
parent
32535987e8
commit
fad0cd515a
9 changed files with 522 additions and 54 deletions
|
|
@ -21,6 +21,7 @@ from litellm.integrations.otel.model.config import OpenTelemetryV2Config
|
|||
from litellm.integrations.otel.model.metadata import (
|
||||
LLMCallEvent,
|
||||
RequestIdentity,
|
||||
auth_metadata,
|
||||
model_from_request_data,
|
||||
)
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
|
|
@ -265,7 +266,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
call.provisional_span_name,
|
||||
parent_context=parent_context,
|
||||
start_time_ns=start_time_ns,
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params),
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params, call.auth_metadata),
|
||||
)
|
||||
self._open_llm_calls[call_id] = _LLMCallSpan(span=span, start_time_ns=start_time_ns)
|
||||
# Evict the oldest open call if the map is over budget. A call that opens
|
||||
|
|
@ -376,6 +377,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
start_time_ns=to_ns(start_time),
|
||||
end_time_ns=to_ns(end_time),
|
||||
links=links,
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, None, auth_metadata(payload, kwargs)),
|
||||
)
|
||||
return True
|
||||
|
||||
|
|
@ -412,6 +414,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
start_time_ns=to_ns(start_time),
|
||||
end_time_ns=to_ns(end_time),
|
||||
links=links,
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, None, auth_metadata(payload, kwargs)),
|
||||
)
|
||||
return True
|
||||
|
||||
|
|
@ -465,7 +468,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
parent_context=parent_ctx,
|
||||
start_time_ns=carrier.start_time_ns,
|
||||
end_time_ns=end_time_ns,
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params),
|
||||
tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params, call.auth_metadata),
|
||||
)
|
||||
|
||||
# ====================================================================== #
|
||||
|
|
|
|||
|
|
@ -36,8 +36,9 @@ model. They coincide on the SDK path, which is correct.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Iterator, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, cast
|
||||
|
||||
from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL
|
||||
|
|
@ -195,6 +196,10 @@ class LLMCallEvent:
|
|||
# The ``standard_callback_dynamic_params`` routing the call to a per-tenant
|
||||
# tracer (its own exporter/endpoint), or ``None`` when the call isn't scoped.
|
||||
dynamic_params: Any
|
||||
# The key/team config the proxy resolved at auth (``user_api_key_auth_metadata``),
|
||||
# routing the call to that tenant's telemetry project. Server-set and so
|
||||
# trusted, unlike ``dynamic_params``, which carries client-supplied metadata.
|
||||
auth_metadata: Mapping[str, str] | None
|
||||
# True for synthetic proxy-gate logs (auth / rate-limit rejections): they fire
|
||||
# the ``pre_call`` hook but never made an upstream call, so they get no span.
|
||||
is_no_upstream_call: bool
|
||||
|
|
@ -214,6 +219,7 @@ class LLMCallEvent:
|
|||
call_id=_call_id(payload, kwargs),
|
||||
payload=payload,
|
||||
dynamic_params=kwargs.get("standard_callback_dynamic_params"),
|
||||
auth_metadata=auth_metadata(payload, kwargs),
|
||||
is_no_upstream_call=bool(kwargs.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL)),
|
||||
provisional_span_name=f"{operation.value} {model}".strip(),
|
||||
time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs),
|
||||
|
|
@ -235,6 +241,64 @@ def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
|
|||
return completion_start - api_call_start
|
||||
|
||||
|
||||
def auth_metadata(payload: StandardLoggingPayload | None, kwargs: Mapping[str, object]) -> Mapping[str, str] | None:
|
||||
"""The key/team config the proxy resolved at auth, or ``None`` off the proxy.
|
||||
|
||||
Read from the payload once the call closes and from ``litellm_params`` at
|
||||
``pre_call``, where no payload exists yet — the LLM-call span is *created* at
|
||||
``pre_call``, so the tracer (and therefore the destination) must be
|
||||
resolvable there. Values arrive untyped, so non-string entries are dropped
|
||||
rather than passed on to header builders.
|
||||
"""
|
||||
return next(
|
||||
(
|
||||
typed
|
||||
for metadata in _metadata_dicts(payload, kwargs)
|
||||
if (typed := _string_entries(metadata.get("user_api_key_auth_metadata")))
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def _as_str_mapping(value: object) -> Mapping[str, object] | None:
|
||||
"""A read-only view of ``value`` when it is a mapping, else ``None``."""
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
return cast("Mapping[str, object]", value) # cast-ok: isinstance-guarded, JSON metadata has str keys
|
||||
|
||||
|
||||
def _string_entries(value: object) -> Mapping[str, str] | None:
|
||||
entries: Final = _as_str_mapping(value)
|
||||
if entries is None:
|
||||
return None
|
||||
typed: Final = MappingProxyType({key: item for key, item in entries.items() if isinstance(item, str)})
|
||||
return typed or None
|
||||
|
||||
|
||||
def _metadata_dicts(
|
||||
payload: StandardLoggingPayload | None, kwargs: Mapping[str, object]
|
||||
) -> Iterator[Mapping[str, object]]:
|
||||
"""Request metadata dicts, closed-call payload first then the live kwargs.
|
||||
|
||||
``litellm_metadata`` is the metadata field on the Anthropic-shaped routes;
|
||||
litellm copies it onto ``metadata``, but both are yielded so a route that
|
||||
populates only one is still covered.
|
||||
"""
|
||||
payload_view: Final = _as_str_mapping(payload)
|
||||
if payload_view is not None:
|
||||
payload_metadata: Final = _as_str_mapping(payload_view.get("metadata"))
|
||||
if payload_metadata is not None:
|
||||
yield payload_metadata
|
||||
params: Final = _as_str_mapping(kwargs.get("litellm_params"))
|
||||
if params is None:
|
||||
return
|
||||
yield from (
|
||||
metadata
|
||||
for key in ("metadata", "litellm_metadata")
|
||||
if (metadata := _as_str_mapping(params.get(key))) is not None
|
||||
)
|
||||
|
||||
|
||||
def _call_id(payload: StandardLoggingPayload | None, kwargs: Mapping[str, Any]) -> str | None:
|
||||
"""The call id from the payload (when closed) or the bare kwargs (at pre_call)."""
|
||||
if payload is not None:
|
||||
|
|
|
|||
|
|
@ -1,31 +1,43 @@
|
|||
"""Per-request multi-tenant tracer routing.
|
||||
|
||||
When a request carries team/key vendor credentials in
|
||||
``standard_callback_dynamic_params``, its spans must export through a
|
||||
``TracerProvider`` whose OTLP headers carry those credentials.
|
||||
``TenantTracerCache`` builds and caches one provider per distinct credential
|
||||
set, and otherwise hands back the logger's default tracer. This lets a single
|
||||
logger fan requests out to many tenants without needing a logger per tenant.
|
||||
``standard_callback_dynamic_params``, or the key/team config resolved at auth
|
||||
names a destination project, its spans must export through a
|
||||
``TracerProvider`` whose OTLP headers carry those credentials / that project.
|
||||
``TenantTracerCache`` builds and caches one provider per distinct
|
||||
(credentials, project) pair, and otherwise hands back the logger's default
|
||||
tracer. This lets a single logger fan requests out to many tenants without
|
||||
needing a logger per tenant.
|
||||
"""
|
||||
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Final
|
||||
from types import MappingProxyType
|
||||
from typing import Any, Final, TypeAlias
|
||||
from urllib.parse import quote
|
||||
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.trace import Tracer
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.plumbing.providers import (
|
||||
build_tracer_provider,
|
||||
get_tracer,
|
||||
)
|
||||
from litellm.integrations.otel.presets import dynamic_otlp_headers
|
||||
from litellm.integrations.otel.presets import (
|
||||
dynamic_otlp_headers,
|
||||
project_routing_headers,
|
||||
)
|
||||
|
||||
# Exporter kinds that ignore headers — never rewritten with dynamic credentials.
|
||||
_NON_OTLP_KINDS: Final = ("console", "in_memory", "inmemory", "memory")
|
||||
|
||||
# gRPC exporters still take dynamic credentials (as gRPC metadata) but not
|
||||
# project headers: the routing headers backends read (Phoenix's
|
||||
# ``x-project-name``) are only honored on the OTLP/HTTP endpoint.
|
||||
_GRPC_KINDS: Final = ("otlp_grpc", "grpc")
|
||||
|
||||
# Cap on distinct credential-scoped providers held at once. ``dynamic_params``
|
||||
# can be populated from request metadata, so an unbounded cache lets a caller
|
||||
# spawn one ``TracerProvider`` (plus its ``BatchSpanProcessor`` background
|
||||
|
|
@ -34,6 +46,10 @@ _NON_OTLP_KINDS: Final = ("console", "in_memory", "inmemory", "memory")
|
|||
# evicted providers so their threads are reclaimed.
|
||||
_MAX_CACHED_PROVIDERS: Final = 256
|
||||
|
||||
_HeaderItems: TypeAlias = tuple[tuple[str, str], ...]
|
||||
|
||||
_NO_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
|
||||
|
||||
|
||||
def _shutdown_provider(provider: TracerProvider) -> None:
|
||||
"""Flush + stop an evicted provider's processors (reclaims their threads).
|
||||
|
|
@ -49,8 +65,19 @@ def _shutdown_provider(provider: TracerProvider) -> None:
|
|||
verbose_logger.debug("OTel V2: error shutting down evicted provider: %s", e)
|
||||
|
||||
|
||||
def _plain_header_string(headers: Mapping[str, str]) -> str:
|
||||
return ",".join(f"{key}={value}" for key, value in headers.items())
|
||||
|
||||
|
||||
def _encoded_header_string(headers: Mapping[str, str]) -> str:
|
||||
"""Percent-encode values so one containing the ``k=v,k=v`` separators (e.g.
|
||||
a project name with a comma) survives; ``parse_env_headers`` decodes it back.
|
||||
"""
|
||||
return ",".join(f"{key}={quote(value, safe='')}" for key, value in headers.items())
|
||||
|
||||
|
||||
class TenantTracerCache:
|
||||
"""Credential-scoped ``TracerProvider`` cache keyed by the dynamic headers."""
|
||||
"""Credential/project-scoped ``TracerProvider`` cache keyed by the routing headers."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
|
@ -61,49 +88,100 @@ class TenantTracerCache:
|
|||
self._config = config
|
||||
self._callback_name = callback_name
|
||||
self._tracer_name = tracer_name
|
||||
self._providers: OrderedDict[tuple[tuple[str, str], ...], TracerProvider] = OrderedDict()
|
||||
self._providers: OrderedDict[tuple[_HeaderItems, _HeaderItems], TracerProvider] = OrderedDict()
|
||||
self._project_routable = any(
|
||||
spec.owner == callback_name and spec.kind.lower() not in (*_NON_OTLP_KINDS, *_GRPC_KINDS)
|
||||
for spec in config.exporters
|
||||
)
|
||||
self._warned_project_unroutable = False
|
||||
|
||||
def tracer_for(self, default: Tracer, dynamic_params: Any) -> Tracer:
|
||||
def tracer_for(
|
||||
self,
|
||||
default: Tracer,
|
||||
dynamic_params: Any,
|
||||
auth_metadata: Mapping[str, str] | None = None,
|
||||
) -> Tracer:
|
||||
"""Return the tracer for this request.
|
||||
|
||||
Use ``default`` unless the request's dynamic credentials require a
|
||||
credential-scoped tracer, in which case build (or reuse) one. The cache
|
||||
is a bounded LRU: the least-recently-used provider is flushed and shut
|
||||
down on overflow so its exporter threads don't accumulate.
|
||||
Use ``default`` unless the request's dynamic credentials or its key/team
|
||||
project require a scoped tracer, in which case build (or reuse) one. The
|
||||
cache is a bounded LRU: the least-recently-used provider is flushed and
|
||||
shut down on overflow so its exporter threads don't accumulate.
|
||||
"""
|
||||
headers: Final = dynamic_otlp_headers(self._callback_name, dynamic_params)
|
||||
if not headers:
|
||||
credential_headers: Final = dynamic_otlp_headers(self._callback_name, dynamic_params) or _NO_HEADERS
|
||||
project_headers: Final = self._project_headers(auth_metadata)
|
||||
if not credential_headers and not project_headers:
|
||||
return default
|
||||
cache_key: Final = tuple(sorted(headers.items()))
|
||||
cache_key: Final = (
|
||||
tuple(sorted(credential_headers.items())),
|
||||
tuple(sorted(project_headers.items())),
|
||||
)
|
||||
provider = self._providers.get(cache_key)
|
||||
if provider is not None:
|
||||
self._providers.move_to_end(cache_key)
|
||||
else:
|
||||
provider = build_tracer_provider(self._config_with_headers(headers))
|
||||
provider = build_tracer_provider(self._routed_config(credential_headers, project_headers))
|
||||
self._providers[cache_key] = provider
|
||||
if len(self._providers) > _MAX_CACHED_PROVIDERS:
|
||||
_, evicted = self._providers.popitem(last=False)
|
||||
_shutdown_provider(evicted)
|
||||
return get_tracer(provider, self._tracer_name)
|
||||
|
||||
def _config_with_headers(self, headers: Mapping[str, str]) -> OpenTelemetryV2Config:
|
||||
"""Clone the config, stamping ``headers`` onto the credential's own exporter.
|
||||
def _project_headers(self, auth_metadata: Mapping[str, str] | None) -> Mapping[str, str]:
|
||||
"""The per-request project-routing headers, if this cache can apply them.
|
||||
|
||||
``headers`` are the per-request credentials of ``self._callback_name`` (the
|
||||
integration that built this cache), so they apply only to the exporter that
|
||||
integration contributed (``spec.owner``). A request that carries one
|
||||
tenant's Arize key must never rewrite the headers of a co-configured
|
||||
Langfuse or self-hosted collector exporter, which would leak that key to a
|
||||
different backend.
|
||||
A gRPC-only exporter can't (the project header route is HTTP-only), so
|
||||
the request warns once and stays on the env-configured default project.
|
||||
"""
|
||||
header_str: Final = ",".join(f"{key}={value}" for key, value in headers.items())
|
||||
header_update: Final[dict[str, str]] = {"headers": header_str}
|
||||
exporters: Final = [
|
||||
(
|
||||
spec.model_copy(update=header_update)
|
||||
if spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS
|
||||
else spec
|
||||
requested: Final = project_routing_headers(self._callback_name, auth_metadata)
|
||||
if not requested or self._project_routable:
|
||||
return requested
|
||||
if not self._warned_project_unroutable:
|
||||
self._warned_project_unroutable = True
|
||||
verbose_logger.warning(
|
||||
"OTel V2: %s key/team config names a per-request project, but its exporter "
|
||||
"is not OTLP/HTTP and the project header is HTTP-only; spans stay in the "
|
||||
"default project.",
|
||||
self._callback_name,
|
||||
)
|
||||
for spec in self._config.exporters
|
||||
return _NO_HEADERS
|
||||
|
||||
def _routed_config(
|
||||
self,
|
||||
credential_headers: Mapping[str, str],
|
||||
project_headers: Mapping[str, str],
|
||||
) -> OpenTelemetryV2Config:
|
||||
"""Clone the config, rewriting headers on the callback's own exporter.
|
||||
|
||||
Both header sets apply only to the exporter ``self._callback_name``
|
||||
contributed (``spec.owner``). A request that carries one tenant's Arize
|
||||
key must never rewrite the headers of a co-configured Langfuse or
|
||||
self-hosted collector exporter, which would leak that key to a
|
||||
different backend.
|
||||
|
||||
Dynamic credentials REPLACE the exporter's headers — they are the
|
||||
tenant's complete credential set. Project headers APPEND instead: the
|
||||
preset's static headers carry the backend auth (Phoenix's
|
||||
``Authorization``), which must survive routing to a project.
|
||||
"""
|
||||
exporters: Final = [
|
||||
self._routed_exporter(spec, credential_headers, project_headers) for spec in self._config.exporters
|
||||
]
|
||||
return self._config.model_copy(update={"exporters": exporters})
|
||||
|
||||
def _routed_exporter(
|
||||
self,
|
||||
spec: ExporterSpec,
|
||||
credential_headers: Mapping[str, str],
|
||||
project_headers: Mapping[str, str],
|
||||
) -> ExporterSpec:
|
||||
kind: Final = spec.kind.lower()
|
||||
if spec.owner != self._callback_name or kind in _NON_OTLP_KINDS:
|
||||
return spec
|
||||
base: Final = _plain_header_string(credential_headers) if credential_headers else spec.headers
|
||||
routed: Final = (
|
||||
",".join(part for part in (base, _encoded_header_string(project_headers)) if part)
|
||||
if project_headers and kind not in _GRPC_KINDS
|
||||
else base
|
||||
)
|
||||
return spec if routed == spec.headers else spec.model_copy(update={"headers": routed})
|
||||
|
|
|
|||
|
|
@ -8,7 +8,8 @@ the factory in ``litellm_logging`` can resolve a name and build a single
|
|||
``OpenTelemetryV2`` instance from the result.
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Mapping
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from litellm.integrations.otel.presets.agentops import agentops_preset
|
||||
|
|
@ -20,7 +21,10 @@ from litellm.integrations.otel.presets.langfuse import (
|
|||
)
|
||||
from litellm.integrations.otel.presets.langtrace import langtrace_preset
|
||||
from litellm.integrations.otel.presets.levo import levo_preset
|
||||
from litellm.integrations.otel.presets.phoenix import phoenix_preset
|
||||
from litellm.integrations.otel.presets.phoenix import (
|
||||
phoenix_preset,
|
||||
phoenix_project_headers,
|
||||
)
|
||||
from litellm.integrations.otel.presets.weave import weave_dynamic_headers, weave_preset
|
||||
from litellm.types.utils import StandardCallbackDynamicParams
|
||||
|
||||
|
|
@ -47,6 +51,23 @@ DYNAMIC_HEADERS_BY_CALLBACK: Final[dict[str, Callable[[StandardCallbackDynamicPa
|
|||
}
|
||||
|
||||
|
||||
#: Callback name → per-request *routing* header builder, sourced from the key/team
|
||||
#: config the proxy resolved at auth. Deliberately separate from
|
||||
#: ``DYNAMIC_HEADERS_BY_CALLBACK``: that one is fed
|
||||
#: ``StandardCallbackDynamicParams``, which is populated from client-supplied
|
||||
#: request metadata. Naming a destination project is a data-exfiltration
|
||||
#: primitive, so it must only ever come from server-set key/team config.
|
||||
PROJECT_HEADERS_BY_CALLBACK: Final[Mapping[str, Callable[[Mapping[str, str] | None], Mapping[str, str]]]] = (
|
||||
MappingProxyType(
|
||||
{
|
||||
"arize_phoenix": phoenix_project_headers,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
_NO_PROJECT_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
|
||||
|
||||
|
||||
def dynamic_otlp_headers(
|
||||
callback_name: str | None,
|
||||
dynamic_params: StandardCallbackDynamicParams | None,
|
||||
|
|
@ -62,9 +83,25 @@ def dynamic_otlp_headers(
|
|||
return headers or None
|
||||
|
||||
|
||||
def project_routing_headers(
|
||||
callback_name: str | None,
|
||||
auth_metadata: Mapping[str, str] | None,
|
||||
) -> Mapping[str, str]:
|
||||
"""Per-request project-routing headers from trusted key/team config.
|
||||
|
||||
Empty means "no per-request project" — the caller keeps its default tracer,
|
||||
whose resource attributes carry the env-configured project.
|
||||
"""
|
||||
builder: Final = PROJECT_HEADERS_BY_CALLBACK.get(callback_name or "")
|
||||
if builder is None:
|
||||
return _NO_PROJECT_HEADERS
|
||||
return builder(auth_metadata)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DYNAMIC_HEADERS_BY_CALLBACK",
|
||||
"PRESET_BY_CALLBACK",
|
||||
"PROJECT_HEADERS_BY_CALLBACK",
|
||||
"Preset",
|
||||
"agentops_preset",
|
||||
"arize_preset",
|
||||
|
|
@ -73,5 +110,6 @@ __all__ = [
|
|||
"langtrace_preset",
|
||||
"levo_preset",
|
||||
"phoenix_preset",
|
||||
"project_routing_headers",
|
||||
"weave_preset",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
"""Arize-Phoenix preset."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from pydantic import AliasChoices, Field
|
||||
|
|
@ -25,6 +27,36 @@ class _PhoenixSettings(BaseSettings):
|
|||
)
|
||||
|
||||
|
||||
#: Phoenix routes an OTLP/HTTP export to a project by this header, which takes
|
||||
#: precedence over the ``openinference.project.name`` resource attribute the env
|
||||
#: var sets. Requires arize-phoenix 15.5.0+; older collectors ignore it and the
|
||||
#: spans land in the resource attribute's project.
|
||||
PHOENIX_PROJECT_HEADER: Final = "x-project-name"
|
||||
|
||||
#: Key/team config fields naming the target project, highest precedence first.
|
||||
_PROJECT_KEYS: Final = ("phoenix_project_name_override", "phoenix_project_name")
|
||||
|
||||
_NO_PROJECT: Final[Mapping[str, str]] = MappingProxyType({})
|
||||
|
||||
|
||||
def phoenix_project_headers(auth_metadata: Mapping[str, str] | None) -> Mapping[str, str]:
|
||||
"""The per-request Phoenix project header for this key/team, if any.
|
||||
|
||||
``auth_metadata`` must be the key/team config the proxy resolved at auth
|
||||
(``user_api_key_auth_metadata``), never client-supplied request metadata:
|
||||
choosing the destination project is a data-exfiltration primitive, so a
|
||||
caller must not be able to name one. Returns an empty mapping when the key
|
||||
and team name no project, leaving the request on the env-configured default.
|
||||
"""
|
||||
if not auth_metadata:
|
||||
return _NO_PROJECT
|
||||
project: Final = next(
|
||||
(stripped for key in _PROJECT_KEYS if (stripped := (auth_metadata.get(key) or "").strip())),
|
||||
"",
|
||||
)
|
||||
return MappingProxyType({PHOENIX_PROJECT_HEADER: project}) if project else _NO_PROJECT
|
||||
|
||||
|
||||
def phoenix_preset(
|
||||
*,
|
||||
config_overrides: OpenTelemetryV2Config | None = None,
|
||||
|
|
|
|||
|
|
@ -217,8 +217,10 @@ _SAFE_CLIENT_CALLBACK_PARAMS: Final[frozenset[str]] = frozenset(
|
|||
_EXTRA_BANNED_OBSERVABILITY_PARAMS: Final[frozenset[str]] = frozenset(
|
||||
{
|
||||
"posthog_api_url",
|
||||
"phoenix_project_name",
|
||||
"phoenix_project_name_override",
|
||||
# ``phoenix_project_name`` / ``phoenix_project_name_override`` are NOT
|
||||
# banned: on the proxy the Phoenix integrations only read them from
|
||||
# ``user_api_key_auth_metadata`` (key/team config), so the bare request
|
||||
# fields are inert and rejecting them just breaks SDK-style callers.
|
||||
# Server-reserved: written exclusively by add_user_api_key_auth_to_request_metadata
|
||||
# from the authenticated key's database record. A caller-supplied value
|
||||
# would survive the server merge and let an authenticated user redirect
|
||||
|
|
|
|||
|
|
@ -8,7 +8,11 @@ sys.path.insert(0, os.path.abspath("../../../.."))
|
|||
from opentelemetry.trace import NoOpTracer
|
||||
|
||||
from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.presets import dynamic_otlp_headers
|
||||
from litellm.integrations.otel.presets import (
|
||||
dynamic_otlp_headers,
|
||||
project_routing_headers,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.providers import parse_headers
|
||||
from litellm.integrations.otel.plumbing.routing import TenantTracerCache
|
||||
|
||||
|
||||
|
|
@ -128,7 +132,7 @@ def test_dynamic_headers_applied_to_otlp_exporter_only():
|
|||
ExporterSpec(kind="in_memory", owner="arize"),
|
||||
],
|
||||
)
|
||||
new_cfg = cache._config_with_headers({"arize-space-id": "S", "api_key": "K"})
|
||||
new_cfg = cache._routed_config({"arize-space-id": "S", "api_key": "K"}, {})
|
||||
otlp, in_mem = new_cfg.exporters
|
||||
assert otlp.headers == "arize-space-id=S,api_key=K"
|
||||
assert in_mem.headers is None # console/in_memory left untouched
|
||||
|
|
@ -138,10 +142,9 @@ def test_dynamic_headers_do_not_leak_to_other_owners_exporter():
|
|||
"""A tenant's Arize credentials must never be stamped onto a co-configured
|
||||
exporter owned by a different backend (a self-hosted collector, Langfuse).
|
||||
|
||||
Regression for the cross-backend credential leak: ``_config_with_headers``
|
||||
used to rewrite the headers of every OTLP exporter, so one request carrying
|
||||
a team's Arize key clobbered the base collector's and Langfuse's headers
|
||||
with that key.
|
||||
Regression for the cross-backend credential leak: the header rewrite used
|
||||
to hit every OTLP exporter, so one request carrying a team's Arize key
|
||||
clobbered the base collector's and Langfuse's headers with that key.
|
||||
"""
|
||||
cache = _cache(
|
||||
"arize",
|
||||
|
|
@ -166,10 +169,131 @@ def test_dynamic_headers_do_not_leak_to_other_owners_exporter():
|
|||
),
|
||||
],
|
||||
)
|
||||
new_cfg = cache._config_with_headers(
|
||||
{"arize-space-id": "TEAMX", "api_key": "TEAMX_KEY"}
|
||||
new_cfg = cache._routed_config(
|
||||
{"arize-space-id": "TEAMX", "api_key": "TEAMX_KEY"}, {}
|
||||
)
|
||||
by_owner = {e.owner: e.headers for e in new_cfg.exporters}
|
||||
assert by_owner["arize"] == "arize-space-id=TEAMX,api_key=TEAMX_KEY"
|
||||
assert by_owner[None] == "x=base-collector"
|
||||
assert by_owner["langfuse_otel"] == "Authorization=Basic base-langfuse"
|
||||
|
||||
|
||||
# --- per-request Phoenix project routing from trusted key/team config --- #
|
||||
|
||||
|
||||
def _phoenix_cache(kind="otlp_http"):
|
||||
return _cache(
|
||||
"arize_phoenix",
|
||||
exporters=[
|
||||
ExporterSpec(
|
||||
kind=kind,
|
||||
endpoint="http://phoenix:6006",
|
||||
headers="Authorization=Bearer phoenix-key",
|
||||
owner="arize_phoenix",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def test_phoenix_project_headers_precedence_and_blanks():
|
||||
assert project_routing_headers(
|
||||
"arize_phoenix", {"phoenix_project_name": "team-proj"}
|
||||
) == {"x-project-name": "team-proj"}
|
||||
assert project_routing_headers(
|
||||
"arize_phoenix",
|
||||
{"phoenix_project_name_override": "override", "phoenix_project_name": "base"},
|
||||
) == {"x-project-name": "override"}
|
||||
assert (
|
||||
project_routing_headers("arize_phoenix", {"phoenix_project_name": " "}) == {}
|
||||
)
|
||||
assert project_routing_headers("arize_phoenix", None) == {}
|
||||
# Only Phoenix participates in project routing.
|
||||
assert project_routing_headers("arize", {"phoenix_project_name": "p"}) == {}
|
||||
|
||||
|
||||
def test_project_header_appends_and_preserves_phoenix_auth():
|
||||
"""Regression: routing to a project must not drop the preset's static
|
||||
``Authorization`` header — a replace would break Phoenix auth entirely."""
|
||||
cache = _phoenix_cache()
|
||||
cfg = cache._routed_config({}, {"x-project-name": "team-proj"})
|
||||
(spec,) = cfg.exporters
|
||||
parsed = parse_headers(spec.headers)
|
||||
assert parsed["authorization"] == "Bearer phoenix-key"
|
||||
assert parsed["x-project-name"] == "team-proj"
|
||||
|
||||
|
||||
def test_project_name_with_header_separators_round_trips():
|
||||
cache = _phoenix_cache()
|
||||
cfg = cache._routed_config({}, {"x-project-name": "my proj, prod=1"})
|
||||
(spec,) = cfg.exporters
|
||||
parsed = parse_headers(spec.headers)
|
||||
assert parsed["x-project-name"] == "my proj, prod=1"
|
||||
assert parsed["authorization"] == "Bearer phoenix-key"
|
||||
|
||||
|
||||
def test_project_header_does_not_touch_other_exporters():
|
||||
cache = _cache(
|
||||
"arize_phoenix",
|
||||
exporters=[
|
||||
ExporterSpec(
|
||||
kind="otlp_http",
|
||||
endpoint="http://collector:4318",
|
||||
headers="x=base-collector",
|
||||
owner=None,
|
||||
),
|
||||
ExporterSpec(
|
||||
kind="otlp_http",
|
||||
endpoint="http://phoenix:6006",
|
||||
headers="Authorization=Bearer phoenix-key",
|
||||
owner="arize_phoenix",
|
||||
),
|
||||
],
|
||||
)
|
||||
cfg = cache._routed_config({}, {"x-project-name": "team-proj"})
|
||||
by_owner = {e.owner: e.headers for e in cfg.exporters}
|
||||
assert by_owner[None] == "x=base-collector"
|
||||
assert parse_headers(by_owner["arize_phoenix"])["x-project-name"] == "team-proj"
|
||||
|
||||
|
||||
def test_provider_cached_per_project():
|
||||
cache = _phoenix_cache()
|
||||
default = NoOpTracer()
|
||||
routed = cache.tracer_for(default, None, {"phoenix_project_name": "proj-a"})
|
||||
assert routed is not default
|
||||
cache.tracer_for(default, None, {"phoenix_project_name": "proj-a"})
|
||||
assert len(cache._providers) == 1
|
||||
cache.tracer_for(default, None, {"phoenix_project_name": "proj-b"})
|
||||
assert len(cache._providers) == 2
|
||||
for provider in cache._providers.values():
|
||||
provider.shutdown()
|
||||
|
||||
|
||||
def test_client_dynamic_params_cannot_choose_phoenix_project():
|
||||
# ``StandardCallbackDynamicParams`` is populated from client-supplied
|
||||
# request metadata; the project may only come from server-set key/team
|
||||
# config (the ``auth_metadata`` argument).
|
||||
cache = _phoenix_cache()
|
||||
default = NoOpTracer()
|
||||
assert cache.tracer_for(default, {"phoenix_project_name": "attacker"}) is default
|
||||
assert (
|
||||
cache.tracer_for(default, {"phoenix_project_name_override": "attacker"})
|
||||
is default
|
||||
)
|
||||
assert cache._providers == {}
|
||||
|
||||
|
||||
def test_auth_metadata_without_project_uses_default_tracer():
|
||||
cache = _phoenix_cache()
|
||||
default = NoOpTracer()
|
||||
assert cache.tracer_for(default, None, {"logging_setting": "x"}) is default
|
||||
assert cache._providers == {}
|
||||
|
||||
|
||||
def test_grpc_exporter_gets_no_project_routing():
|
||||
# ``x-project-name`` is only honored on the OTLP/HTTP endpoint, so a
|
||||
# gRPC-only Phoenix exporter stays on the default project (warned once).
|
||||
cache = _phoenix_cache(kind="otlp_grpc")
|
||||
default = NoOpTracer()
|
||||
assert cache.tracer_for(default, None, {"phoenix_project_name": "proj"}) is default
|
||||
assert cache._providers == {}
|
||||
assert cache._warned_project_unroutable is True
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ from litellm.integrations.otel.plumbing.context import ( # noqa: E402
|
|||
set_request_root_span,
|
||||
)
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402
|
||||
from litellm.integrations.otel.model.config import ExporterSpec # noqa: E402
|
||||
from litellm.integrations.otel.model.spans import ( # noqa: E402
|
||||
LITELLM_PROXY_REQUEST_SPAN_NAME,
|
||||
SpanRole,
|
||||
|
|
@ -2277,3 +2278,110 @@ def test_metrics_disabled_by_default_records_nothing(monkeypatch):
|
|||
)
|
||||
)
|
||||
assert _emitted_metric_names(reader) == set()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Per-request Phoenix project routing (key/team auth metadata)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _phoenix_routing_logger(capture_kind):
|
||||
"""A Phoenix-shaped logger whose owned exporter is a registered factory kind
|
||||
that captures the exporter built per routed header set, so the test can
|
||||
assert which destination each span actually exported through."""
|
||||
captured = {}
|
||||
|
||||
def factory(spec):
|
||||
exporter = InMemorySpanExporter()
|
||||
captured[spec.headers] = exporter
|
||||
return exporter
|
||||
|
||||
providers.register_exporter_factory(capture_kind, factory)
|
||||
cfg = OpenTelemetryV2Config(
|
||||
exporters=[
|
||||
ExporterSpec(
|
||||
kind=capture_kind,
|
||||
endpoint="http://phoenix:6006",
|
||||
headers="Authorization=Bearer phoenix-key",
|
||||
owner="arize_phoenix",
|
||||
)
|
||||
]
|
||||
)
|
||||
default_exporter = InMemorySpanExporter()
|
||||
tracer_provider = providers.build_tracer_provider(cfg, exporter=default_exporter)
|
||||
logger = OpenTelemetryV2(
|
||||
config=cfg, callback_name="arize_phoenix", tracer_provider=tracer_provider
|
||||
)
|
||||
return logger, default_exporter, captured
|
||||
|
||||
|
||||
def test_key_team_auth_metadata_routes_llm_span_to_phoenix_project():
|
||||
"""The proxy stamps the key/team config into ``user_api_key_auth_metadata``;
|
||||
a ``phoenix_project_name`` there must route the LLM span through an exporter
|
||||
carrying the ``x-project-name`` header while keeping the preset's auth."""
|
||||
logger, default_exporter, captured = _phoenix_routing_logger("capture_route_a")
|
||||
auth_md = {"phoenix_project_name": "team-proj"}
|
||||
payload = _payload(metadata={"user_api_key_auth_metadata": auth_md})
|
||||
kwargs = {
|
||||
"standard_logging_object": payload,
|
||||
"litellm_params": {"metadata": {"user_api_key_auth_metadata": auth_md}},
|
||||
}
|
||||
_emit_llm(logger, kwargs)
|
||||
|
||||
assert [s.name for s in default_exporter.get_finished_spans()] == []
|
||||
(headers,) = captured
|
||||
parsed = providers.parse_headers(headers)
|
||||
assert parsed["x-project-name"] == "team-proj"
|
||||
assert parsed["authorization"] == "Bearer phoenix-key"
|
||||
routed_spans = captured[headers].get_finished_spans()
|
||||
assert len(routed_spans) == 1
|
||||
|
||||
|
||||
def test_client_request_metadata_cannot_route_phoenix_project():
|
||||
"""A bare ``phoenix_project_name`` in client request metadata (not the
|
||||
server-set ``user_api_key_auth_metadata``) must be ignored: the span stays
|
||||
on the default tracer and no routed exporter is ever built."""
|
||||
logger, default_exporter, captured = _phoenix_routing_logger("capture_route_b")
|
||||
payload = _payload(metadata={"phoenix_project_name": "attacker-project"})
|
||||
kwargs = {
|
||||
"standard_logging_object": payload,
|
||||
"litellm_params": {"metadata": {"phoenix_project_name": "attacker-project"}},
|
||||
}
|
||||
_emit_llm(logger, kwargs)
|
||||
|
||||
assert captured == {}
|
||||
assert len(default_exporter.get_finished_spans()) == 1
|
||||
|
||||
|
||||
def test_project_routing_resolves_at_pre_call_before_payload_exists():
|
||||
"""Production ``pre_call`` runs before the standard logging payload exists,
|
||||
so the destination project must resolve from ``litellm_params`` alone — the
|
||||
span is created (and its exporter chosen) right there."""
|
||||
logger, default_exporter, captured = _phoenix_routing_logger("capture_route_c")
|
||||
auth_md = {"phoenix_project_name": "team-proj"}
|
||||
litellm_params = {"metadata": {"user_api_key_auth_metadata": auth_md}}
|
||||
server = logger._emitter.start_span(
|
||||
SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME
|
||||
)
|
||||
with trace.use_span(server, end_on_exit=False):
|
||||
logger.log_pre_api_call(
|
||||
model="gpt-4o",
|
||||
messages=[],
|
||||
kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params},
|
||||
)
|
||||
server.end()
|
||||
assert len(captured) == 1 # routed exporter already built at pre_call
|
||||
|
||||
close_kwargs = {
|
||||
"standard_logging_object": _payload(
|
||||
metadata={"user_api_key_auth_metadata": auth_md}
|
||||
),
|
||||
"litellm_params": litellm_params,
|
||||
}
|
||||
asyncio.run(logger.async_log_success_event(close_kwargs, None, None, None))
|
||||
|
||||
(headers,) = captured
|
||||
assert [s.name for s in captured[headers].get_finished_spans()] == ["chat gpt-4o"]
|
||||
assert all(
|
||||
s.name != "chat gpt-4o" for s in default_exporter.get_finished_spans()
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2653,8 +2653,6 @@ class TestObservabilityCallbackBans:
|
|||
"posthog_api_url",
|
||||
"braintrust_api_key",
|
||||
"braintrust_project",
|
||||
"phoenix_project_name",
|
||||
"phoenix_project_name_override",
|
||||
"wandb_api_key",
|
||||
"weave_project_id",
|
||||
"gcs_bucket_name",
|
||||
|
|
@ -2685,8 +2683,7 @@ class TestObservabilityCallbackBans:
|
|||
"langsmith_api_key",
|
||||
"posthog_api_url",
|
||||
"braintrust_project",
|
||||
"phoenix_project_name",
|
||||
"phoenix_project_name_override",
|
||||
"user_api_key_auth_metadata",
|
||||
],
|
||||
)
|
||||
def test_observability_field_in_metadata_dict_is_rejected(
|
||||
|
|
@ -2707,6 +2704,28 @@ class TestObservabilityCallbackBans:
|
|||
)
|
||||
assert field in str(exc.value)
|
||||
|
||||
@pytest.mark.parametrize("metadata_key", ["metadata", "litellm_metadata"])
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
["phoenix_project_name", "phoenix_project_name_override"],
|
||||
)
|
||||
def test_phoenix_project_fields_in_metadata_are_accepted(self, metadata_key, field):
|
||||
# The Phoenix integrations only honor the project from
|
||||
# ``user_api_key_auth_metadata`` on the proxy, so the bare metadata
|
||||
# fields are inert and must not 400 SDK-style callers that send them.
|
||||
assert (
|
||||
is_request_body_safe(
|
||||
request_body={
|
||||
"model": "gpt-4",
|
||||
metadata_key: {field: "client-project"},
|
||||
},
|
||||
general_settings={},
|
||||
llm_router=None,
|
||||
model="gpt-4",
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
def test_observability_field_in_litellm_params_metadata_is_rejected(self):
|
||||
with pytest.raises(ValueError) as exc:
|
||||
is_request_body_safe(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue