scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 85

fix(otel): keep the operator's database endpoint out of tenant traces

Browse source 
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
This commit is contained in:
Yucheng He 2026-09-05 05:14:05 -07:00
parent 0cce313796
commit 9bb728f354
2 changed files with 129 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
litellm/integrations/otel/plumbing/providers.py
View file

@ -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:

81
tests/test_litellm/integrations/otel/test_otel_v2_destinations.py
View file

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