From 74392efbd4292dd79e9cc769c8a56e44d4569346 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Wed, 28 Oct 2026 10:48:34 -0700 Subject: [PATCH] fix(otel): label configured pass-through routes litellm.passthrough and read the operation from the upstream (#48675) --- .../opentelemetry_utils/gen_ai_semconv.py | 6 +- litellm/integrations/otel/logger.py | 1 + litellm/integrations/otel/model/metadata.py | 6 +- litellm/integrations/otel/model/payloads.py | 7 +- litellm/integrations/otel/model/semconv.py | 10 +- litellm/integrations/otel/plumbing/metrics.py | 5 +- .../pass_through_endpoints.py | 12 ++ .../upstream_usage_headers.py | 23 ++- .../test_passthrough_operation_name.py | 163 ++++++++++++++++++ .../integrations/otel/test_otel_v2_metrics.py | 16 ++ .../otel/test_otel_v2_sources_of_truth.py | 11 ++ tests/unit/integrations/test_opentelemetry.py | 10 ++ .../test_upstream_usage_headers.py | 23 +++ 13 files changed, 284 insertions(+), 9 deletions(-) create mode 100644 tests/integration/observability/test_passthrough_operation_name.py diff --git a/litellm/integrations/opentelemetry_utils/gen_ai_semconv.py b/litellm/integrations/opentelemetry_utils/gen_ai_semconv.py index 035e5c5db10..84c4b2e58ab 100644 --- a/litellm/integrations/opentelemetry_utils/gen_ai_semconv.py +++ b/litellm/integrations/opentelemetry_utils/gen_ai_semconv.py @@ -35,6 +35,7 @@ from enum import Enum from typing import TYPE_CHECKING, Final, Protocol from litellm.integrations.otel.mappers.utils import drop_none +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation from litellm.litellm_core_utils.safe_json_dumps import safe_dumps if TYPE_CHECKING: @@ -148,8 +149,11 @@ class OTELGenAISemconvMixin: """Map a LiteLLM ``call_type`` to spec ``gen_ai.operation.name``. Substring match (e.g. ``aembedding`` -> ``embeddings``); defaults to - ``chat``. + ``chat``. A pass-through call's operation, stored under ``PASSTHROUGH_OPERATION_KEY``, wins. """ + passthrough_operation: Final = kwargs.get(PASSTHROUGH_OPERATION_KEY) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # kwargs is an untyped dict + if isinstance(passthrough_operation, GenAIOperation): + return passthrough_operation.value call_type: Final = kwargs.get("call_type", "") or "" match call_type: case s if "embedding" in s: diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index e09226ac9cd..4c5e65ff07d 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -600,6 +600,7 @@ class OpenTelemetryV2(CustomLogger): trace=call.trace, session_id=call.session_id, metadata_keys=tuple(self.config.baggage_metadata_keys), + operation=call.operation, ) end_time_ns: Final = to_ns(end_time) if carrier is not None and carrier.span is not None: diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index b96ed55872d..a9a57de04ac 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -49,7 +49,7 @@ from litellm.constants import ( LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, SESSION_ID_GENERATED_METADATA_KEY, ) -from litellm.integrations.otel.model.semconv import resolve_operation +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation, resolve_operation from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds from litellm.types.utils import InternalCallOrigin @@ -262,6 +262,7 @@ class LLMCallEvent: # never hands off to a provider, so ``upstream_started`` stays False even # though the call completed and is worth a span. cache_hit: bool + operation: GenAIOperation # A best-effort ``"{operation} {model}"`` name known at ``pre_call`` time. The # span is renamed from the typed payload at close (``finish_span``); this only # needs to be reasonable for a span that never gets closed (a leak). @@ -275,7 +276,7 @@ class LLMCallEvent: def from_dict(cls, kwargs: Mapping[str, object]) -> LLMCallEvent: raw_payload: Final = kwargs.get("standard_logging_object") payload: Final = cast("StandardLoggingPayload", raw_payload) if raw_payload else None - operation: Final = resolve_operation(as_str(kwargs.get("call_type"))) + operation: Final = resolve_operation(as_str(kwargs.get("call_type")), kwargs.get(PASSTHROUGH_OPERATION_KEY)) model: Final = as_str(kwargs.get("model")) or "" trace: Final = caller_trace_controls(kwargs) return cls( @@ -288,6 +289,7 @@ class LLMCallEvent: upstream_start_seconds=_epoch_seconds(kwargs.get("api_call_start_time")), purpose=internal_call_origin(payload, kwargs), cache_hit=bool(payload.get("cache_hit")) if payload else False, # pyright: ignore[reportUnknownMemberType] # TypedDict .get on a str key is partially unknown + operation=operation, provisional_span_name=f"{operation.value} {model}".strip(), time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs), completion_start_seconds=_epoch_seconds(kwargs.get("completion_start_time")), diff --git a/litellm/integrations/otel/model/payloads.py b/litellm/integrations/otel/model/payloads.py index 21b2f44f077..7a62ed6aed9 100644 --- a/litellm/integrations/otel/model/payloads.py +++ b/litellm/integrations/otel/model/payloads.py @@ -464,6 +464,7 @@ class LLMCallSpanData: session_id: str | None = None, *, metadata_keys: tuple[str, ...] = (), + operation: GenAIOperation | None = None, ) -> LLMCallSpanData: params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {}) cache_hit: Final = as_bool(payload.get("cache_hit")) is True # pyright: ignore[reportUnknownMemberType] # TypedDict .get on a str key is partially unknown @@ -486,13 +487,13 @@ class LLMCallSpanData: # no prompt/response text. finish_reasons: Final = _finish_reasons(choices_out) call_type: Final = as_str(payload.get("call_type")) - operation: Final = resolve_operation(call_type) + resolved_operation: Final = operation or resolve_operation(call_type) embedding_output: Final = ( - EmbeddingOutput.from_response(response) if operation is GenAIOperation.EMBEDDINGS else None + EmbeddingOutput.from_response(response) if resolved_operation is GenAIOperation.EMBEDDINGS else None ) cost_lookup_failed: Final = payload.get("response_cost_failure_debug_info") is not None and not cache_hit return cls( - operation=operation, + operation=resolved_operation, provider=resolve_provider(as_str(payload.get("custom_llm_provider"))), request_model=context.request_model, response_model=context.response_model, diff --git a/litellm/integrations/otel/model/semconv.py b/litellm/integrations/otel/model/semconv.py index b7f955a75ed..8d186a2715a 100644 --- a/litellm/integrations/otel/model/semconv.py +++ b/litellm/integrations/otel/model/semconv.py @@ -34,6 +34,7 @@ class GenAIOperation(str, Enum): LITELLM_VECTOR_STORE_FILE_MANAGEMENT = "litellm.vector_store_file_management" LITELLM_RESPONSES_MANAGEMENT = "litellm.responses_management" LITELLM_MODERATION = "litellm.moderation" + LITELLM_PASSTHROUGH = "litellm.passthrough" class GenAIOutputType(str, Enum): @@ -482,14 +483,21 @@ def resolve_provider(custom_llm_provider: str | None) -> str: return mapped.value if mapped is not None else custom_llm_provider -def resolve_operation(call_type: str | None) -> GenAIOperation: +PASSTHROUGH_OPERATION_KEY: Final = "passthrough_gen_ai_operation" + + +def resolve_operation(call_type: str | None, passthrough_operation: object = None) -> GenAIOperation: """Map a litellm ``call_type`` to a ``gen_ai.operation.name`` value. An unmapped call type still falls back to ``chat`` so every series keeps an operation label, but it logs at debug rather than falling through silently: a new call type mislabelled as ``chat`` mixes its latency and cost into everyone's chat charts, which is invisible until someone reads the numbers. + + A pass-through call's operation, stored under ``PASSTHROUGH_OPERATION_KEY``, wins over its call type. """ + if isinstance(passthrough_operation, GenAIOperation): + return passthrough_operation if not call_type: return GenAIOperation.CHAT mapped: Final = _OPERATION_BY_CALL_TYPE.get(call_type.lower()) diff --git a/litellm/integrations/otel/plumbing/metrics.py b/litellm/integrations/otel/plumbing/metrics.py index 72d51471ba2..1197ae055d0 100644 --- a/litellm/integrations/otel/plumbing/metrics.py +++ b/litellm/integrations/otel/plumbing/metrics.py @@ -26,6 +26,7 @@ from litellm.integrations.opentelemetry import ( ) from litellm.integrations.otel.model.metadata import time_to_first_chunk_seconds from litellm.integrations.otel.model.semconv import ( + PASSTHROUGH_OPERATION_KEY, Error, GenAI, Metric, @@ -280,7 +281,9 @@ class GenAIMetricRecorder: def _common_attributes(self, kwargs: Mapping[str, Any]) -> dict: params: Final = kwargs.get("litellm_params") or {} common_attrs: Final[dict] = { - GenAI.OPERATION_NAME: resolve_operation(kwargs.get("call_type")).value, + GenAI.OPERATION_NAME: resolve_operation( + kwargs.get("call_type"), kwargs.get(PASSTHROUGH_OPERATION_KEY) + ).value, **_provider_attributes(params.get("custom_llm_provider")), GenAI.REQUEST_MODEL: kwargs.get("model"), "gen_ai.framework": "litellm", diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index f2ea1752a86..650dcae3239 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -54,6 +54,7 @@ from litellm.integrations.custom_guardrail import ( without_server_streaming_classification, ) from litellm.integrations.custom_logger import CustomLogger +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation from litellm.litellm_core_utils.core_helpers import ( bind_budget_reservation_to_callbacks, get_metadata_variable_name_from_kwargs, @@ -151,6 +152,7 @@ from .success_handler import PassThroughEndpointLogging from .upstream_usage_headers import ( UpstreamReportedUsage, apply_upstream_reported_model, + apply_upstream_reported_operation, apply_upstream_reported_usage, ) @@ -1134,6 +1136,7 @@ async def pass_through_request( guardrails_config: dict | None = None, timeout: float | None = None, upstream_headers_filter: Callable[[Mapping[str, str]], Mapping[str, str]] = _unfiltered_headers, + operation: GenAIOperation | None = None, ): """ Pass through endpoint handler, makes the httpx request for pass-through endpoints and ensures logging hooks are called @@ -1155,6 +1158,7 @@ async def pass_through_request( timeout: Optional per-endpoint timeout in seconds. Falls back to general_settings.pass_through_request_timeout, then 600s. upstream_headers_filter: Optional last rewrite of the headers sent upstream, after x-pass- expansion + operation: Optional GenAI operation the route reports, unless the upstream names its own """ from litellm.exceptions import ModifyResponseException from litellm.litellm_core_utils.litellm_logging import Logging @@ -1353,6 +1357,8 @@ async def pass_through_request( call_type="pass_through_endpoint", ) logging_obj.model_call_details["litellm_call_id"] = litellm_call_id + if operation is not None: + logging_obj.model_call_details[PASSTHROUGH_OPERATION_KEY] = operation ## PASSTHROUGH MANAGED ID RESOLUTION (INPUT) ## # Resolve managed IDs in path, query params, and body back to raw @@ -1553,6 +1559,7 @@ async def pass_through_request( logging_obj=logging_obj, headers=response.headers, ) + apply_upstream_reported_operation(logging_obj, response.headers) relay_response: Final = await _log_passthrough_upstream_failure( response=response, @@ -1646,6 +1653,7 @@ async def pass_through_request( logging_obj=logging_obj, headers=response.headers, ) + apply_upstream_reported_operation(logging_obj, response.headers) if _is_streaming_response(response) is True: logging_obj.stream = True @@ -2161,6 +2169,7 @@ def create_pass_through_route( config_file_path: str | None = None, timeout: float | None = None, upstream_headers_filter: Callable[[Mapping[str, str]], Mapping[str, str]] = _unfiltered_headers, + operation: GenAIOperation | None = None, ): # check if target is an adapter.py or a url from litellm._uuid import uuid @@ -2301,6 +2310,7 @@ def create_pass_through_route( guardrails_config=cast(dict | None, param_guardrails), timeout=cast(float | None, param_timeout), upstream_headers_filter=upstream_headers_filter, + operation=operation, ) finally: if hasattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY): @@ -3232,6 +3242,7 @@ class InitPassThroughEndpointHelpers: guardrails=guardrails, config_file_path=config_file_path, timeout=timeout, + operation=GenAIOperation.LITELLM_PASSTHROUGH, ), methods=methods, dependencies=dependencies, @@ -3316,6 +3327,7 @@ class InitPassThroughEndpointHelpers: guardrails=guardrails, config_file_path=config_file_path, timeout=timeout, + operation=GenAIOperation.LITELLM_PASSTHROUGH, ), methods=methods, dependencies=dependencies, diff --git a/litellm/proxy/pass_through_endpoints/upstream_usage_headers.py b/litellm/proxy/pass_through_endpoints/upstream_usage_headers.py index e9d93fb306d..223f6a5e760 100644 --- a/litellm/proxy/pass_through_endpoints/upstream_usage_headers.py +++ b/litellm/proxy/pass_through_endpoints/upstream_usage_headers.py @@ -5,18 +5,21 @@ price the request from the response body. Those targets report the totals for the whole HTTP request in ``x-litellm-*`` response headers instead; LiteLLM records the reported values without recomputing them. A target may also name the request in ``x-litellm-upstream-model`` so its spend can be told apart from -other requests that carry no ``model`` in the body. +other requests that carry no ``model`` in the body, and name the GenAI +operation it performed, which labels the request's trace. """ import math from collections.abc import Mapping from dataclasses import dataclass +from types import MappingProxyType from typing import Final import httpx from litellm._logging import verbose_proxy_logger from litellm.constants import MAX_SPEND_LOG_MODEL_NAME_LENGTH +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.types.utils import Usage @@ -25,6 +28,8 @@ UPSTREAM_TOTAL_TOKENS_HEADER: Final = "x-litellm-total-tokens" # Deliberately not x-litellm-model-name/-id/-group: the gateway already sends # those to clients, and pass-through relays upstream headers back to the client. UPSTREAM_MODEL_HEADER: Final = "x-litellm-upstream-model" +UPSTREAM_OPERATION_NAME_HEADER: Final = "x-litellm-operation-name" +_OPERATIONS_BY_NAME: Final = MappingProxyType({operation.value: operation for operation in GenAIOperation}) # model_call_details key holding what the upstream reported, so later stages of # the success path can tell an upstream-reported cost apart from one LiteLLM @@ -175,3 +180,19 @@ def has_upstream_reported_usage(logging_obj: LiteLLMLoggingObj) -> bool: means zero, never a fallback to someone else's estimate. """ return isinstance(logging_obj.model_call_details.get(UPSTREAM_REPORTED_USAGE_KEY), UpstreamReportedUsage) + + +def apply_upstream_reported_operation(logging_obj: LiteLLMLoggingObj, headers: Mapping[str, str]) -> None: + """Record the GenAI operation the upstream named, overriding the route's label. An unknown name is ignored.""" + raw_operation: Final = headers.get(UPSTREAM_OPERATION_NAME_HEADER) + if raw_operation is None: + return + operation: Final = _OPERATIONS_BY_NAME.get(raw_operation) + if operation is None: + verbose_proxy_logger.warning( + "pass_through_endpoint: upstream sent unknown %s=%r; ignoring it", + UPSTREAM_OPERATION_NAME_HEADER, + raw_operation, + ) + return + logging_obj.model_call_details[PASSTHROUGH_OPERATION_KEY] = operation # rebind-ok: annotates the request's record diff --git a/tests/integration/observability/test_passthrough_operation_name.py b/tests/integration/observability/test_passthrough_operation_name.py new file mode 100644 index 00000000000..7b01389a993 --- /dev/null +++ b/tests/integration/observability/test_passthrough_operation_name.py @@ -0,0 +1,163 @@ +import json +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import httpx +import pytest +import yaml +from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.otlp_sink import Span, SpanSinks, recorded_spans +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, wire_server + +pytestmark: Final = pytest.mark.timeout(180) + +_EXPORTERS: Final = { + "otel-v1-semconv": {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"}, + "otel-v2": {"LITELLM_OTEL_V2": "1"}, +} +_ANTHROPIC_MESSAGE: Final = { + "id": "msg_passthrough_operation", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-5-5", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 3, "output_tokens": 1}, +} +_STREAM_REQUESTS: Final = {"stream-requested": {"query": "q", "stream": True}, "stream-detected": {"query": "q"}} + + +@dataclass(frozen=True, slots=True) +class _Rig: + proxy: Gateway + sink: str + + +def _upstream(request: Request) -> Reply: + if request.target.endswith("/v1/messages"): + return Reply(body=json.dumps(_ANTHROPIC_MESSAGE).encode()) + _, _, reported = request.target.partition("/report/") + headers: Final = {"x-litellm-operation-name": reported} if reported else {} + if request.target.startswith("/app/stream/cut"): + return Reply( + content_type="text/event-stream", + chunks=(b'data: {"chunk": 1}\n\n', b'data: {"chunk": 2}\n\n'), + abort_after=1, + ) + if request.target.startswith("/app/stream"): + return Reply(content_type="text/event-stream", chunks=(b'data: {"chunk": 1}\n\n',), headers=headers) + return Reply(body=b'{"chunks": []}', headers=headers) + + +@pytest.fixture(scope="module", params=sorted(_EXPORTERS)) +def rig( + request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory, audit_sinks: SpanSinks +) -> Iterator[_Rig]: + directory: Final = tmp_path_factory.mktemp("passthrough-operation") + exporter: Final = _EXPORTERS[request.param] + with gateway_from_environment() as base, wire_server(_upstream) as upstream: + shared: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config: Final = { + **shared, + "litellm_settings": {**shared["litellm_settings"], "callbacks": ["otel"]}, + "callback_settings": { + "otel": {"exporter": "http/json", "endpoint": audit_sinks.operator, "use_simple_processor": True} + }, + "general_settings": { + **shared["general_settings"], + "pass_through_endpoints": [ + {"path": "/doc-app", "target": upstream.url + "/app", "include_subpath": True} + ], + }, + "environment_variables": { + **shared["environment_variables"], + "ANTHROPIC_API_BASE": upstream.url, + "ANTHROPIC_API_KEY": "scripted", + }, + } + path: Final = directory / "passthrough-operation.yaml" + path.write_text(yaml.safe_dump(config)) + overrides: Final = {"OTEL_EXPORTER": "http/json", "OTEL_ENDPOINT": audit_sinks.operator, **exporter} + unset: Final = tuple( + name for name in ("LITELLM_OTEL_V2", "OTEL_SEMCONV_STABILITY_OPT_IN") if name not in exporter + ) + with owned_proxy(base, directory, overrides, config=path, remove_environment=unset) as proxy: + yield _Rig(proxy, audit_sinks.operator) + + +def _span_of(rig: _Rig, since: int, call_id: str) -> Span: + spans: Final = eventually( + lambda: tuple( + span + for span in recorded_spans(rig.sink, since)[1] + if span["attributes"].get("litellm.call_id") == call_id and "gen_ai.operation.name" in span["attributes"] + ), + lambda found: len(found) == 1, + seconds=60, + return_last_on_timeout=True, + ) + assert len(spans) == 1, spans + return spans[0] + + +def _operation_span(rig: _Rig, method: str, path: str, body: dict[str, object] | None = None) -> Span: + since, _ = recorded_spans(rig.sink) + response: Final = rig.proxy.request(method, path, body) + assert response.status_code == 200, response.text + return _span_of(rig, since, response.headers["x-litellm-call-id"]) + + +def _cut_stream_span(rig: _Rig, path: str) -> Span: + since, _ = recorded_spans(rig.sink) + headers: Final = {"Authorization": f"Bearer {rig.proxy.key}"} + with rig.proxy.client.stream("POST", path, json={"query": "q", "stream": True}, headers=headers) as response: + assert response.status_code == 200 + with pytest.raises(httpx.RemoteProtocolError): + response.read() + return _span_of(rig, since, response.headers["x-litellm-call-id"]) + + +def _assert_operation(span: Span, operation: str) -> None: + assert span["attributes"]["gen_ai.operation.name"] == operation, span + assert span["name"].split(" ")[0] == operation, span + + +def test_configured_route_without_the_header_is_labelled_passthrough(rig: _Rig) -> None: + _assert_operation(_operation_span(rig, "POST", "/doc-app/search", {"query": "q"}), "litellm.passthrough") + + +def test_operation_reported_by_the_target_labels_the_span(rig: _Rig) -> None: + _assert_operation(_operation_span(rig, "POST", "/doc-app/report/retrieval", {"query": "q"}), "retrieval") + + +def test_unknown_operation_reported_by_the_target_is_ignored(rig: _Rig) -> None: + _assert_operation(_operation_span(rig, "POST", "/doc-app/report/summarize", {"query": "q"}), "litellm.passthrough") + + +@pytest.mark.parametrize("body", _STREAM_REQUESTS.values(), ids=_STREAM_REQUESTS.keys()) +def test_streamed_configured_route_without_the_header_is_labelled_passthrough( + rig: _Rig, body: dict[str, object] +) -> None: + _assert_operation(_operation_span(rig, "POST", "/doc-app/stream", body), "litellm.passthrough") + + +@pytest.mark.parametrize("body", _STREAM_REQUESTS.values(), ids=_STREAM_REQUESTS.keys()) +def test_operation_reported_by_a_streamed_target_labels_the_span(rig: _Rig, body: dict[str, object]) -> None: + _assert_operation(_operation_span(rig, "POST", "/doc-app/stream/report/retrieval", body), "retrieval") + + +def test_configured_route_stream_cut_by_the_target_is_labelled_passthrough(rig: _Rig) -> None: + _assert_operation(_cut_stream_span(rig, "/doc-app/stream/cut"), "litellm.passthrough") + + +def test_provider_route_stays_chat(rig: _Rig) -> None: + body: Final = {"model": "claude-sonnet-5-5", "max_tokens": 8, "messages": [{"role": "user", "content": "hi"}]} + _assert_operation(_operation_span(rig, "POST", "/anthropic/v1/messages", body), "chat") + + +def test_provider_route_litellm_does_not_parse_stays_chat(rig: _Rig) -> None: + _assert_operation(_operation_span(rig, "GET", "/anthropic/v1/models"), "chat") diff --git a/tests/unit/integrations/otel/test_otel_v2_metrics.py b/tests/unit/integrations/otel/test_otel_v2_metrics.py index 016dbcd824b..74921512825 100644 --- a/tests/unit/integrations/otel/test_otel_v2_metrics.py +++ b/tests/unit/integrations/otel/test_otel_v2_metrics.py @@ -24,6 +24,7 @@ dashboard already querying that histogram. import asyncio import json from datetime import datetime, timedelta +from typing import Final import pytest @@ -40,6 +41,10 @@ from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402 from litellm.integrations.otel.model.config import ( # noqa: E402 OpenTelemetryV2Config, ) +from litellm.integrations.otel.model.semconv import ( # noqa: E402 + PASSTHROUGH_OPERATION_KEY, + GenAIOperation, +) from litellm.integrations.otel.plumbing.metrics import ( # noqa: E402 GenAIMetricRecorder, create_genai_metrics, @@ -500,6 +505,17 @@ def test_agent_message_is_not_labelled_as_chat(call_type): assert dp.attributes[OPERATION_KEY] == "invoke_agent" +def test_passthrough_operation_labels_the_metrics() -> None: + reader: Final = InMemoryMetricReader() + kwargs, response_obj, start, end = _build_call(call_type="pass_through_endpoint") + labelled: Final = {**kwargs, PASSTHROUGH_OPERATION_KEY: GenAIOperation.RETRIEVAL} + + asyncio.run(_logger(reader, enable_metrics=True).async_log_success_event(labelled, response_obj, start, end)) + + for dp in _metrics_by_name(reader)[OPERATION_DURATION]: + assert dp.attributes[OPERATION_KEY] == "retrieval" + + def test_provider_name_is_filterable(): """gen_ai.provider.name is a member of the metric-attribute allowlist, so an operator can include or exclude it; an unlisted name raises instead.""" diff --git a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py index 318859cb3ed..2d0b4701c9f 100644 --- a/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py +++ b/tests/unit/integrations/otel/test_otel_v2_sources_of_truth.py @@ -38,6 +38,7 @@ from litellm.integrations.otel.model.payloads import ( RequestIdentity, _upstream_address_port, ) +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY from litellm.integrations.otel.model.spans import ( SPAN_REGISTRY, LiteLLMSpanKind, @@ -223,6 +224,16 @@ def test_operation_resolution(): assert resolve_operation("call_mcp_tool") is GenAIOperation.EXECUTE_TOOL +def test_passthrough_operation_wins_over_the_call_type() -> None: + assert resolve_operation("pass_through_endpoint", GenAIOperation.RETRIEVAL) is GenAIOperation.RETRIEVAL + assert resolve_operation("pass_through_endpoint", "retrieval") is GenAIOperation.CHAT + event: Final = LLMCallEvent.from_dict( + {"call_type": "pass_through_endpoint", "model": "unknown", PASSTHROUGH_OPERATION_KEY: GenAIOperation.RETRIEVAL} + ) + assert event.operation is GenAIOperation.RETRIEVAL + assert event.provisional_span_name == "retrieval unknown" + + @pytest.mark.parametrize("call_type", ["vector_store_search", "avector_store_search"]) def test_vector_store_search_is_a_retrieval_operation(call_type): """A vector-store search is a retrieval, so its duration and cost must not diff --git a/tests/unit/integrations/test_opentelemetry.py b/tests/unit/integrations/test_opentelemetry.py index f908c685382..f0fca94c370 100644 --- a/tests/unit/integrations/test_opentelemetry.py +++ b/tests/unit/integrations/test_opentelemetry.py @@ -48,6 +48,7 @@ from litellm.integrations.opentelemetry import ( RAW_REQUEST_SPAN_NAME, _normalize_team_metadata_keys, ) +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.proxy import proxy_server from litellm.proxy._types import ManagementEndpointLoggingPayload, ProxyException, SpanAttributes @@ -1048,6 +1049,15 @@ class TestOpenTelemetrySemconvStability(unittest.TestCase): kwargs = {"model": "claude-sonnet-4-5", "call_type": "unknown"} self.assertEqual(h._get_span_name(kwargs), "chat claude-sonnet-4-5") + def test_passthrough_operation_names_the_span(self) -> None: + h: Final = self._make(env="gen_ai_latest_experimental") + kwargs: Final = { + "model": "unknown", + "call_type": "pass_through_endpoint", + PASSTHROUGH_OPERATION_KEY: GenAIOperation.RETRIEVAL, + } + self.assertEqual(h._get_span_name(kwargs), "retrieval unknown") + def test_generation_name_metadata_overrides_semconv_name(self): h = self._make(env="gen_ai_latest_experimental") kwargs = { diff --git a/tests/unit/proxy/pass_through_endpoints/test_upstream_usage_headers.py b/tests/unit/proxy/pass_through_endpoints/test_upstream_usage_headers.py index 9dd0233ad8b..33328e418cd 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_upstream_usage_headers.py +++ b/tests/unit/proxy/pass_through_endpoints/test_upstream_usage_headers.py @@ -4,10 +4,12 @@ import httpx import pytest from litellm.constants import MAX_SPEND_LOG_MODEL_NAME_LENGTH +from litellm.integrations.otel.model.semconv import PASSTHROUGH_OPERATION_KEY, GenAIOperation from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.proxy.pass_through_endpoints.upstream_usage_headers import ( UpstreamReportedUsage, apply_upstream_reported_model, + apply_upstream_reported_operation, apply_upstream_reported_usage, has_upstream_reported_usage, parse_upstream_reported_model, @@ -211,3 +213,24 @@ def test_model_header_alone_does_not_make_the_upstream_own_the_cost(): assert has_upstream_reported_usage(logging_obj) is False assert "response_cost" not in logging_obj.model_call_details + + +@pytest.mark.parametrize("operation", [GenAIOperation.RETRIEVAL, GenAIOperation.EMBEDDINGS]) +def test_apply_operation_records_the_operation_the_upstream_named(operation: GenAIOperation) -> None: + logging_obj: Final = _logging_obj() + + apply_upstream_reported_operation(logging_obj, _headers(**{"x-litellm-operation-name": operation.value})) + + assert logging_obj.model_call_details[PASSTHROUGH_OPERATION_KEY] is operation + + +@pytest.mark.parametrize( + "headers", + [{}, {"x-litellm-operation-name": "summarize"}, {"x-litellm-operation-name": "Retrieval"}], +) +def test_apply_operation_ignores_a_missing_or_unknown_name(headers: dict[str, str]) -> None: + logging_obj: Final = _logging_obj() + + apply_upstream_reported_operation(logging_obj, _headers(**headers)) + + assert PASSTHROUGH_OPERATION_KEY not in logging_obj.model_call_details