diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index 6806188c97c..5bd8aca55fa 100644 --- a/litellm/integrations/callback_configs.json +++ b/litellm/integrations/callback_configs.json @@ -259,6 +259,13 @@ "ui_name": "Tracing Environment", "description": "Langfuse tracing environment (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT)", "required": false + }, + "langfuse_span_scope": { + "type": "select", + "ui_name": "Span Scope", + "description": "full sends the whole request trace, llm_only sends just the model-call spans", + "options": ["full", "llm_only"], + "required": false } }, "description": "Langfuse v3 OTEL Logging Integration" diff --git a/litellm/integrations/otel/model/config.py b/litellm/integrations/otel/model/config.py index 5bda66ed618..5447a8ee80a 100644 --- a/litellm/integrations/otel/model/config.py +++ b/litellm/integrations/otel/model/config.py @@ -12,6 +12,7 @@ from litellm.integrations.otel.model.baggage import ( DEFAULT_BAGGAGE_METADATA_KEYS, DEFAULT_BAGGAGE_TEAM_METADATA_KEYS, ) +from litellm.types.utils import OtelSpanScope #: Master feature-flag env var. The logger is inert until this is truthy. OTEL_V2_ENV: Final = "LITELLM_OTEL_V2" @@ -163,6 +164,15 @@ class OpenTelemetryV2Config(BaseSettings): validation_alias=AliasChoices("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"), ) legacy_compat: bool = Field(default=True, validation_alias=AliasChoices("LITELLM_OTEL_LEGACY_COMPAT")) + langfuse_span_scope: OtelSpanScope = Field( + default="full", + validation_alias=AliasChoices("langfuse_span_scope", "LITELLM_OTEL_LANGFUSE_SPAN_SCOPE"), + description=( + "``llm_only`` keeps just the model-call spans on the operator's own Langfuse " + "exporter (the spec whose owner is ``langfuse_otel``). Other exporters and " + "key/team destinations are not affected." + ), + ) # ----- explicit multi-destination / vocabulary configuration ------------ # @@ -245,6 +255,13 @@ class OpenTelemetryV2Config(BaseSettings): return value.lower() return value + @field_validator("langfuse_span_scope", mode="before") + @classmethod + def _normalize_langfuse_span_scope(cls, value: object) -> object: + if isinstance(value, str): + return value.strip().lower() + return value + @field_validator( "baggage_promoted_keys", "baggage_metadata_keys", diff --git a/litellm/integrations/otel/model/destination.py b/litellm/integrations/otel/model/destination.py index 299253cac77..c9c035f24a1 100644 --- a/litellm/integrations/otel/model/destination.py +++ b/litellm/integrations/otel/model/destination.py @@ -10,6 +10,8 @@ from urllib.parse import quote from pydantic import BaseModel, ConfigDict, Field +from litellm.types.utils import OtelSpanScope + class OtelDestination(BaseModel): model_config = ConfigDict(frozen=True) @@ -25,6 +27,10 @@ class OtelDestination(BaseModel): "scheme: Arize's ``https://otlp.arize.com/v1`` is gRPC." ), ) + span_scope: OtelSpanScope = Field( + default="full", + description="``llm_only`` keeps just the model-call spans; the rest of the request tree is not forwarded.", + ) def header_string(self) -> str: """Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects. @@ -37,7 +43,12 @@ class OtelDestination(BaseModel): return ",".join(f"{key}={quote(value, safe='')}" for key, value in self.headers.items()) def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]: - """Identity for processor reuse, so one destination means one exporter.""" + """Identity for processor reuse, so one destination means one exporter. + + ``span_scope`` is left out on purpose: the scope decides which spans reach the + processor, not how the processor exports them, so a full and an ``llm_only`` + view of the same account share one exporter. + """ return ( self.endpoint, tuple(sorted(self.headers.items())), diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 81f22c8c642..2c4375ce5f7 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -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, Status, Tracer +from opentelemetry.trace import Span, SpanContext, 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.mappers.langfuse import LANGFUSE_TRACE_NAME +from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config from litellm.integrations.otel.model.semconv import ( DB, MCP, @@ -63,6 +64,7 @@ if TYPE_CHECKING: from opentelemetry.sdk.metrics.export import MetricReader from litellm.integrations.otel.model.destination import OtelDestination + from litellm.types.utils import OtelSpanScope _SPAN_KIND_BY_ROLE_KIND: Final[dict[LiteLLMSpanKind, SpanKind]] = { LiteLLMSpanKind.SERVER: SpanKind.SERVER, @@ -379,8 +381,8 @@ _URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"}) _URL_QUERY_KEY: Final = "url.query" -class _TenantSpanView(ReadableSpan): - """A ``ReadableSpan`` view for one destination, leaving the operator's own span alone.""" +class _SpanView(ReadableSpan): + """A ``ReadableSpan`` view for one exporter, leaving the span every other exporter sees alone.""" def __init__( self, @@ -389,11 +391,12 @@ class _TenantSpanView(ReadableSpan): attributes: Attributes, events: Sequence[Event], status: Status, + parent: SpanContext | None, ) -> None: super().__init__( name=inner.name, context=inner.context, - parent=inner.parent, + parent=parent, resource=resource, attributes=attributes, events=events, @@ -414,6 +417,39 @@ def _is_tenant_owned_span(attributes: Mapping[str, AttributeValue]) -> bool: return any(key in attributes for key in _TENANT_OWNED_KEYS) +def is_llm_call_span(span: ReadableSpan) -> bool: + """Whether ``span`` is the model call itself. + + The GenAI mapper stamps ``gen_ai.operation.name`` on the model call and on the + MCP tool call, so the MCP method name tells the two apart. Guardrail, request + root, auth and database spans never carry the operation name; ``gen_ai.request.model`` + would not do, since baggage promotes it onto every child span. + """ + attributes: Final = span.attributes or _NO_ATTRIBUTES + return GenAI.OPERATION_NAME in attributes and MCP.METHOD_NAME not in attributes + + +def _in_scope(span: ReadableSpan, scope: "OtelSpanScope") -> bool: + return scope == "full" or is_llm_call_span(span) + + +def _scoped(span: ReadableSpan, scope: "OtelSpanScope") -> ReadableSpan: + """Under ``llm_only`` the model call is the only span the exporter gets, so it goes out as the + trace's root (its parent is the request span that is held back) and, unless the caller named the + trace, its own name doubles as ``langfuse.trace.name`` so Langfuse does not show "Unnamed trace".""" + if scope == "full": + return span + attributes: Final = span.attributes or _NO_ATTRIBUTES + named: Final = ( + attributes + if LANGFUSE_TRACE_NAME in attributes + else MappingProxyType({**attributes, LANGFUSE_TRACE_NAME: span.name}) + ) + if span.parent is None and named is attributes: + return span + return _SpanView(span, span.resource, named, span.events, span.status, parent=None) + + def _guardrail_unreachable(attributes: Mapping[str, AttributeValue]) -> bool: return attributes.get(LiteLLM.GUARDRAIL_STATUS) in _GUARDRAIL_UNREACHABLE_STATUSES @@ -484,7 +520,7 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read return span resource: Final = span.resource.merge(Resource(extra)) if extra else span.resource status: Final = span.status if owned else Status(span.status.status_code) - return _TenantSpanView(span, resource, kept, events, status) + return _SpanView(span, resource, kept, events, status, parent=span.parent) class TenantFanOutSpanProcessor(SpanProcessor): @@ -507,7 +543,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): self, processor_factory: 'Callable[["OtelDestination"], SpanProcessor | None] | None' = None, shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS, - operator_sinks: frozenset[_SinkKey] = frozenset(), + operator_sinks: 'Mapping[_SinkKey, "OtelSpanScope"]' = MappingProxyType({}), pending_drains: int = _MAX_PENDING_DRAINS, drain_pool: _DrainPool | None = None, ) -> None: @@ -527,29 +563,36 @@ class TenantFanOutSpanProcessor(SpanProcessor): def on_end(self, span: ReadableSpan) -> None: suppressed: Final = suppressed_backends() for destination in request_destinations(): - if self._operator_already_writes(destination, suppressed): + if self._operator_already_writes(span, destination, suppressed) or not _in_scope( + span, destination.span_scope + ): continue processor = self._acquire(destination) # rebind-ok: loop variable; pyright forbids Final in a loop if processor is None: continue try: - processor.on_end(_for_destination(span, destination)) + processor.on_end(_scoped(_for_destination(span, destination), destination.span_scope)) 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: self._release(processor) - def _operator_already_writes(self, destination: "OtelDestination", suppressed: frozenset[str]) -> bool: + def _operator_already_writes( + self, span: ReadableSpan, destination: "OtelDestination", suppressed: frozenset[str] + ) -> bool: """Whether the operator's own exporter is sending this span to the same account. Only reachable under ``additive``, where nothing is suppressed: a team that names the operator's own project would otherwise have every span written - there twice, once by the operator's exporter and once by the fan-out. + there twice, once by the operator's exporter and once by the fan-out. The + operator's exporter may itself be narrowed to the model calls, in which case + the rest of the tree is still the fan-out's to deliver. """ - return ( - destination.callback_name not in suppressed - and _sink_key(destination.endpoint, destination.headers) in self._operator_sinks - ) + sink: Final = _sink_key(destination.endpoint, destination.headers) + if destination.callback_name in suppressed or sink is None: + return False + operator_scope: Final = self._operator_sinks.get(sink) + return operator_scope is not None and _in_scope(span, operator_scope) def shutdown(self) -> None: """Close every destination processor, once the spans in flight have landed. @@ -753,19 +796,43 @@ class _OverriddenBackendFilter(SpanProcessor): Under ``additive`` mode nothing is suppressed, so the wrapper passes every span straight through and the operator keeps its copy. + + ``scope`` narrows what the exporter receives independently of that: under + ``llm_only`` the model-call spans go through as trace roots and the rest of the + tree is held back, unless a destination of the request names ``sink``, the account + this exporter writes to, with a wider scope: the fan-out then delivers the rest of + the tree there and the model call keeps its place in it. """ - def __init__(self, inner: SpanProcessor, owner: str) -> None: + def __init__( + self, + inner: SpanProcessor, + owner: str | None, + scope: "OtelSpanScope" = "full", + sink: _SinkKey | None = None, + ) -> None: self._inner: Final = inner self._owner: Final = owner + self._scope: Final = scope + self._sink: Final = sink def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: self._inner.on_start(span, parent_context) def on_end(self, span: ReadableSpan) -> None: - if self._owner in suppressed_backends(): + if self._owner in suppressed_backends() or not _in_scope(span, self._scope): return - self._inner.on_end(span) + self._inner.on_end(_scoped(span, self._account_scope())) + + def _account_scope(self) -> "OtelSpanScope": + if self._scope == "full" or self._sink is None: + return self._scope + shared: Final = tuple( + destination.span_scope + for destination in request_destinations() + if _sink_key(destination.endpoint, destination.headers) == self._sink + ) + return _widest((self._scope, *shared)) def shutdown(self) -> None: self._inner.shutdown() @@ -1040,6 +1107,9 @@ def build_tracer_provider( tenant is a separate job, done once by :func:`attach_tenant_fan_out`. The per-tenant providers this same function builds must leave it off, or they would filter out the very spans they exist to carry. + + ``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel`` + alone; a collector or any other backend in the same config keeps the full tree. """ provider: Final = TracerProvider(resource=build_resource(config)) if baggage_processor is None: @@ -1060,9 +1130,13 @@ def build_tracer_provider( exp, (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), ) - owner = spec.owner.value if spec.owner is not None else None + owner = spec.owner.value if tenant_overrides and spec.owner is not None else None + scope = _operator_scope(config, spec) + sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None provider.add_span_processor( - _OverriddenBackendFilter(processor, owner) if tenant_overrides and owner is not None else processor + _OverriddenBackendFilter(processor, owner, scope, sink) + if owner is not None or scope != "full" + else processor ) return provider @@ -1084,7 +1158,7 @@ def attach_tenant_fan_out(provider: TracerProvider, *configs: OpenTelemetryV2Con 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(*configs))) + provider.add_span_processor(TenantFanOutSpanProcessor(operator_sinks=operator_sink_scopes(*configs))) def deliverable_destinations( @@ -1109,7 +1183,7 @@ def deliverable_destinations( return fan_out.deliverable(destinations) if fan_out is not None else () -def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]: +def operator_sink_scopes(*configs: OpenTelemetryV2Config) -> 'Mapping[_SinkKey, "OtelSpanScope"]': """The accounts the operator's own exporters write to, in destination terms. Every v2 logger's config counts, since each logger exports through its own @@ -1118,12 +1192,21 @@ def operator_sink_keys(*configs: OpenTelemetryV2Config) -> frozenset[_SinkKey]: 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. """ - return frozenset( - key + scoped: Final[tuple[tuple[_SinkKey, OtelSpanScope], ...]] = tuple( + (key, _operator_scope(config, spec)) 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 ) + return MappingProxyType({key: _widest(scope for other, scope in scoped if other == key) for key, _ in scoped}) + + +def _operator_scope(config: OpenTelemetryV2Config, spec: ExporterSpec) -> "OtelSpanScope": + return config.langfuse_span_scope if spec.owner is ExporterOwner.LANGFUSE_OTEL else "full" + + +def _widest(scopes: "Iterable[OtelSpanScope]") -> "OtelSpanScope": + return "full" if any(scope == "full" for scope in scopes) else "llm_only" def _exports_to_the_wire(spec: ExporterSpec) -> bool: diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index f78d18d943c..b2d1f50f370 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -374,10 +374,8 @@ class TenantTracerCache: self._routed_exporter(spec, credential_headers, project_headers, endpoint) for spec in self._config.exporters ] - update: Final = ( - {"exporters": exporters} if service_name is None else {"exporters": exporters, "service_name": service_name} - ) - return self._config.model_copy(update=update) + routed: Final = self._config.model_copy(update={"exporters": exporters, "langfuse_span_scope": "full"}) + return routed if service_name is None else routed.model_copy(update={"service_name": service_name}) def _routed_exporter( self, diff --git a/litellm/integrations/otel/presets/destinations.py b/litellm/integrations/otel/presets/destinations.py index 2bf9bfa5261..63801e623af 100644 --- a/litellm/integrations/otel/presets/destinations.py +++ b/litellm/integrations/otel/presets/destinations.py @@ -15,7 +15,7 @@ import litellm from litellm._logging import verbose_logger from litellm.integrations.otel.model.destination import OtelDestination from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host -from litellm.types.utils import StandardCallbackDynamicParams +from litellm.types.utils import OtelSpanScope, StandardCallbackDynamicParams #: An endpoint plus the OTLP transport to reach it with, or ``None`` when the backend #: names no destination. The transport is ``None`` where the backend has only one. @@ -111,6 +111,12 @@ _REQUIRED_HEADERS_BY_CALLBACK: Final[Mapping[str, frozenset[str]]] = MappingProx _NO_ATTRS: Final[Mapping[str, str]] = MappingProxyType({}) +def _span_scope(callback_name: str, params: StandardCallbackDynamicParams) -> OtelSpanScope: + if callback_name != "langfuse_otel": + return "full" + return params.get("langfuse_span_scope") or "full" + + def destination_capable_backends() -> frozenset[str]: """Backends a key or team can point at its own account.""" from litellm.integrations.otel.presets import DYNAMIC_HEADERS_BY_CALLBACK @@ -149,4 +155,5 @@ def destination_for( resource_attributes=MappingProxyType({"service.name": service_name}) if service_name else _NO_ATTRS, callback_name=callback_name, protocol=protocol, + span_scope=_span_scope(callback_name, params), ) diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py index 65c5b0d9799..e4744079622 100644 --- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py +++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py @@ -2,7 +2,7 @@ import re from collections.abc import Iterator, Mapping from typing import Any, Final -from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams +from litellm.types.utils import OTEL_SPAN_SCOPES, TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams _CLIENT_CALLBACK_METADATA_SLOTS: Final[tuple[str, ...]] = ("litellm_metadata", "metadata") @@ -62,6 +62,11 @@ def validate_langfuse_environment_value(value: str) -> None: ) +def validate_langfuse_span_scope_value(value: str) -> None: + if value not in OTEL_SPAN_SCOPES: + raise ValueError(f"Invalid langfuse_span_scope {value!r}: must be one of {sorted(OTEL_SPAN_SCOPES)}") + + # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict _supported_callback_params: Final[tuple[str, ...]] = ( "langfuse_public_key", diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 56b3f590210..d24e451cfc1 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -24,6 +24,7 @@ from litellm._uuid import uuid from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( validate_langfuse_environment_value, + validate_langfuse_span_scope_value, validate_no_callback_env_reference, ) from litellm.types.integrations.compression_interception import ( @@ -2222,6 +2223,8 @@ class AddTeamCallback(LiteLLMPydanticObjectBase): validate_no_callback_env_reference(key, callback_vars[key], source="key/team callback metadata") if key == "langfuse_environment": validate_langfuse_environment_value(callback_vars[key]) + if key == "langfuse_span_scope": + validate_langfuse_span_scope_value(callback_vars[key]) return values diff --git a/litellm/proxy/common_utils/callback_config_validation.py b/litellm/proxy/common_utils/callback_config_validation.py index c9d97068313..30c4ab31d6f 100644 --- a/litellm/proxy/common_utils/callback_config_validation.py +++ b/litellm/proxy/common_utils/callback_config_validation.py @@ -11,14 +11,18 @@ from typing import Final _NEWRELIC_CALLBACK: Final = "newrelic" _NEWRELIC_VAR_PREFIX: Final = "newrelic_" +_LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel" +_LANGFUSE_SPAN_SCOPE_VAR: Final = "langfuse_span_scope" def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None: if not callback_vars: return None - env_error: Final = _langfuse_environment_error(callback_vars) - if env_error is not None: - return env_error + langfuse_error: Final = _langfuse_environment_error(callback_vars) or _langfuse_span_scope_error( + callback_name, callback_vars + ) + if langfuse_error is not None: + return langfuse_error if callback_name != _NEWRELIC_CALLBACK: return None return _newrelic_config_error(callback_vars) @@ -44,6 +48,25 @@ def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None: return None +def _langfuse_span_scope_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None: + value: Final = callback_vars.get(_LANGFUSE_SPAN_SCOPE_VAR) + if value is None: + return None + if callback_name != _LANGFUSE_OTEL_CALLBACK: + return ( + f"{_LANGFUSE_SPAN_SCOPE_VAR} applies to the {_LANGFUSE_OTEL_CALLBACK} callback only, not {callback_name!r}" + ) + from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( + validate_langfuse_span_scope_value, + ) + + try: + validate_langfuse_span_scope_value(value) + except ValueError as e: + return str(e) + return None + + # Which credential family a dynamic variable belongs to. The families are the # integrations that share one account: every langfuse_* variable configures the # same Langfuse project whether it rides the classic callback or the OTel one, @@ -63,13 +86,18 @@ _VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType( } ) +_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset({_LANGFUSE_SPAN_SCOPE_VAR}) + def _family_of(var: str) -> str | None: """The credential family ``var`` configures, or ``None`` if it configures none. ``turn_off_message_logging`` and friends belong to no backend, so they carry - no credentials anyone could redirect. + no credentials anyone could redirect. ``langfuse_span_scope`` shares the Langfuse + prefix but is a fixed enum choosing what the family exports, not where to. """ + if var in _FAMILY_OPTION_VARS: + return None return next((family for prefix, family in _VAR_FAMILIES.items() if var.startswith(prefix)), None) @@ -129,6 +157,24 @@ def cross_entry_family_error( ) +def conflicting_span_scope_error( + callback_vars: Mapping[str, str] | None, + stored_vars_by_entry: Sequence[Mapping[str, str]], +) -> str | None: + incoming: Final = None if callback_vars is None else callback_vars.get(_LANGFUSE_SPAN_SCOPE_VAR) + if incoming is None: + return None + return next( + ( + f"{_LANGFUSE_SPAN_SCOPE_VAR} is already set to {stored!r} by another callback entry. " + f"Every entry shares one scope: remove that entry or send the same value." + for entry in stored_vars_by_entry + if (stored := entry.get(_LANGFUSE_SPAN_SCOPE_VAR)) not in (None, incoming) + ), + None, + ) + + def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None: """Validate every ``logging`` entry of a team/key metadata payload.""" if not metadata: @@ -136,23 +182,34 @@ def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str entries: Final = metadata.get("logging") if not isinstance(entries, Sequence) or isinstance(entries, (str, bytes)): return None + entry_vars: Final = tuple(_entry_callback_vars(entry) for entry in entries) return next( - (error for error in (_logging_entry_error(entry) for entry in entries) if error is not None), + ( + error + for error in ( + *(_logging_entry_error(entry) for entry in entries), + *(conflicting_span_scope_error(entry_vars[i], entry_vars[:i]) for i in range(len(entry_vars))), + ) + if error is not None + ), None, ) +def _entry_callback_vars(entry: object) -> Mapping[str, str]: + callback_vars: Final = entry.get("callback_vars") if isinstance(entry, Mapping) else None + if not isinstance(callback_vars, Mapping): + return MappingProxyType({}) + return MappingProxyType({str(key): str(value) for key, value in callback_vars.items()}) + + def _logging_entry_error(entry: object) -> str | None: if not isinstance(entry, Mapping): return None callback_name: Final = entry.get("callback_name") - callback_vars: Final = entry.get("callback_vars") - if not isinstance(callback_name, str) or not isinstance(callback_vars, Mapping): + if not isinstance(callback_name, str) or not isinstance(entry.get("callback_vars"), Mapping): return None - return callback_config_error( - callback_name, - MappingProxyType({str(key): str(value) for key, value in callback_vars.items()}), - ) + return callback_config_error(callback_name, _entry_callback_vars(entry)) def _newrelic_config_error(callback_vars: Mapping[str, str]) -> str | None: diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index 1932e89717b..ac13b6150b7 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -31,6 +31,7 @@ from litellm.proxy._types import ( from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.common_utils.callback_config_validation import ( callback_config_error, + conflicting_span_scope_error, cross_entry_family_error, ) from litellm.proxy.common_utils.callback_utils import ( @@ -283,6 +284,7 @@ async def add_team_callbacks( - langfuse_secret: The secret for the Langfuse callback - langfuse_host: The host for the Langfuse callback - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT) + - langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans - gcs_bucket_name: The name of the GCS bucket - gcs_path_service_account: The path to the GCS service account - langsmith_api_key: The API key for the Langsmith callback @@ -343,6 +345,16 @@ async def add_team_callbacks( if team_callback_settings is None or not isinstance(team_callback_settings, list): team_callback_settings = [] + # Decrypted, because the checks compare the incoming values against + # the stored ones and the credentials are encrypted at rest. + decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging") + stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else () + stored_entry_vars: Final = [ # mutable-ok: read-only input to the checks, never stored + entry.get("callback_vars") or {} for entry in stored_entries + ] + scope_error: Final = conflicting_span_scope_error(data.callback_vars, stored_entry_vars) + if scope_error is not None: + raise _callback_config_error(scope_error) # One entry has to own a credential family end to end. The entries are # flattened into one dict before a request reads them, so an entry # naming only a destination would pair with a key written on another @@ -351,13 +363,6 @@ async def add_team_callbacks( # fine, which is how one integration covers both events. Proxy admins # are exempt: they already hold every credential the proxy has. if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: - # Decrypted, because the check compares the incoming values against - # the stored ones and the credentials are encrypted at rest. - decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging") - stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else () - stored_entry_vars: Final = [ # mutable-ok: read-only input to the check, never stored - entry.get("callback_vars") or {} for entry in stored_entries - ] family_error: Final = cross_entry_family_error(data.callback_vars, stored_entry_vars) if family_error is not None: raise HTTPException( diff --git a/litellm/types/utils.py b/litellm/types/utils.py index b725acf6906..cc9b7931f56 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3527,6 +3527,10 @@ OPENAI_RESPONSE_HEADERS: Final = [ ] +OtelSpanScope = Literal["full", "llm_only"] +OTEL_SPAN_SCOPES: Final[frozenset[str]] = frozenset(get_args(OtelSpanScope)) + + class StandardCallbackDynamicParams(TypedDict, total=False): # Langfuse dynamic params langfuse_public_key: str | None @@ -3534,6 +3538,7 @@ class StandardCallbackDynamicParams(TypedDict, total=False): langfuse_secret_key: str | None langfuse_host: str | None langfuse_environment: ReadOnly[str | None] + langfuse_span_scope: ReadOnly[OtelSpanScope | None] # Langfuse prompt version langfuse_prompt_version: int | 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 67695d5aed8..9cb3dbb9deb 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -42,7 +42,7 @@ from litellm.integrations.otel.plumbing.providers import ( _sink_key, build_tracer_provider, deliverable_destinations, - operator_sink_keys, + operator_sink_scopes, ) from litellm.integrations.otel.plumbing.routing import TenantTracerCache, get_tracer from litellm.integrations.otel.presets.arize import arize_preset @@ -82,6 +82,12 @@ def isolate_published_provider(monkeypatch): monkeypatch.setattr(otel_logger, "_published_v2_provider", None) +@pytest.fixture(autouse=True) +def forget_otel_v2_flag_after_each_test(): + yield + is_otel_v2_enabled.cache_clear() + + def in_fresh_context(fn, *args): """Run ``fn`` in its own context so one test's destinations never leak.""" return contextvars.copy_context().run(fn, *args) @@ -237,7 +243,7 @@ class TestRoutingMode: provider.add_span_processor( TenantFanOutSpanProcessor( processor_factory=lambda _d: SimpleSpanProcessor(shared), - operator_sinks=frozenset({self.OPERATOR_SINK}), + operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}), ) ) @@ -260,7 +266,7 @@ class TestRoutingMode: provider.add_span_processor( TenantFanOutSpanProcessor( processor_factory=lambda _d: SimpleSpanProcessor(shared), - operator_sinks=frozenset({self.OPERATOR_SINK}), + operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}), ) ) @@ -277,7 +283,7 @@ class TestRoutingMode: provider.add_span_processor( TenantFanOutSpanProcessor( processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter), - operator_sinks=frozenset({self.OPERATOR_SINK}), + operator_sinks=MappingProxyType({self.OPERATOR_SINK: "full"}), ) ) @@ -330,7 +336,7 @@ class TestRoutingMode: assert global_exporter.get_finished_spans() == () - def test_operator_sink_keys_skips_an_exporter_with_no_endpoint_of_its_own(self): + def test_operator_sink_scopes_skips_an_exporter_with_no_endpoint_of_its_own(self): """Such an exporter resolves its endpoint from the environment at export time, so it has no identity to compare a destination against.""" config = OpenTelemetryV2Config( @@ -340,9 +346,9 @@ class TestRoutingMode: ) ) - assert operator_sink_keys(config) == frozenset({self.OPERATOR_SINK}) + assert dict(operator_sink_scopes(config)) == {self.OPERATOR_SINK: "full"} - def test_operator_sink_keys_skips_exporters_that_never_reach_the_wire(self): + def test_operator_sink_scopes_skips_exporters_that_never_reach_the_wire(self): """A console kind ignores the endpoint and a header-gated spec with no credentials is dropped when the provider is built, so treating either as an account the operator writes to would silently withhold a team's own spans @@ -355,9 +361,9 @@ class TestRoutingMode: ) ) - assert operator_sink_keys(config) == frozenset({self.OPERATOR_SINK}) + assert dict(operator_sink_scopes(config)) == {self.OPERATOR_SINK: "full"} - def test_operator_sink_keys_spans_every_config_it_is_handed(self): + def test_operator_sink_scopes_spans_every_config_it_is_handed(self): first = OpenTelemetryV2Config( exporters=( ExporterSpec( @@ -377,11 +383,27 @@ class TestRoutingMode: ) ) - assert operator_sink_keys(first, second) == { - self.OPERATOR_SINK, - _sink_key("https://otlp.arize.com/v1/traces", {"space_id": "s", "api_key": "k"}), + assert dict(operator_sink_scopes(first, second)) == { + self.OPERATOR_SINK: "full", + _sink_key("https://otlp.arize.com/v1/traces", {"space_id": "s", "api_key": "k"}): "full", } + @pytest.mark.parametrize("langfuse_first", [False, True]) + def test_two_operator_exporters_on_one_account_record_the_wider_scope(self, langfuse_first): + langfuse = ExporterSpec( + kind="otlp_http", + endpoint=self.OPERATOR_SINK[0], + headers="authorization=Basic op", + owner=ExporterOwner.LANGFUSE_OTEL, + ) + collector = ExporterSpec(kind="otlp_http", endpoint=self.OPERATOR_SINK[0], headers="authorization=Basic op") + config = OpenTelemetryV2Config( + langfuse_span_scope="llm_only", + exporters=(langfuse, collector) if langfuse_first else (collector, langfuse), + ) + + assert dict(operator_sink_scopes(config)) == {self.OPERATOR_SINK: "full"} + 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 @@ -397,7 +419,7 @@ class TestRoutingMode: provider.add_span_processor( TenantFanOutSpanProcessor( processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter), - operator_sinks=operator_sink_keys(config), + operator_sinks=operator_sink_scopes(config), ) ) @@ -416,7 +438,7 @@ class TestRoutingMode: monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-op") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-op") monkeypatch.setattr(litellm, "provider_url_destination_allowed_hosts", ["lf.internal"], raising=False) - operator = operator_sink_keys(langfuse_preset()) + operator = operator_sink_scopes(langfuse_preset()) def sink(public_key, secret_key): destination = destination_for( @@ -450,7 +472,7 @@ class TestRoutingMode: monkeypatch.setenv("ARIZE_SPACE_ID", "space-op") monkeypatch.setenv("ARIZE_API_KEY", "key-op") monkeypatch.delenv("ARIZE_SPACE_KEY", raising=False) - operator = operator_sink_keys(arize_preset()) + operator = operator_sink_scopes(arize_preset()) def sink(space, api_key): destination = destination_for( @@ -1039,7 +1061,9 @@ class TestProviderWiring: 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(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" @@ -1403,6 +1427,480 @@ class TestDestinationResolution: assert parse_headers(destination.header_string())["authorization"] == destination.headers["Authorization"] +LLM_ONLY_DEST = OtelDestination( + endpoint="http://tenant.local/api/public/otel", + headers={"Authorization": "Basic dGVuYW50"}, + callback_name="langfuse_otel", + span_scope="llm_only", +) + +#: Every span kind the proxy emits for one chat request, plus the two spans that +#: look like a model call to a naive classifier: the MCP tool call carries +#: ``gen_ai.operation.name`` too, and baggage promotes ``gen_ai.request.model`` +#: onto children that are not the call. +REQUEST_TREE = frozenset( + { + "POST /v1/chat/completions", + "auth /v1/chat/completions", + "postgres SELECT", + "redis GET", + "execute_guardrail pii", + "tools/call get_weather", + "chat gpt-4", + "chat claude-haiku", + "cost_tracking", + } +) +LLM_SPANS = frozenset({"chat gpt-4", "chat claude-haiku"}) +TRACE_CONTROLS = MappingProxyType( + { + "langfuse.observation.type": "generation", + "langfuse.trace.name": "checkout", + "user.id": "user-7", + "session.id": "sess-1", + "langfuse.trace.tags": ("beta", "eu"), + } +) + + +def request_tree(provider: TracerProvider) -> None: + tracer = get_tracer(provider, "litellm") + with tracer.start_as_current_span("POST /v1/chat/completions"): + with tracer.start_as_current_span("auth /v1/chat/completions"): + with tracer.start_as_current_span("postgres SELECT") as db: + db.set_attribute("db.system", "postgresql") + with tracer.start_as_current_span("redis GET") as cache: + cache.set_attribute("db.system", "redis") + with tracer.start_as_current_span("execute_guardrail pii") as guard: + guard.set_attributes({"litellm.guardrail.name": "pii", "litellm.guardrail.status": "success"}) + with tracer.start_as_current_span("tools/call get_weather") as tool: + tool.set_attributes({"gen_ai.operation.name": "execute_tool", "mcp.method.name": "tools/call"}) + with tracer.start_as_current_span("chat gpt-4") as llm: + llm.set_attributes({"gen_ai.operation.name": "chat", "gen_ai.request.model": "gpt-4", **TRACE_CONTROLS}) + with tracer.start_as_current_span("cost_tracking") as child: + child.set_attribute("gen_ai.request.model", "gpt-4") + with tracer.start_as_current_span("chat claude-haiku") as retry: + retry.set_attributes({"gen_ai.operation.name": "chat", "gen_ai.request.model": "claude-haiku"}) + + +def names(exporter: InMemorySpanExporter) -> frozenset[str]: + return frozenset(s.name for s in exporter.get_finished_spans()) + + +class TestSpanScope: + @staticmethod + def _additive(monkeypatch): + monkeypatch.setattr(litellm, "otel_tenant_destination_mode", "additive", raising=False) + + @staticmethod + def _run(provider, destinations): + def run(): + set_request_destinations(destinations) + request_tree(provider) + + in_fresh_context(run) + + @staticmethod + def _operator_provider(operator_exporter, dest_exporter, scope="full"): + provider = TracerProvider() + provider.add_span_processor( + _OverriddenBackendFilter(SimpleSpanProcessor(operator_exporter), "langfuse_otel", scope) + ) + provider.add_span_processor( + TenantFanOutSpanProcessor(processor_factory=lambda _d: SimpleSpanProcessor(dest_exporter)) + ) + return provider + + def test_off_and_off_is_the_full_tree_on_both_sides(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant), (LANGFUSE_DEST,)) + + assert names(operator) == REQUEST_TREE + assert names(tenant) == REQUEST_TREE + + def test_a_tenant_asking_for_llm_only_gets_just_the_model_calls(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant), (LLM_ONLY_DEST,)) + + assert names(tenant) == LLM_SPANS + assert names(operator) == REQUEST_TREE, "the tenant's scope must not narrow the operator's exporter" + + def test_an_operator_asking_for_llm_only_keeps_the_tenants_tree_whole(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LANGFUSE_DEST,)) + + assert names(operator) == LLM_SPANS + assert names(tenant) == REQUEST_TREE, "the operator's scope must not narrow a tenant destination" + + def test_both_on_narrows_both(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LLM_ONLY_DEST,)) + + assert names(operator) == LLM_SPANS + assert names(tenant) == LLM_SPANS + + def test_an_operator_scope_does_not_undo_the_override(self): + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LLM_ONLY_DEST,)) + + assert operator.get_finished_spans() == () + assert names(tenant) == LLM_SPANS + + @staticmethod + def _same_account_provider(shared, operator_scope): + provider = TracerProvider() + provider.add_span_processor( + _OverriddenBackendFilter( + SimpleSpanProcessor(shared), "langfuse_otel", operator_scope, TestRoutingMode.OPERATOR_SINK + ) + ) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda _d: SimpleSpanProcessor(shared), + operator_sinks=MappingProxyType({TestRoutingMode.OPERATOR_SINK: operator_scope}), + ) + ) + return provider + + @staticmethod + def _same_account_destination(span_scope): + return OtelDestination( + endpoint=TestRoutingMode.SAME_ACCOUNT_ENDPOINT, + headers=MappingProxyType({"Authorization": "Basic op"}), + callback_name="langfuse_otel", + span_scope=span_scope, + ) + + @pytest.mark.parametrize( + ("operator_scope", "tenant_scope", "expected"), + [ + ("llm_only", "full", REQUEST_TREE), + ("full", "llm_only", REQUEST_TREE), + ("llm_only", "llm_only", LLM_SPANS), + ("full", "full", REQUEST_TREE), + ], + ) + def test_a_team_naming_the_operators_project_gets_the_wider_of_the_two_scopes_once( + self, monkeypatch, operator_scope, tenant_scope, expected + ): + self._additive(monkeypatch) + shared = InMemorySpanExporter() + + self._run(self._same_account_provider(shared, operator_scope), (self._same_account_destination(tenant_scope),)) + + finished = [s.name for s in shared.get_finished_spans()] + assert frozenset(finished) == expected + assert len(finished) == len(expected), "the same account received a span twice" + + def test_a_full_team_on_the_operators_llm_only_project_gets_one_whole_tree(self, monkeypatch): + """The operator's exporter writes the model call, the fan-out the rest, and Langfuse + upserts by span id: a re-rooted, self-named generation there would replace the one + parented under the request span and rename the whole trace after itself.""" + self._additive(monkeypatch) + shared = InMemorySpanExporter() + + self._run(self._same_account_provider(shared, "llm_only"), (self._same_account_destination("full"),)) + + whole = {s.name: s for s in shared.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + + def test_an_llm_only_team_on_the_operators_llm_only_project_gets_re_rooted_generations(self, monkeypatch): + self._additive(monkeypatch) + shared = InMemorySpanExporter() + + self._run(self._same_account_provider(shared, "llm_only"), (self._same_account_destination("llm_only"),)) + + kept = {s.name: s for s in shared.get_finished_spans()}["chat claude-haiku"] + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_full_team_on_another_account_does_not_widen_the_operators_llm_only_exporter(self, monkeypatch): + self._additive(monkeypatch) + operator = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor( + _OverriddenBackendFilter( + SimpleSpanProcessor(operator), "langfuse_otel", "llm_only", TestRoutingMode.OPERATOR_SINK + ) + ) + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=lambda _d: None)) + + self._run(provider, (LANGFUSE_DEST,)) + + kept = {s.name: s for s in operator.get_finished_spans()}["chat claude-haiku"] + assert names(operator) == LLM_SPANS + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_built_provider_knows_which_account_its_llm_only_exporter_writes_to(self, monkeypatch): + self._additive(monkeypatch) + shared = InMemorySpanExporter() + monkeypatch.setattr(otel_providers, "_exporter_from_spec", lambda _spec: shared) + config = OpenTelemetryV2Config( + langfuse_span_scope="llm_only", + exporters=[ + ExporterSpec( + kind="otlp_http", + endpoint=TestRoutingMode.OPERATOR_SINK[0], + headers="authorization=Basic op", + owner=ExporterOwner.LANGFUSE_OTEL, + ) + ], + ) + provider = build_tracer_provider(config, use_simple_processor=True) + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda _d: SimpleSpanProcessor(shared), + operator_sinks=operator_sink_scopes(config), + ) + ) + + self._run(provider, (self._same_account_destination("full"),)) + + whole = {s.name: s for s in shared.get_finished_spans()} + assert frozenset(whole) == REQUEST_TREE + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + + def test_a_kept_generation_becomes_the_root_of_the_request_trace_with_its_trace_controls(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant), (LLM_ONLY_DEST,)) + + full = {s.name: s for s in operator.get_finished_spans()} + kept = {s.name: s for s in tenant.get_finished_spans()}["chat gpt-4"] + assert kept.context == full["chat gpt-4"].context, "same trace id and span id as the operator's copy" + assert kept.parent is None, "its parent is the request span the tenant never receives" + assert {k: kept.attributes[k] for k in TRACE_CONTROLS} == dict(TRACE_CONTROLS), "the caller's trace name wins" + assert full["chat gpt-4"].parent == full["POST /v1/chat/completions"].context, ( + "the operator's copy is untouched" + ) + + def test_a_kept_generation_with_no_trace_name_is_named_after_itself(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LLM_ONLY_DEST,)) + + for exporter in (operator, tenant): + kept = {s.name: s for s in exporter.get_finished_spans()}["chat claude-haiku"] + assert kept.parent is None + assert kept.attributes["langfuse.trace.name"] == "chat claude-haiku" + assert kept.attributes["gen_ai.request.model"] == "claude-haiku", "the rest of the attributes stay" + + def test_narrowing_one_exporter_leaves_the_other_exporters_view_of_the_span_alone(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant, scope="llm_only"), (LANGFUSE_DEST,)) + + whole = {s.name: s for s in tenant.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + narrowed = {s.name: s for s in operator.get_finished_spans()}["chat claude-haiku"] + assert narrowed.parent is None + assert narrowed.attributes["langfuse.trace.name"] == "chat claude-haiku" + + def test_a_full_scope_exporter_gets_the_generation_under_its_request_span_and_unnamed(self, monkeypatch): + self._additive(monkeypatch) + operator, tenant = InMemorySpanExporter(), InMemorySpanExporter() + + self._run(self._operator_provider(operator, tenant), (LANGFUSE_DEST,)) + + for exporter in (operator, tenant): + whole = {s.name: s for s in exporter.get_finished_spans()} + assert whole["chat claude-haiku"].parent == whole["POST /v1/chat/completions"].context + assert "langfuse.trace.name" not in whole["chat claude-haiku"].attributes + + def test_a_non_langfuse_destination_of_the_same_request_keeps_the_full_tree(self, monkeypatch): + self._additive(monkeypatch) + by_backend = {"langfuse_otel": InMemorySpanExporter(), "arize": InMemorySpanExporter()} + provider = TracerProvider() + provider.add_span_processor( + TenantFanOutSpanProcessor( + processor_factory=lambda d: SimpleSpanProcessor(by_backend[d.callback_name]), + ) + ) + arize = OtelDestination(endpoint="https://otlp.arize.com", headers={"api_key": "k"}, callback_name="arize") + + self._run(provider, (LLM_ONLY_DEST, arize)) + + assert names(by_backend["langfuse_otel"]) == LLM_SPANS + assert names(by_backend["arize"]) == REQUEST_TREE + + def test_two_views_of_one_account_share_the_exporter_but_not_the_filter(self): + built, tenant = [], InMemorySpanExporter() + provider = TracerProvider() + + def factory(destination): + built.append(destination) + return SimpleSpanProcessor(tenant) + + provider.add_span_processor(TenantFanOutSpanProcessor(processor_factory=factory)) + + self._run(provider, (LLM_ONLY_DEST,)) + assert names(tenant) == LLM_SPANS + tenant.clear() + + self._run(provider, (LANGFUSE_DEST,)) + assert names(tenant) == REQUEST_TREE + assert len(built) == 1, "the same account must not get a second exporter for a second scope" + + def test_the_config_scope_reaches_only_the_exporter_langfuse_owns(self, monkeypatch): + exporters = {} + + def exporter_for(spec): + return exporters.setdefault(spec.owner, InMemorySpanExporter()) + + monkeypatch.setattr(otel_providers, "_exporter_from_spec", exporter_for) + config = OpenTelemetryV2Config( + langfuse_span_scope="llm_only", + exporters=[ + ExporterSpec(kind="in_memory", owner=ExporterOwner.LANGFUSE_OTEL), + ExporterSpec(kind="in_memory", owner=ExporterOwner.ARIZE_AX), + ExporterSpec(kind="in_memory"), + ], + ) + + self._run(build_tracer_provider(config, use_simple_processor=True), ()) + + assert names(exporters[ExporterOwner.LANGFUSE_OTEL]) == LLM_SPANS + assert names(exporters[ExporterOwner.ARIZE_AX]) == REQUEST_TREE + assert names(exporters[None]) == REQUEST_TREE, "a bare collector must never be narrowed" + + @pytest.mark.parametrize("tenant_overrides", [False, True]) + def test_the_config_default_leaves_every_exporter_on_the_full_tree(self, monkeypatch, tenant_overrides): + exporters = {} + monkeypatch.setattr( + otel_providers, + "_exporter_from_spec", + lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()), + ) + config = OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory", owner=ExporterOwner.LANGFUSE_OTEL)]) + + self._run(build_tracer_provider(config, use_simple_processor=True, tenant_overrides=tenant_overrides), ()) + + assert names(exporters[ExporterOwner.LANGFUSE_OTEL]) == REQUEST_TREE + + def test_the_env_var_sets_the_operator_scope(self, monkeypatch): + monkeypatch.setenv("LITELLM_OTEL_LANGFUSE_SPAN_SCOPE", "llm_only") + + assert OpenTelemetryV2Config().langfuse_span_scope == "llm_only" + + def test_the_env_var_narrows_the_exporter_the_langfuse_preset_builds(self, monkeypatch): + monkeypatch.setenv("LITELLM_OTEL_LANGFUSE_SPAN_SCOPE", "llm_only") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk") + exporters = {} + monkeypatch.setattr( + otel_providers, + "_exporter_from_spec", + lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()), + ) + config = langfuse_preset(config_overrides=OpenTelemetryV2Config(exporters=[ExporterSpec(kind="in_memory")])) + + self._run(build_tracer_provider(config, use_simple_processor=True), ()) + + assert names(exporters[ExporterOwner.LANGFUSE_OTEL]) == LLM_SPANS + assert names(exporters[None]) == REQUEST_TREE + + def test_an_unknown_scope_is_rejected_by_the_config(self): + with pytest.raises(ValueError, match="langfuse_span_scope"): + OpenTelemetryV2Config(langfuse_span_scope="everything") + + @pytest.mark.parametrize("spelling", ["LLM_ONLY", "Llm_Only", " llm_only\n"]) + def test_the_env_var_is_read_case_and_whitespace_insensitively(self, monkeypatch, spelling): + """A misspelt env var would otherwise fail validation inside the logger builder, + which swallows the error and leaves the proxy up with OTel v2 silently off.""" + monkeypatch.setenv("LITELLM_OTEL_LANGFUSE_SPAN_SCOPE", spelling) + + assert OpenTelemetryV2Config().langfuse_span_scope == "llm_only" + + def test_the_operator_scope_does_not_reach_a_tenants_routed_provider(self, monkeypatch): + """The routed clone carries the tenant's credentials on the operator's Langfuse + exporter. The operator's ``llm_only`` is a choice about the operator's account, + so the clone must export the full tree, as the field's contract promises.""" + tenant = InMemorySpanExporter() + monkeypatch.setattr(otel_providers, "_exporter_from_spec", lambda _spec: tenant) + config = OpenTelemetryV2Config( + langfuse_span_scope="llm_only", + exporters=[ExporterSpec(kind="otlp_http", endpoint="http://op.local", owner=ExporterOwner.LANGFUSE_OTEL)], + ) + cache = TenantTracerCache(config, "langfuse_otel", "litellm") + route = cache.route_for( + get_tracer(TracerProvider(), "litellm"), {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"} + ) + assert route.provider is not None + + request_tree(route.provider) + route.provider.force_flush() + + assert names(tenant) == REQUEST_TREE + + def test_a_team_callback_var_becomes_the_destinations_scope(self, monkeypatch, allow_test_hosts): + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + auth = UserAPIKeyAuth( + team_metadata={ + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": { + "langfuse_public_key": "pk-team", + "langfuse_secret_key": "sk-team", + "langfuse_host": "http://team.local", + "langfuse_span_scope": "llm_only", + }, + } + ] + } + ) + + assert [d.span_scope for d in resolve_tenant_otel_destinations(auth)] == ["llm_only"] + + def test_a_team_that_named_no_scope_gets_the_full_tree(self, allow_test_hosts): + creds = {"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_host": "http://x"} + + assert destination_for("langfuse_otel", creds).span_scope == "full" + + def test_only_langfuse_honours_the_scope_var(self): + arize = destination_for( + "arize", {"arize_api_key": "k", "arize_space_id": "s", "langfuse_span_scope": "llm_only"} + ) + + assert arize is not None and arize.span_scope == "full" + + @pytest.mark.parametrize("scope", ["everything", "LLM_ONLY", ""]) + def test_an_unknown_scope_is_rejected_when_the_callback_is_saved(self, scope): + with pytest.raises(ValueError, match=r"Invalid langfuse_span_scope .*must be one of \['full', 'llm_only'\]"): + AddTeamCallback( + callback_name="langfuse_otel", + callback_type="success", + callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": scope}, + ) + + def test_a_known_scope_is_accepted_when_the_callback_is_saved(self): + saved = AddTeamCallback( + callback_name="langfuse_otel", + callback_type="success", + callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "llm_only"}, + ) + + assert saved.callback_vars["langfuse_span_scope"] == "llm_only" + + #: Anything that makes ``OpenTelemetryV2Config`` synthesize a real operator destination. _OTEL_SHORTHAND_ENV = ( "OTEL_ENDPOINT", @@ -2066,7 +2564,9 @@ class TestEvictionSafety: assert len(built) == _MAX_CACHED_DESTINATION_PROCESSORS + 3, "a processor per request during the outage" assert sum(1 for accepted in anchored if accepted) == len(built), "anchored what it could not build" - assert fan_out.deliverable((self._dest(999),)) == (), "the span would vanish instead of staying with the operator" + assert fan_out.deliverable((self._dest(999),)) == (), ( + "the span would vanish instead of staying with the operator" + ) finally: release.set() for _ in range(500): diff --git a/tests/test_litellm/proxy/common_utils/test_callback_config_validation.py b/tests/test_litellm/proxy/common_utils/test_callback_config_validation.py index 5a06bb92059..a707c92dc15 100644 --- a/tests/test_litellm/proxy/common_utils/test_callback_config_validation.py +++ b/tests/test_litellm/proxy/common_utils/test_callback_config_validation.py @@ -1,5 +1,9 @@ +import pytest + from litellm.proxy.common_utils.callback_config_validation import ( callback_config_error, + conflicting_span_scope_error, + logging_metadata_config_error, ) @@ -10,3 +14,71 @@ def test_callback_config_error_rejects_invalid_langfuse_environment(): assert callback_config_error("langfuse", {"langfuse_environment": "team-a-prod"}) is None assert callback_config_error("langfuse", {"langfuse_public_key": "pk"}) is None + + +def test_callback_config_error_rejects_an_unknown_langfuse_span_scope(): + for bad in ["everything", "LLM_ONLY", "llm-only", ""]: + error = callback_config_error("langfuse_otel", {"langfuse_span_scope": bad}) + assert error is not None and "langfuse_span_scope" in error and "llm_only" in error + + assert callback_config_error("langfuse_otel", {"langfuse_span_scope": "llm_only"}) is None + assert callback_config_error("langfuse_otel", {"langfuse_span_scope": "full"}) is None + + +def test_a_span_scope_on_a_callback_that_does_not_read_it_is_rejected(): + """Only langfuse_otel filters on the scope. Accepting it on the classic Langfuse + callback or on an unrelated backend would store a setting that never takes + effect, with the full tree still exported.""" + for callback_name in ["langfuse", "datadog", "otel", None]: + error = callback_config_error(callback_name, {"langfuse_span_scope": "llm_only"}) + assert error is not None and "langfuse_span_scope" in error and "langfuse_otel" in error + + assert callback_config_error("langfuse", {"langfuse_environment": "team-a-prod"}) is None + + +def test_a_bad_span_scope_is_reported_even_when_the_environment_is_fine(): + error = callback_config_error( + "langfuse_otel", {"langfuse_environment": "team-a-prod", "langfuse_span_scope": "everything"} + ) + assert error is not None and "langfuse_span_scope" in error + + +@pytest.mark.parametrize( + "new_vars, stored, rejected", + [ + ({"langfuse_span_scope": "llm_only"}, [{"langfuse_span_scope": "full"}], True), + ({"langfuse_span_scope": "full"}, [{"langfuse_public_key": "pk"}, {"langfuse_span_scope": "llm_only"}], True), + ({"langfuse_span_scope": "llm_only"}, [{"langfuse_span_scope": "llm_only"}], False), + ({"langfuse_span_scope": "llm_only"}, [{"langfuse_public_key": "pk"}], False), + ({"langfuse_span_scope": "llm_only"}, [], False), + ({"langfuse_public_key": "pk"}, [{"langfuse_span_scope": "llm_only"}], False), + (None, [{"langfuse_span_scope": "llm_only"}], False), + ], +) +def test_one_span_scope_per_team(new_vars, stored, rejected): + """The entries flatten last-wins, so a second scope would export whichever entry + was stored last. An entry that names no scope leaves the stored one in charge.""" + error = conflicting_span_scope_error(new_vars, stored) + assert (error is not None) is rejected + if rejected: + assert "langfuse_span_scope" in error and stored[-1]["langfuse_span_scope"] in error + + +def test_key_logging_entries_may_not_disagree_on_the_span_scope(): + disagreeing = { + "logging": [ + {"callback_name": "langfuse_otel", "callback_type": "success", "callback_vars": {"langfuse_span_scope": "full"}}, + {"callback_name": "langfuse_otel", "callback_type": "failure", "callback_vars": {"langfuse_span_scope": "llm_only"}}, + ] + } + error = logging_metadata_config_error(disagreeing) + assert error is not None and "langfuse_span_scope" in error and "'full'" in error + + agreeing = { + "logging": [ + {"callback_name": "langfuse_otel", "callback_type": "success", "callback_vars": {"langfuse_span_scope": "llm_only"}}, + {"callback_name": "langfuse_otel", "callback_type": "failure", "callback_vars": {"langfuse_span_scope": "llm_only"}}, + {"callback_name": "otel", "callback_type": "success", "callback_vars": {}}, + ] + } + assert logging_metadata_config_error(agreeing) is None diff --git a/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py index 272a8ffa972..5c5cfd0814d 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py @@ -285,6 +285,20 @@ class TestNewRelicCallbackConfig: assert "NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED" not in params +class TestLangfuseOtelCallbackConfig: + def test_span_scope_is_a_select_over_exactly_the_scopes_the_validator_accepts(self): + from litellm.types.utils import OTEL_SPAN_SCOPES + + client = TestClient(app) + response = client.get("/callbacks/configs", headers={"Authorization": "Bearer sk-1234"}) + assert response.status_code == 200 + langfuse_otel = next(config for config in response.json() if config.get("id") == "langfuse_otel") + scope = langfuse_otel["dynamic_params"]["langfuse_span_scope"] + assert scope["type"] == "select" + assert frozenset(scope["options"]) == OTEL_SPAN_SCOPES + assert scope["required"] is False + + class TestNewRelicTeamCallbackValidation: def _data(self, callback_vars): from litellm.proxy._types import AddTeamCallback diff --git a/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py index bdc12dad4bc..acc7c8ca21a 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py @@ -1538,6 +1538,12 @@ async def test_proxy_admin_still_told_the_team_is_unknown(): ({"langsmith_api_key": "k"}, [{"dd_api_key": "k"}], False), # variables that configure no backend carry nothing to redirect ({"turn_off_message_logging": "true"}, [{"langfuse_secret_key": "sk"}], False), + # the span scope picks what the family exports, not where to, so a second + # entry may set either legal value next to the family's credentials + ({"langfuse_span_scope": "llm_only"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False), + ({"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "full"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "llm_only"}], False), + # the scope on the stored entry must not shield a redirect riding next to it + ({"langfuse_host": "http://attacker.invalid", "langfuse_span_scope": "llm_only"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "llm_only"}], True), # the same integration registered for a second event: identical values # flatten to the identical dict, so there is nothing to redirect ({"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False), @@ -1559,3 +1565,48 @@ def test_one_entry_owns_a_credential_family(new_vars, stored, rejected): """ error = cross_entry_family_error(new_vars, stored) assert (error is not None) is rejected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("caller", [_admin_auth(), UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="victim_admin", api_key="sk-team-admin")]) +async def test_a_second_entry_may_not_flip_the_span_scope(patched_prisma, caller): + """The entries flatten last-wins at request time, so a failure entry saying + llm_only next to a success entry saying full would export whichever is stored + last. Neither a proxy admin nor a team admin gets to store the disagreement.""" + patched_prisma.get_data = AsyncMock( + return_value=_team_row( + metadata={ + "logging": [ + { + "callback_name": "langfuse_otel", + "callback_type": "success", + "callback_vars": {"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "full"}, + } + ] + } + ) + ) + data = AddTeamCallback( + callback_name="langfuse_otel", + callback_type="failure", + callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_span_scope": "llm_only"}, + ) + with pytest.raises(HTTPException) as exc: + await add_team_callbacks( + data=data, + http_request=Mock(spec=Request), + team_id="team-victim", + user_api_key_dict=caller, + ) + assert exc.value.status_code == 400 + assert "langfuse_span_scope" in str(exc.value.detail) and "'full'" in str(exc.value.detail) + patched_prisma.db.litellm_teamtable.update.assert_not_called() + + data.callback_vars["langfuse_span_scope"] = "full" + await add_team_callbacks( + data=data, + http_request=Mock(spec=Request), + team_id="team-victim", + user_api_key_dict=caller, + ) + patched_prisma.db.litellm_teamtable.update.assert_awaited_once() diff --git a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx index 8e343e1a6e0..b43a85d18c5 100644 --- a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx +++ b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx @@ -17,6 +17,7 @@ interface CallbackConfig { logo?: string; supports_key_team_logging: boolean; dynamic_params: Record; + dynamic_param_options?: Record; description: string; } @@ -124,6 +125,10 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [ langfuse_secret_key: "password", langfuse_host: "text", langfuse_environment: "text", + langfuse_span_scope: "select", + }, + dynamic_param_options: { + langfuse_span_scope: ["full", "llm_only"], }, description: "Langfuse v3 OTEL Logging Integration", }, diff --git a/ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx b/ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx index b63a6cb98aa..f7ca6d516f5 100644 --- a/ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx +++ b/ui/litellm-dashboard/src/components/team/LoggingSettings.test.tsx @@ -216,6 +216,29 @@ describe("LoggingSettings", () => { expect(mockOnChange).toHaveBeenCalledWith([expect.objectContaining({ callback_type: "failure" })]); }); + it("offers the Langfuse OTEL span scope as a pick between full and llm_only rather than free text", async () => { + const user = userEvent.setup({ delay: null }); + const mockOnChange = vi.fn(); + const initialValue = [ + { + callback_name: "langfuse_otel", + callback_type: "success", + callback_vars: {}, + }, + ]; + + renderWithProviders(); + + expect(screen.queryByPlaceholderText("os.environ/LANGFUSE_SPAN_SCOPE")).not.toBeInTheDocument(); + await user.click(screen.getByRole("combobox", { name: "langfuse span scope" })); + expect((await screen.findAllByRole("option")).map((option) => option.textContent)).toEqual(["full", "llm_only"]); + await user.click(screen.getByRole("option", { name: "llm_only" })); + + expect(mockOnChange).toHaveBeenCalledWith([ + expect.objectContaining({ callback_vars: expect.objectContaining({ langfuse_span_scope: "llm_only" }) }), + ]); + }); + it("correctly handles numerical input with decimal values", () => { const mockOnChange = vi.fn(); diff --git a/ui/litellm-dashboard/src/components/team/LoggingSettings.tsx b/ui/litellm-dashboard/src/components/team/LoggingSettings.tsx index d526b710bcf..2f66e68eca4 100644 --- a/ui/litellm-dashboard/src/components/team/LoggingSettings.tsx +++ b/ui/litellm-dashboard/src/components/team/LoggingSettings.tsx @@ -135,6 +135,57 @@ const LoggingSettings: React.FC = ({ handleChange(updatedConfigs); }; + const renderParamControl = ( + config: LoggingConfig, + configIndex: number, + paramName: string, + param: { type: string; options: readonly string[] }, + ) => { + const { type: paramType, options } = param; + const label = paramName.replace(/_/g, " "); + if (options.length > 0) { + return ( + + ); + } + if (paramType === "number") { + return ( + ) => + updateCallbackVar(configIndex, paramName, e.target.value) + } + /> + ); + } + return ( + updateCallbackVar(configIndex, paramName, newValue)} + /> + ); + }; + const renderDynamicParams = (config: LoggingConfig, configIndex: number) => { if (!config.callback_name) return null; @@ -144,6 +195,7 @@ const LoggingSettings: React.FC = ({ if (!callbackDisplayName) return null; const dynamicParams = callbackInfo[callbackDisplayName]?.dynamic_params || {}; + const paramOptions = callbackInfo[callbackDisplayName]?.dynamic_param_options || {}; if (Object.keys(dynamicParams).length === 0) return null; @@ -166,22 +218,10 @@ const LoggingSettings: React.FC = ({ {paramType === "number" && ( Value must be between 0 and 1 )} - {paramType === "number" ? ( - updateCallbackVar(configIndex, paramName, e.target.value)} - /> - ) : ( - updateCallbackVar(configIndex, paramName, newValue)} - /> - )} + {renderParamControl(config, configIndex, paramName, { + type: paramType, + options: paramType === "select" ? paramOptions[paramName] || [] : [], + })} ))} diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 75f83c44a6b..57ff2fe9d87 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -16215,6 +16215,7 @@ export interface paths { * - langfuse_secret: The secret for the Langfuse callback * - langfuse_host: The host for the Langfuse callback * - langfuse_environment: The tracing environment for the Langfuse callback (lowercase; falls back to LANGFUSE_TRACING_ENVIRONMENT) + * - langfuse_span_scope: For langfuse_otel, "full" (default) sends the whole request trace, "llm_only" sends only the model-call spans * - gcs_bucket_name: The name of the GCS bucket * - gcs_path_service_account: The path to the GCS service account * - langsmith_api_key: The API key for the Langsmith callback