diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index a9bd93e6420..02cfa62e938 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -13,7 +13,7 @@ from typing_extensions import ReadOnly, TypedDict import litellm from litellm._logging import verbose_logger -from litellm.constants import KUBERNETES_POD_ROUTING_KEY +from litellm.constants import KUBERNETES_POD_ROUTING_KEY, LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL from litellm.integrations._types.open_inference import ( OpenInferenceSpanKindValues, ) @@ -1701,9 +1701,11 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): return key not in self._metric_attr_exclude return True - def _record_metrics(self, kwargs, response_obj, start_time, end_time): - duration_s: Final = (end_time - start_time).total_seconds() - params: Final = kwargs.get("litellm_params") or {} + def _metric_common_attributes( + self, + kwargs: dict[str, Any], # mutable-ok: the v1 logging kwargs contract + ) -> dict[str, Any]: # mutable-ok: the v1 metric helpers take a dict of attributes + params: Final[Mapping[str, Any]] = kwargs.get("litellm_params") or {} provider: Final = _provider_label(params.get("custom_llm_provider")) common_attrs = { @@ -1731,7 +1733,12 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): if hidden_params: common_attrs["hidden_params"] = safe_dumps(hidden_params) - common_attrs = self._filter_metric_attributes(common_attrs) + return self._filter_metric_attributes(common_attrs) + + def _record_metrics(self, kwargs, response_obj, start_time, end_time): + duration_s: Final = (end_time - start_time).total_seconds() + params: Final = kwargs.get("litellm_params") or {} + common_attrs: Final = self._metric_common_attributes(kwargs) if self._operation_duration_histogram: self._operation_duration_histogram.record(duration_s, attributes=common_attrs) @@ -2244,6 +2251,8 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): ) self._create_guardrail_span(kwargs=kwargs, context=guardrail_ctx) + self._record_failure_metrics(kwargs, start_time, end_time) + # Do NOT end parent span - it should be managed by its creator # External spans (from Langfuse, user code, HTTP headers, global context) must not be closed by LiteLLM # However, proxy-created spans should be closed here @@ -2254,6 +2263,25 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): ): parent_otel_span.end(end_time=self._to_ns(end_time)) + def _record_failure_metrics( + self, + kwargs: dict[str, Any], # mutable-ok: the v1 logging kwargs contract + start_time: datetime, + end_time: datetime, + ) -> None: + histogram: Final = self._operation_duration_histogram + if histogram is None or kwargs.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL): + return + # Imported here: the v2 metrics module imports this module at load time. + from litellm.integrations.otel.plumbing.metrics import resolve_error_type + + # hidden_params carries the provider's error response headers, a new value per request + attributes: Final = { + **{k: v for k, v in self._metric_common_attributes(kwargs).items() if k != "hidden_params"}, + "error.type": resolve_error_type(kwargs), + } + histogram.record((end_time - start_time).total_seconds(), attributes=attributes) + def _record_exception_on_span(self, span: Span, kwargs: _FailureKwargsView) -> str | None: """ Record exception information on the span using OTEL standard methods. diff --git a/tests/unit/integrations/test_opentelemetry.py b/tests/unit/integrations/test_opentelemetry.py index 5f749e951b1..8bd8817bd50 100644 --- a/tests/unit/integrations/test_opentelemetry.py +++ b/tests/unit/integrations/test_opentelemetry.py @@ -25,7 +25,7 @@ from opentelemetry.sdk._logs import ReadableLogRecord from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider from opentelemetry.sdk._logs.export import InMemoryLogExporter, LogRecordExportResult, SimpleLogRecordProcessor from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.metrics.export import InMemoryMetricReader, MetricsData +from opentelemetry.sdk.metrics.export import HistogramDataPoint, InMemoryMetricReader, MetricsData from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor, SpanExportResult @@ -55,7 +55,11 @@ from litellm.proxy._types import ManagementEndpointLoggingPayload, ProxyExceptio from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth from litellm.types.services import ServiceLoggerPayload, ServiceTypes from collections.abc import AsyncIterator, Callable, Mapping -from litellm.constants import KUBERNETES_POD_ROUTING_KEY, LOGGING_WORKER_MAX_TIME_PER_COROUTINE +from litellm.constants import ( + KUBERNETES_POD_ROUTING_KEY, + LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, + LOGGING_WORKER_MAX_TIME_PER_COROUTINE, +) from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.types.utils import ChatCompletionMessageToolCall, Function from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome @@ -1889,6 +1893,79 @@ class TestOpenTelemetry(unittest.TestCase): logs = log_exporter.get_finished_logs() self.assertFalse(logs, "Did not expect any logs") + @patch.dict(os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True) + def test_handle_failure_records_duration_with_error_type(self): + """A failed call records one operation duration point tagged with error.type.""" + for kwargs, expected in ( + ( + {"standard_logging_object": {"error_information": {"error_class": "RateLimitError"}}}, + "RateLimitError", + ), + ({"exception": ValueError("boom")}, "ValueError"), + ): + with self.subTest(expected=expected): + tracer_provider = TracerProvider() + metric_reader = InMemoryMetricReader() + otel = OpenTelemetry( + tracer_provider=tracer_provider, + meter_provider=MeterProvider(metric_readers=[metric_reader]), + ) + otel.tracer = tracer_provider.get_tracer(__name__) + start = datetime.now() + end = start + timedelta(seconds=2) + + otel.log_failure_event( + {"model": self.MODEL, "litellm_params": {"custom_llm_provider": "openai"}, **kwargs}, + None, + start, + end, + ) + + duration_metric = self.wait_for_metric(metric_reader, "gen_ai.client.operation.duration") + assert duration_metric is not None, "failure duration was not recorded" + points = [p for p in duration_metric.data.data_points if isinstance(p, HistogramDataPoint)] + self.assertEqual(len(points), 1) + self.assertEqual(points[0].count, 1) + self.assertEqual(points[0].sum, 2.0) + self.assertEqual(dict(points[0].attributes or {}).get("error.type"), expected) + self.assertEqual(dict(points[0].attributes or {}).get("gen_ai.request.model"), self.MODEL) + + def _failure_metric_points(self, failures: list[dict[str, object]]) -> list[HistogramDataPoint]: + tracer_provider = TracerProvider() + metric_reader = InMemoryMetricReader() + otel = OpenTelemetry(tracer_provider=tracer_provider, meter_provider=MeterProvider(metric_readers=[metric_reader])) + otel.tracer = tracer_provider.get_tracer(__name__) + start = datetime.now() + for kwargs in failures: + otel.log_failure_event(kwargs, None, start, start + timedelta(seconds=1)) + metric = self.wait_for_metric(metric_reader, "gen_ai.client.operation.duration") + if metric is None: + return [] + return [p for p in metric.data.data_points if isinstance(p, HistogramDataPoint)] + + @patch.dict(os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True) + def test_handle_failure_error_headers_do_not_split_series(self): + def failure(request_id: str) -> dict[str, object]: + return { + "model": self.MODEL, + "exception": ValueError("boom"), + "standard_logging_object": { + "hidden_params": {"additional_headers": {"x-request-id": request_id, "date": request_id}} + }, + } + + points = self._failure_metric_points([failure("req-1"), failure("req-2")]) + self.assertEqual(len(points), 1) + self.assertEqual(points[0].count, 2) + self.assertNotIn("hidden_params", dict(points[0].attributes or {})) + + @patch.dict(os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True) + def test_handle_failure_skips_proxy_gate_rejections(self): + points = self._failure_metric_points( + [{"model": "client-chosen", "exception": ValueError("bad key"), LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True}] + ) + self.assertEqual(points, []) + def test_get_span_name_with_generation_name(self): """Test _get_span_name returns generation_name when present""" otel = OpenTelemetry()