fix(otel): skip synthetic line-item durations in v2 metrics; isolate prometheus test

Follow-ups from review on the sink guards: record_failure() now skips
batch line items too (a failed line's interval measures retrieval and
emission, not its model call), and the prometheus guard test snapshots
and restores the process-wide registry collectors instead of leaving
them unregistered. The OTel v2 test is now a flag on/off A/B over the
whole metric snapshot: cost and usage values must match exactly and no
metric may gain datapoints with the flag on.

litellm/integrations/otel/plumbing/metrics.py

@@ -222,22 +222,24 @@ class GenAIMetricRecorder:
         start_time: datetime,
         end_time: datetime,
     ) -> None:
+        # A batch line item is metered by the aggregate aretrieve_batch event, and
+        # its child interval (parent retrieve start -> line emission) measures
+        # retrieval and callback processing rather than the line's model call, so
+        # skip every metric for it; spans for line items are still emitted.
+        if is_batch_line_item_event(kwargs):
+            return
         common_attrs: Final = self._filter_attributes(self._bounded_attributes(kwargs))
         duration_s: Final = (end_time - start_time).total_seconds()
         usage_is_replayed: Final = is_unbilled_non_inference_call_from_params(
             kwargs.get("call_type"), kwargs.get("litellm_params"), response_obj
         )
-        # A batch line item's tokens and cost are metered by the aggregate
-        # aretrieve_batch event; per-line samples here would double-count them.
-        # Duration samples for line items are still recorded.
-        batch_line_item: Final = is_batch_line_item_event(kwargs)
 
         self._metrics.operation_duration.record(duration_s, attributes=common_attrs)
-        if not usage_is_replayed and not batch_line_item:
+        if not usage_is_replayed:
             self._record_token_usage(response_obj, common_attrs)
 
         cost: Final = kwargs.get("response_cost")
-        if cost and not batch_line_item:
+        if cost:
             self._metrics.token_cost.record(cost, attributes=common_attrs)
 
         self._record_time_to_first_token(kwargs, common_attrs)
@@ -251,6 +253,10 @@ class GenAIMetricRecorder:
         start_time: datetime,
         end_time: datetime,
     ) -> None:
