Merge 6f92ab91c1 into 9ffb82e139

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.

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