From 1b02cb64fb1799784f1ea9cb58d9644c074bb087 Mon Sep 17 00:00:00 2001 From: Yucheng He Date: Mon, 5 Oct 2026 12:47:25 -0700 Subject: [PATCH] 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 | 18 +++-- .../test_batch_line_item_metering_sinks.py | 78 +++++++++++++++---- 2 files changed, 73 insertions(+), 23 deletions(-) diff --git a/litellm/integrations/otel/plumbing/metrics.py b/litellm/integrations/otel/plumbing/metrics.py index 005658edb45..3af54f7f459 100644 --- a/litellm/integrations/otel/plumbing/metrics.py +++ b/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``. diff --git a/tests/unit/integrations/test_batch_line_item_metering_sinks.py b/tests/unit/integrations/test_batch_line_item_metering_sinks.py index a89d8ea9a53..525f6ae15d0 100644 --- a/tests/unit/integrations/test_batch_line_item_metering_sinks.py +++ b/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)