+        # A batch line item's interval measures retrieval and emission, not the
+        # line's model call, so its duration would be synthetic here too.
+        if is_batch_line_item_event(kwargs):
+            return
         """Record the one metric a failed request can honestly report: the
         operation's duration, tagged with ``error.type``.
 

tests/unit/integrations/test_batch_line_item_metering_sinks.py

@@ -144,9 +144,9 @@ class RecordingHTTP:
         return SimpleNamespace(status_code=202, text="ok", raise_for_status=lambda: None)
 
 
-async def _log_completed_batch(monkeypatch: pytest.MonkeyPatch, loggers: list) -> None:
+async def _log_completed_batch(monkeypatch: pytest.MonkeyPatch, loggers: list, flag: bool = True) -> None:
     batch_line_item_claim_cache.in_memory_cache.flush_cache()
-    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", flag, raising=False)
     saved_success = list(litellm._async_success_callback)
     saved_failure = list(litellm._async_failure_callback)
     litellm._async_success_callback = list(loggers)
@@ -273,16 +273,26 @@ async def test_prometheus_meters_only_the_aggregate_batch_event(monkeypatch: pyt
     from litellm.integrations.prometheus import PrometheusLogger
     from prometheus_client import REGISTRY
 
-    for collector in list(REGISTRY._collector_to_names.keys()):
+    saved_collectors = list(REGISTRY._collector_to_names.keys())
+    for collector in saved_collectors:
         REGISTRY.unregister(collector)
     logger = PrometheusLogger()
 
-    await _log_completed_batch(monkeypatch, [logger])
+    try:
+        await _log_completed_batch(monkeypatch, [logger])
+        spend_samples = []
+        for metric in REGISTRY.collect():
+            if metric.name == "litellm_spend_metric":
+                spend_samples = [sample.value for sample in metric.samples if sample.name.endswith("_total")]
+    finally:
+        for collector in list(REGISTRY._collector_to_names.keys()):
+            REGISTRY.unregister(collector)
+        for collector in saved_collectors:
+            try:
+                REGISTRY.register(collector)
+            except Exception:  # noqa: BLE001  # already re-registered by another holder
+                pass
 
-    spend_samples = []
-    for metric in REGISTRY.collect():
-        if metric.name == "litellm_spend_metric":
-            spend_samples = [sample.value for sample in metric.samples if sample.name.endswith("_total")]
     assert spend_samples, "litellm_spend_metric saw no samples at all"
     assert sum(spend_samples) == pytest.approx(AGGREGATE_COST), (
         f"Prometheus spend metric double-counted batch line items: {spend_samples}; "
@@ -348,9 +358,25 @@ async def test_otel_v1_meter_cost_skips_line_items_but_spans_stay(
     )
 
 
-@pytest.mark.asyncio
-async def test_otel_v2_meter_cost_skips_line_items(monkeypatch: pytest.MonkeyPatch) -> None:
-    pytest.importorskip("opentelemetry.sdk")
+def _reader_snapshot(reader: Any) -> dict[str, list[float]]:
+    """Every metric's datapoint values, keyed by metric name (sums for histograms)."""
+    data = reader.get_metrics_data()
+    out: dict[str, list[float]] = {}
+    if data is None:
+        return out
+    for resource_metrics in data.resource_metrics:
+        for scope_metrics in resource_metrics.scope_metrics:
+            for metric in scope_metrics.metrics:
+                for data_point in metric.data.data_points:
+                    value = getattr(data_point, "sum", None)
+                    if value is None:
+                        value = getattr(data_point, "value", None)
+                    if value is not None:
+                        out.setdefault(metric.name, []).append(value)
+    return out
+
+
+async def _run_batch_with_otel_v2(monkeypatch: pytest.MonkeyPatch, flag: bool) -> dict[str, list[float]]:
     from opentelemetry.sdk.metrics import MeterProvider
     from opentelemetry.sdk.metrics.export import InMemoryMetricReader
     from opentelemetry.sdk.trace import TracerProvider
@@ -365,11 +391,29 @@ async def test_otel_v2_meter_cost_skips_line_items(monkeypatch: pytest.MonkeyPat
         tracer_provider=TracerProvider(),
         meter_provider=MeterProvider(metric_readers=[reader]),
     )
+    await _log_completed_batch(monkeypatch, [logger], flag=flag)
+    return _reader_snapshot(reader)
 
-    await _log_completed_batch(monkeypatch, [logger])
 
-    costs = _cost_points(reader)
-    assert sum(costs) == pytest.approx(AGGREGATE_COST), (
-        f"OTel v2 gen_ai.usage.cost double-counted batch line items: {costs}; "
-        "line items must be metered only by the aggregate aretrieve_batch event"
-    )
+@pytest.mark.asyncio
+async def test_otel_v2_metrics_are_identical_with_the_flag_on_and_off(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    pytest.importorskip("opentelemetry.sdk")
+
+    off = await _run_batch_with_otel_v2(monkeypatch, flag=False)
+    on = await _run_batch_with_otel_v2(monkeypatch, flag=True)
+
+    for name in ("gen_ai.usage.cost", "gen_ai.client.token.usage"):
+        assert on.get(name) == off.get(name), (
+            f"OTel v2 {name} changed when the line-item flag was turned on; line items "
+            "must be metered only by the aggregate aretrieve_batch event. "
+            f"off={off.get(name)} on={on.get(name)}"
+        )
+    for name, values in off.items():
+        on_counts = [len(on.get(name, [])), len(values)]
+        assert on_counts[0] == on_counts[1], (
+            f"OTel v2 {name} gained synthetic per-line samples with the flag on: "
+            f"{on_counts[0]} datapoints on vs {on_counts[1]} off"
+        )
+    assert sum(off.get("gen_ai.usage.cost", [])) == pytest.approx(AGGREGATE_COST)