fix(server-metrics): parse OpenMetrics exposition for the vLLM Rust frontend

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:

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.

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