fix(otel): close final tenant routing gaps

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

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
     )

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=<virtual 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")