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