From 5f55d1719db64234c0d69e4dad99d46b01a27a0d Mon Sep 17 00:00:00 2001 From: "Jeff (Junze) Ma" <93145857+majunze2001@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:25:11 -0700 Subject: [PATCH] fix(server-metrics): parse OpenMetrics exposition for the vLLM Rust frontend (#1060) Signed-off-by: Jeff Ma (cherry picked from commit 8ba75f9a9e06377c0569f36f6d0a966d5307135a) --- .../mixins/base_metrics_collector_mixin.py | 18 +++++- src/aiperf/server_metrics/data_collector.py | 28 ++++++++- .../test_server_metrics_data_collector.py | 57 ++++++++++++++++++- 3 files changed, 98 insertions(+), 5 deletions(-) diff --git a/src/aiperf/common/mixins/base_metrics_collector_mixin.py b/src/aiperf/common/mixins/base_metrics_collector_mixin.py index 014a87bff1..beb045f710 100644 --- a/src/aiperf/common/mixins/base_metrics_collector_mixin.py +++ b/src/aiperf/common/mixins/base_metrics_collector_mixin.py @@ -23,6 +23,11 @@ from aiperf.transports.aiohttp_client import create_tcp_connector from aiperf.transports.http_defaults import AioHttpDefaults +# Response Content-Type prefixes for routing/rejecting metrics payloads. Matched +# against the lowercased header value, which may carry `; version=...; charset=...`. +JSON_CONTENT_TYPE_PREFIX = "application/json" +OPENMETRICS_CONTENT_TYPE_PREFIX = "application/openmetrics-text" + @dataclass(slots=True) class HttpTraceTiming: @@ -123,11 +128,17 @@ class FetchResult: is_duplicate: True if response content hash matches previous fetch, indicating metrics haven't changed. Callers can skip parsing when True to save CPU on repetitive data. + content_type: Lowercased HTTP `Content-Type` header from the response + (e.g. "application/openmetrics-text; version=1.0.0" for + vLLM's Rust frontend), or None when absent. Callers use it + to route between the OpenMetrics and classic Prometheus + exposition parsers. """ text: str | None trace_timing: HttpTraceTiming is_duplicate: bool = False + content_type: str | None = None # Type variables for records returned by collectors @@ -530,7 +541,7 @@ async def _fetch_metrics_text(self) -> FetchResult: # serve a JSON iteration-stats array at /metrics, which the # Prometheus parser cannot interpret. Reject up front so the # caller can auto-disable instead of looping on parse errors. - if content_type.startswith("application/json"): + if content_type.startswith(JSON_CONTENT_TYPE_PREFIX): raise IncompatibleMetricsEndpointError( f"endpoint {self._endpoint_url!r} returned non-Prometheus " f"content-type {content_type!r}; expected text/plain " @@ -548,7 +559,10 @@ async def _fetch_metrics_text(self) -> FetchResult: is_duplicate = response_hash == self._last_response_hash self._last_response_hash = response_hash return FetchResult( - text=text, trace_timing=timing, is_duplicate=is_duplicate + text=text, + trace_timing=timing, + is_duplicate=is_duplicate, + content_type=content_type or None, ) except (aiohttp.ClientConnectionError, RuntimeError) as e: if self.stop_requested or session.closed: diff --git a/src/aiperf/server_metrics/data_collector.py b/src/aiperf/server_metrics/data_collector.py index 7f5532913a..655578f12f 100644 --- a/src/aiperf/server_metrics/data_collector.py +++ b/src/aiperf/server_metrics/data_collector.py @@ -8,13 +8,19 @@ from dataclasses import dataclass, field from prometheus_client.metrics_core import Metric +from prometheus_client.openmetrics.parser import ( + text_string_to_metric_families as openmetrics_text_string_to_metric_families, +) from prometheus_client.parser import text_string_to_metric_families from aiperf.common.enums import PrometheusMetricType from aiperf.common.environment import Environment from aiperf.common.exceptions import IncompatibleMetricsEndpointError from aiperf.common.mixins import BaseMetricsCollectorMixin -from aiperf.common.mixins.base_metrics_collector_mixin import FetchResult +from aiperf.common.mixins.base_metrics_collector_mixin import ( + OPENMETRICS_CONTENT_TYPE_PREFIX, + FetchResult, +) from aiperf.common.models import ErrorDetails from aiperf.common.models.server_metrics_models import ( MetricFamily, @@ -260,7 +266,9 @@ def _parse_metrics_to_records( metrics_dict: dict[str, MetricFamily] = {} try: - for family in text_string_to_metric_families(fetch_result.text): + for family in self._parse_families( + fetch_result.text, fetch_result.content_type + ): # Skip _created metrics - these are timestamps indicating when the parent metric was created, not actual metric data # or _uptime metrics - these are timestamps indicating how long the server has been running. if ( @@ -326,6 +334,18 @@ def _parse_metrics_to_records( is_duplicate=fetch_result.is_duplicate, ) + def _parse_families(self, text: str, content_type: str | None) -> list[Metric]: + """Parse the body with the OpenMetrics parser for + ``application/openmetrics-text`` (classic mistypes its counters), falling + back to the classic parser if the stricter OpenMetrics parser rejects it. + """ + if content_type and content_type.startswith(OPENMETRICS_CONTENT_TYPE_PREFIX): + try: + return list(openmetrics_text_string_to_metric_families(text)) + except ValueError: + pass + return list(text_string_to_metric_families(text)) + def _process_simple_family(self, family: Metric) -> list[MetricSample]: """Process counter, gauge, or untyped metrics with de-duplication. @@ -347,6 +367,10 @@ def _process_simple_family(self, family: Metric) -> list[MetricSample]: samples_by_labels: dict[tuple, float] = {} for sample in family.samples: + # Skip OpenMetrics `_created` samples: they share the `_total` labels + # and would otherwise overwrite the real value in the de-dup below. + if sample.name.endswith("_created"): + continue # Skip samples with missing or non-finite values. Some servers emit # NaN while metrics are warming up; JSON transport converts that to # null, which is not a valid simple MetricSample on the receiver. diff --git a/tests/unit/server_metrics/test_server_metrics_data_collector.py b/tests/unit/server_metrics/test_server_metrics_data_collector.py index 2518998aeb..325e9927bc 100644 --- a/tests/unit/server_metrics/test_server_metrics_data_collector.py +++ b/tests/unit/server_metrics/test_server_metrics_data_collector.py @@ -17,7 +17,11 @@ from aiperf.server_metrics.data_collector import ServerMetricsDataCollector -def make_fetch_result(metrics_text: str, latency_ns: int = 1_000_000) -> FetchResult: +def make_fetch_result( + metrics_text: str, + latency_ns: int = 1_000_000, + content_type: str | None = None, +) -> FetchResult: """Create a FetchResult for testing.""" return FetchResult( text=metrics_text, @@ -28,6 +32,7 @@ def make_fetch_result(metrics_text: str, latency_ns: int = 1_000_000) -> FetchRe end_perf_ns=latency_ns, ), is_duplicate=False, + content_type=content_type, ) @@ -195,6 +200,56 @@ def test_skip_created_metrics(self): assert "histogram_seconds" in record.metrics assert "histogram_seconds_created" not in record.metrics + def test_openmetrics_counter_created_sample_does_not_overwrite_total(self) -> None: + """OpenMetrics keeps `foo_created` alongside `foo_total` under family + `foo` with the same labels, so without skipping it the creation + timestamp would win the last-value de-dup and replace the real value.""" + metrics_text = """# TYPE foo counter +# HELP foo Test counter +foo_total{label="a"} 5.0 +foo_created{label="a"} 123.0 +# EOF +""" + collector = ServerMetricsDataCollector("http://localhost:8081/metrics") + record = collector._parse_metrics_to_records( + make_fetch_result( + metrics_text, + content_type="application/openmetrics-text; version=1.0.0; charset=utf-8", + ) + ) + + assert record is not None + assert "foo" in record.metrics + assert record.metrics["foo"].type == PrometheusMetricType.COUNTER + samples = record.metrics["foo"].samples + assert len(samples) == 1 + assert samples[0].value == 5.0 + + def test_openmetrics_content_type_without_eof_falls_back_to_classic_parser( + self, + ) -> None: + """A body advertised as OpenMetrics but missing the strict `# EOF` trailer + makes the OpenMetrics parser raise; the collector must fall back to the + lenient classic parser instead of raising (which would disable all + server-metrics collection for the run).""" + metrics_text = """# TYPE foo counter +# HELP foo Test counter +foo_total{label="a"} 5.0 +""" + collector = ServerMetricsDataCollector("http://localhost:8081/metrics") + record = collector._parse_metrics_to_records( + make_fetch_result( + metrics_text, + content_type="application/openmetrics-text; version=1.0.0; charset=utf-8", + ) + ) + + # Classic fallback mistypes the counter into an `unknown` family named + # `foo_total`, but collection stays alive rather than disabling. + assert record is not None + assert "foo_total" in record.metrics + assert record.metrics["foo_total"].samples[0].value == 5.0 + def test_parse_metrics_with_labels(self): """Test parsing metrics with multiple label combinations.""" metrics_text = """# HELP http_requests_total Total HTTP requests