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()