mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(otel): label configured pass-through routes litellm.passthrough and read the operation from the upstream (#48675)
This commit is contained in:
parent
4fa860b64c
commit
74392efbd4
13 changed files with 284 additions and 9 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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")),
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue