From 7680c355c622138ed117620507fb44368310ebca Mon Sep 17 00:00:00 2001 From: Yucheng He Date: Sat, 5 Sep 2026 17:28:59 -0700 Subject: [PATCH] fix(otel): close final tenant routing gaps --- litellm/integrations/otel/logger.py | 12 +- .../integrations/otel/plumbing/providers.py | 54 +++++--- .../otel/test_otel_v2_destinations.py | 116 ++++++++++++++++++ 3 files changed, 163 insertions(+), 19 deletions(-) diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 2353472aefa..630aa313dc9 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -885,12 +885,22 @@ def publish_global_otel_v2_provider( """ global _published_v2_provider logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered) - attach_tenant_fan_out(logger.tracer_provider, logger.config) + 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 diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 70948f466ff..afd502962cf 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -356,6 +356,10 @@ _DATASTORE_ENDPOINT_KEYS: Final = frozenset({Server.ADDRESS, Server.PORT, DB.NAM # 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. @@ -402,11 +406,17 @@ def _is_tenant_owned_span(attributes: Mapping[str, AttributeValue]) -> bool: return any(key in attributes for key in _TENANT_OWNED_KEYS) -def _tenant_visible(key: str, database: bool, owned: bool) -> bool: +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 @@ -440,18 +450,24 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read 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. 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. + 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)} + { + 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 () @@ -1039,19 +1055,21 @@ def build_tracer_provider( _FAN_OUT_ATTACH_LOCK: Final = threading.Lock() -def attach_tenant_fan_out(provider: TracerProvider, config: OpenTelemetryV2Config | None = None) -> None: +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. ``config`` names the operator's own exporters so an - additive destination pointing at one of them is delivered once rather than twice. + 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(config))) + provider.add_span_processor(TenantFanOutSpanProcessor(operator_sinks=operator_sink_keys(*configs))) def deliverable_destinations( @@ -1076,18 +1094,18 @@ def deliverable_destinations( return fan_out.deliverable(destinations) if fan_out is not None else () -def operator_sink_keys(config: OpenTelemetryV2Config | None) -> frozenset[_SinkKey]: +def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]: """The accounts the operator's own exporters write to, in destination terms. - 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. + 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. """ - if config is None: - return frozenset() 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 ) diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py index c98144d1776..1799381bada 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -28,6 +28,7 @@ from litellm.integrations.otel.model.config import ( is_otel_v2_enabled, ) from litellm.integrations.otel.model.destination import OtelDestination +from litellm.integrations.otel.plumbing import providers as otel_providers from litellm.integrations.otel.plumbing.context import ( destination_backends, request_destinations, @@ -351,6 +352,31 @@ class TestRoutingMode: assert operator_sink_keys(config) == frozenset({self.OPERATOR_SINK}) + def test_operator_sink_keys_spans_every_config_it_is_handed(self): + first = OpenTelemetryV2Config( + exporters=( + ExporterSpec( + kind="otlp_http", + endpoint=self.OPERATOR_SINK[0], + headers="authorization=Basic op", + ), + ) + ) + second = OpenTelemetryV2Config( + exporters=( + ExporterSpec( + kind="otlp_http", + endpoint="https://otlp.arize.com/v1/traces", + headers="space_id=s,api_key=k", + ), + ) + ) + + assert operator_sink_keys(first, second) == { + self.OPERATOR_SINK, + _sink_key("https://otlp.arize.com/v1/traces", {"space_id": "s", "api_key": "k"}), + } + def test_a_team_pointing_at_a_credential_less_operator_exporter_still_gets_its_spans(self, monkeypatch): """Under additive the fan-out skips a destination the operator already writes to. An exporter the provider never built writes nothing, so skipping it would @@ -571,6 +597,50 @@ class TestFanOut: assert operator_db.status.description == unreachable assert [event.name for event in operator_db.events] == ["exception"] + @pytest.mark.parametrize("failure_status", ["guardrail_failed_to_respond", "failure"]) + def test_a_guardrails_failure_text_does_not_ride_along_to_the_tenant(self, failure_status): + dest_exporter, operator_exporter = InMemorySpanExporter(), InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(operator_exporter)) + provider.add_span_processor( + TenantFanOutSpanProcessor(processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter)) + ) + tracer = get_tracer(provider, "litellm") + unreachable = "Cannot connect to host guardrail.internal.example:9000" + verdict = '{"action": "block", "categories": ["pii"]}' + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + with tracer.start_as_current_span("POST /v1/chat/completions"): + with tracer.start_as_current_span("execute_guardrail pii") as down: + down.set_attributes( + { + "litellm.guardrail.name": "pii", + "litellm.guardrail.status": failure_status, + "litellm.guardrail.response": unreachable, + } + ) + with tracer.start_as_current_span("execute_guardrail toxicity") as up: + up.set_attributes( + { + "litellm.guardrail.name": "toxicity", + "litellm.guardrail.status": "guardrail_intervened", + "litellm.guardrail.response": verdict, + } + ) + + in_fresh_context(run) + + tenant = {s.name: s for s in dest_exporter.get_finished_spans()} + operator = {s.name: s for s in operator_exporter.get_finished_spans()} + assert dict(tenant["execute_guardrail pii"].attributes) == { + "litellm.guardrail.name": "pii", + "litellm.guardrail.status": failure_status, + } + assert "guardrail.internal.example" not in tenant["execute_guardrail pii"].to_json() + assert tenant["execute_guardrail toxicity"].attributes["litellm.guardrail.response"] == verdict + assert operator["execute_guardrail pii"].attributes["litellm.guardrail.response"] == unreachable + def test_the_callers_key_in_the_query_string_does_not_ride_along_to_the_tenant(self): """A Google AI Studio style request authenticates with ``?key=``, and the instrumentor stamps the full request URL on the server span. The @@ -925,6 +995,52 @@ class TestProviderWiring: assert kinds(published).count("TenantFanOutSpanProcessor") == 1 assert "TenantFanOutSpanProcessor" not in kinds(other) + @pytest.mark.parametrize("canonical", ["langfuse_otel", "arize"]) + def test_publishing_tells_the_fan_out_about_every_v2_loggers_account(self, monkeypatch, canonical): + monkeypatch.setenv("LITELLM_OTEL_TENANT_DESTINATION_MODE", "additive") + shared = InMemorySpanExporter() + monkeypatch.setattr(otel_providers, "_destination_processor", lambda _d: SimpleSpanProcessor(shared)) + accounts = { + "langfuse_otel": ( + "https://cloud.langfuse.com/api/public/otel/v1/traces", + "authorization=Basic op", + ), + "arize": ( + "https://otlp.arize.com/v1/traces", + "space_id=space-op,api_key=key-op", + ), + } + loggers = { + name: OpenTelemetryV2( + config=OpenTelemetryV2Config( + exporters=(ExporterSpec(kind="otlp_http", endpoint=endpoint, headers=headers),) + ), + callback_name=name, + tracer_provider=TracerProvider(), + ) + for name, (endpoint, headers) in accounts.items() + } + other = "arize" if canonical == "langfuse_otel" else "langfuse_otel" + published = publish_global_otel_v2_provider( + [loggers[other]], + lambda _p: None, + registered=loggers[canonical], + ) + + def destination(name, headers): + return OtelDestination(endpoint=accounts[name][0], headers=headers, callback_name=name) + + def run(destinations): + set_request_destinations(destinations) + emit(published.tracer_provider) + + in_fresh_context(run, (destination(canonical, dict(pair.split("=") for pair in accounts[canonical][1].split(","))),)) + in_fresh_context(run, (destination(other, dict(pair.split("=") for pair in accounts[other][1].split(","))),)) + assert shared.get_finished_spans() == (), "an account the operator already writes to was written twice" + + in_fresh_context(run, (destination(other, {"authorization": "Basic team"}),)) + assert [s.name for s in shared.get_finished_spans()] == ["chat gpt-4"] + def test_publishing_twice_does_not_double_export(self): config = OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.ARIZE_AX)]) logger = OpenTelemetryV2(config=config, callback_name="arize")