From 9bb728f354c8c84ac6b5a99d5dd5a1abdfe7d445 Mon Sep 17 00:00:00 2001 From: Yucheng He Date: Sat, 5 Sep 2026 05:14:05 -0700 Subject: [PATCH] fix(otel): keep the operator's database endpoint out of tenant traces A database span forwarded to a key or team destination carried the proxy's own Postgres host, port and schema, and on failure the Prisma error text naming them. The fan-out now hands tenants a view of each database span without those keys, its events or its status text, while the operator's own copy is untouched and model endpoints such as server.address on the LLM span still travel --- .../integrations/otel/plumbing/providers.py | 74 +++++++++++++---- .../otel/test_otel_v2_destinations.py | 81 ++++++++++++++++--- 2 files changed, 129 insertions(+), 26 deletions(-) diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index ff1da6c5f4b..64a45151ab5 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -4,7 +4,7 @@ import queue import threading import time from collections import OrderedDict -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal @@ -24,7 +24,7 @@ from opentelemetry.sdk._logs.export import ( ) from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor, TracerProvider +from opentelemetry.sdk.trace import Event, ReadableSpan, SpanProcessor, TracerProvider from opentelemetry.sdk.trace import Span as SDKSpan from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, @@ -35,13 +35,14 @@ from opentelemetry.sdk.trace.export import ( from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) -from opentelemetry.trace import Span, SpanKind, Tracer +from opentelemetry.trace import Span, SpanKind, Status, Tracer from opentelemetry.util.re import parse_env_headers +from opentelemetry.util.types import Attributes, AttributeValue from litellm._logging import verbose_logger from litellm._version import version as litellm_version from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config -from litellm.integrations.otel.model.semconv import LiteLLM +from litellm.integrations.otel.model.semconv import DB, Error, LiteLLM, LiteLLMError, Server from litellm.integrations.otel.model.spans import LiteLLMSpanKind from litellm.integrations.otel.plumbing.context import ( request_destinations, @@ -312,34 +313,75 @@ class _DrainPool: _shutdown_quietly(processor) -class _ResourceWrappedReadableSpan(ReadableSpan): - """A ``ReadableSpan`` view with an overridden Resource, leaving the original alone.""" +_NO_ATTRIBUTES: Final[Mapping[str, AttributeValue]] = MappingProxyType({}) +_DB_SYSTEM_KEYS: Final = frozenset({DB.SYSTEM_NAME, DB.SYSTEM_LEGACY}) +# Keys on a database span that describe the proxy's own datastore: its host, its +# port, its schema, and the Prisma error text that spells the first two out again. +_OPERATOR_INFRASTRUCTURE_KEYS: Final = frozenset( + {Server.ADDRESS, Server.PORT, DB.NAMESPACE, Error.MESSAGE, LiteLLMError.STACK_TRACE} +) - def __init__(self, inner: ReadableSpan, resource: Resource) -> None: + +class _TenantSpanView(ReadableSpan): + """A ``ReadableSpan`` view for one destination, leaving the operator's own span alone.""" + + def __init__( + self, + inner: ReadableSpan, + resource: Resource, + attributes: Attributes, + events: Sequence[Event], + status: Status, + ) -> None: super().__init__( name=inner.name, context=inner.context, parent=inner.parent, resource=resource, - attributes=inner.attributes, - events=inner.events, + attributes=attributes, + events=events, links=inner.links, kind=inner.kind, - status=inner.status, + status=status, start_time=inner.start_time, end_time=inner.end_time, instrumentation_scope=inner.instrumentation_scope, ) -def _with_destination_resource(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan: +def _is_database_span(span: ReadableSpan) -> bool: + attributes: Final = span.attributes or _NO_ATTRIBUTES + return any(key in attributes for key in _DB_SYSTEM_KEYS) + + +def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan: + """The view of ``span`` a tenant destination receives. + + A database span describes the operator's own Postgres rather than the tenant's + request, so its endpoint and its error text come off on the way out. The span + itself stays, so the tenant still gets the whole trace tree. + """ extra: Final = destination.resource_attributes - if not extra: + redacted: Final = _is_database_span(span) + if not extra and not redacted: return span - merged: Final = Resource.create( - {**dict(span.resource.attributes), **dict(extra)} # mutable-ok: the OTel SDK takes a concrete attribute mapping + resource: Final = ( + Resource.create( + {**dict(span.resource.attributes), **dict(extra)} # mutable-ok: the OTel SDK takes a concrete mapping + ) + if extra + else span.resource + ) + if not redacted: + return _TenantSpanView(span, resource, span.attributes, span.events, span.status) + attributes: Final = span.attributes or _NO_ATTRIBUTES + return _TenantSpanView( + span, + resource, + MappingProxyType({key: value for key, value in attributes.items() if key not in _OPERATOR_INFRASTRUCTURE_KEYS}), + (), + Status(span.status.status_code), ) - return _ResourceWrappedReadableSpan(span, merged) class TenantFanOutSpanProcessor(SpanProcessor): @@ -386,7 +428,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): if processor is None: continue try: - processor.on_end(_with_destination_resource(span, destination)) + processor.on_end(_for_destination(span, destination)) except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc) finally: 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 a603990f5e7..513e0512903 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -9,8 +9,15 @@ import pytest from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import Status, StatusCode import litellm +from litellm.integrations.otel import logger as otel_logger +from litellm.integrations.otel.logger import ( + OpenTelemetryV2, + fan_out_provider, + publish_global_otel_v2_provider, +) from litellm.integrations.otel.model.config import ( ExporterOwner, ExporterSpec, @@ -18,12 +25,6 @@ from litellm.integrations.otel.model.config import ( is_otel_v2_enabled, ) from litellm.integrations.otel.model.destination import OtelDestination -from litellm.integrations.otel import logger as otel_logger -from litellm.integrations.otel.logger import ( - OpenTelemetryV2, - fan_out_provider, - publish_global_otel_v2_provider, -) from litellm.integrations.otel.plumbing.context import ( destination_backends, request_destinations, @@ -38,15 +39,15 @@ from litellm.integrations.otel.plumbing.providers import ( operator_sink_keys, ) from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer +from litellm.integrations.otel.presets.arize import arize_preset from litellm.integrations.otel.presets.destinations import ( destination_capable_backends, destination_for, ) -from litellm.integrations.otel.presets.arize import arize_preset from litellm.integrations.otel.presets.langfuse import langfuse_preset from litellm.proxy._types import UserAPIKeyAuth -from litellm.types.utils import StandardCallbackDynamicParams from litellm.proxy.litellm_pre_call_utils import resolve_tenant_otel_destinations +from litellm.types.utils import StandardCallbackDynamicParams LANGFUSE_DEST = OtelDestination( endpoint="http://tenant.local/api/public/otel", @@ -504,6 +505,67 @@ class TestFanOut: assert {s.resource.attributes["service.name"] for s in dest.get_finished_spans()} == {"team-checkout"} + def test_the_operators_database_endpoint_does_not_ride_along_to_the_tenant(self): + """A database span describes the proxy's own Postgres, so the tenant gets the + span and its timing without the host, the port, the schema or the error text + that names them. The operator's own copy keeps everything.""" + 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 = "Can't reach database server at db.internal.example:15400" + + def run(): + set_request_destinations((LANGFUSE_DEST,)) + with tracer.start_as_current_span("postgres get_data") as db_span: + db_span.set_attributes( + { + "db.system.name": "postgresql", + "db.system": "postgresql", + "db.operation.name": "get_data", + "server.address": "db.internal.example", + "server.port": 15400, + "db.namespace": "litellm", + "error.type": "PrismaError", + "error.message": unreachable, + "litellm.provider.error.stack_trace": f"Traceback: {unreachable}", + } + ) + db_span.add_event("exception", {"exception.message": unreachable}) + db_span.set_status(Status(StatusCode.ERROR, unreachable)) + with tracer.start_as_current_span("chat claude-haiku") as llm_span: + llm_span.set_attribute("server.address", "api.anthropic.com") + + 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 set(tenant) == {"postgres get_data", "chat claude-haiku"}, "the tenant keeps the whole tree" + tenant_db = tenant["postgres get_data"] + assert dict(tenant_db.attributes) == { + "db.system.name": "postgresql", + "db.system": "postgresql", + "db.operation.name": "get_data", + "error.type": "PrismaError", + } + assert list(tenant_db.events) == [] + assert tenant_db.status.status_code is StatusCode.ERROR, "the tenant still sees that the call failed" + assert tenant_db.status.description is None + assert "db.internal.example" not in tenant_db.to_json() + assert tenant["chat claude-haiku"].attributes["server.address"] == "api.anthropic.com", ( + "only the operator's datastore is redacted, never the model endpoint" + ) + operator_db = operator["postgres get_data"] + assert operator_db.attributes["server.address"] == "db.internal.example" + assert operator_db.attributes["server.port"] == 15400 + assert operator_db.attributes["db.namespace"] == "litellm" + assert operator_db.attributes["error.message"] == unreachable + assert operator_db.status.description == unreachable + assert [event.name for event in operator_db.events] == ["exception"] + def test_a_destination_that_cannot_build_a_processor_is_skipped_quietly(self): """An unbuildable destination must not cost the caller its request.""" attempts = [] @@ -1442,8 +1504,7 @@ class TestEvictionSafety: fan_out._release(fan_out._acquire(self._dest(index))) threads = [ - threading.Thread(target=shed, args=(_MAX_CACHED_DESTINATION_PROCESSORS + index,)) - for index in range(16) + threading.Thread(target=shed, args=(_MAX_CACHED_DESTINATION_PROCESSORS + index,)) for index in range(16) ] for thread in threads: thread.start()