diff --git a/cvs/lib/inference/unittests/test_vllm_config_loader.py b/cvs/lib/inference/unittests/test_vllm_config_loader.py index 4275708a..bb3e00ee 100644 --- a/cvs/lib/inference/unittests/test_vllm_config_loader.py +++ b/cvs/lib/inference/unittests/test_vllm_config_loader.py @@ -13,6 +13,7 @@ from cvs.lib.inference.utils.vllm_config_loader import ( GATED_GPU_METRICS, + GATED_PROM_METRICS, Run, SeqCombo, Sweep, @@ -26,9 +27,10 @@ def _combo(name, isl="128", osl="2048"): def _full_gated_specs(): - """A spec for every gated client.* and gpu.* metric -- the minimum that - satisfies coverage. Values are inert so the set passes without asserting - anything; these tests pin the coverage gate, not the numbers.""" + """A spec for every gated client.*, gpu.*, and prom.* metric -- the + minimum that satisfies coverage. Values are inert so the set passes + without asserting anything; these tests pin the coverage gate, not the + numbers.""" out = {} for m in GATED_METRICS: kind = "max_ms" if m.endswith("_ms") else "max" if m == "failed" else "min" @@ -36,6 +38,8 @@ def _full_gated_specs(): for m in GATED_GPU_METRICS: kind = "max" if m in ("peak_gpu_memory_mb", "model_load_memory_mb", "model_load_s") else "min" out[f"gpu.{m}"] = {"kind": kind, "value": 0 if kind == "min" else 1e12} + for m in GATED_PROM_METRICS: + out[f"prom.{m}"] = {"kind": "max_ms", "value": 1e12} return out @@ -100,5 +104,71 @@ def test_all_five_gpu_metrics_are_gated(self): ) +class TestPromGatedMetricCoverage(unittest.TestCase): + """The prom.* axis of vllm_config_loader's _check_thresholds_cover_sweep. + + Mirrors TestGpuGatedMetricCoverage: prom.* is a fully separate, parallel + gated family (VLLM_PROMETHEUS_METRICS_SPEC.md Sec 6), not part of + client.*'s tiering machinery, so its coverage is proven independently + here rather than in test_vllm_report_preset.py. + """ + + _CELL = "ISL=128,OSL=2048,TP=8,CONC=16" + + def _variant_with(self, thresholds, enforce): + sw = Sweep( + sequence_combinations=[_combo("a")], + runs=[Run(combo="a", concurrency=16)], + ) + return VariantConfig( + schema_version=1, + framework="vllm", + gpu_arch="mi300x", + enforce_thresholds=enforce, + paths={ + "shared_fs": "/home/x", + "models_dir": "/home/x/models", + "log_dir": "/home/x/LOGS", + "hf_token_file": "/home/x/.hf", + }, + model={"id": "amd/Llama-3.1-70B-Instruct-FP8-KV", "remote": 0}, + params={"tensor_parallelism": "8"}, + sweep=sw, + thresholds=thresholds, + ) + + def test_full_gated_set_constructs(self): + vc = self._variant_with({self._CELL: _full_gated_specs()}, enforce=True) + self.assertEqual(vc.enforce_thresholds, True) + + def test_missing_prom_metric_raises_when_enforced(self): + specs = _full_gated_specs() + del specs["prom.queue_time_p50_ms"] + with self.assertRaises(ValidationError) as ctx: + self._variant_with({self._CELL: specs}, enforce=True) + self.assertIn("missing gated-metric specs", str(ctx.exception)) + self.assertIn("prom.queue_time_p50_ms", str(ctx.exception)) + + def test_missing_prom_metric_warns_when_record_only(self): + specs = _full_gated_specs() + del specs["prom.prefill_time_p95_ms"] + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + self._variant_with({self._CELL: specs}, enforce=False) + self.assertTrue(any("missing gated-metric specs" in str(x.message) for x in caught)) + + def test_all_four_prom_metrics_are_gated(self): + self.assertEqual( + GATED_PROM_METRICS, + { + "queue_time_p50_ms", + "queue_time_p95_ms", + "prefill_time_p50_ms", + "prefill_time_p95_ms", + }, + ) + + if __name__ == "__main__": unittest.main() diff --git a/cvs/lib/inference/unittests/test_vllm_server_metrics.py b/cvs/lib/inference/unittests/test_vllm_server_metrics.py new file mode 100644 index 00000000..8f9dbaaa --- /dev/null +++ b/cvs/lib/inference/unittests/test_vllm_server_metrics.py @@ -0,0 +1,330 @@ +''' +Copyright 2025 Advanced Micro Devices, Inc. +All rights reserved. + +Unit tests for cvs.lib.inference.utils.vllm_server_metrics. + +Black-box tests authored from the behavioral spec only (impl-blind). The +module contains pure parsers for vLLM's engine-side Prometheus `/metrics` +exposition-format text: no I/O, no hardware, pure text/dict transformations. + +Contract under test (from VLLM_PROMETHEUS_METRICS_SPEC.md Sec 3.2/7): + parse_prometheus_text(raw) -> {metric_name: {"buckets": {le: count}, + "sum": float, "count": float}} for histograms, {metric_name: float} + for bare gauges. Degrades to {} on empty/unparseable input; never raises. + diff_histogram(before, after) -> {le: after_count - before_count}, clamped + to >= 0. None if `after` has no buckets. + histogram_quantile(buckets, q) -> linear interpolation between bucket + boundaries; None on empty/zero-count buckets. + to_prom_metrics(before_text, after_text) -> the composed prom.* dict; + all-None (never partial, never a raise) if either scrape is + missing/unparseable. + +Framework: unittest.TestCase + self.subTest + unittest.mock (no pytest), +matching test_gpu.py's conventions. +''' + +import unittest +from unittest.mock import MagicMock + +from cvs.lib.inference.utils.vllm_server_metrics import ( + PROM_METRICS, + PROM_METRIC_UNITS, + diff_histogram, + histogram_quantile, + parse_prometheus_text, + to_prom_metrics, +) +from cvs.lib.inference.vllm_job import scrape_vllm_metrics + +# Shared bucket list (queue/prefill/decode/inference/e2e), seconds -- per +# VLLM_PROMETHEUS_METRICS_SPEC.md Sec 1. +_BUCKETS = [0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 50.0, 60.0, 120.0, 240.0, 480.0, 960.0, 1920.0, 7680.0] + + +def _histogram_text(name: str, cumulative_counts: dict, total_sum: float) -> str: + """Build real Prometheus exposition-format text for one histogram metric. + + cumulative_counts: {le_str: cumulative_count}, must include "+Inf". + """ + lines = [f"# HELP {name} test histogram", f"# TYPE {name} histogram"] + for le, count in cumulative_counts.items(): + lines.append(f'{name}_bucket{{le="{le}"}} {count}') + total = cumulative_counts["+Inf"] + lines.append(f"{name}_sum {total_sum}") + lines.append(f"{name}_count {total}") + return "\n".join(lines) + + +def _full_scrape_text(queue_counts, queue_sum, prefill_counts, prefill_sum) -> str: + parts = [ + _histogram_text("vllm:request_queue_time_seconds", queue_counts, queue_sum), + _histogram_text("vllm:request_prefill_time_seconds", prefill_counts, prefill_sum), + "# HELP vllm:num_requests_waiting test gauge", + "# TYPE vllm:num_requests_waiting gauge", + "vllm:num_requests_waiting 0", + ] + return "\n".join(parts) + + +def _cumulative(observations: list) -> dict: + """Bucket a list of raw second-values into cumulative le-counts using + the real shared bucket list, plus '+Inf'.""" + counts = {} + running = 0 + for b in _BUCKETS: + running += sum(1 for o in observations if o <= b) + counts[str(b)] = float(running) + counts["+Inf"] = float(len(observations)) + return counts + + +class TestParsePrometheusText(unittest.TestCase): + def test_empty_and_none_degrade_to_empty_dict(self): + for raw in (None, "", " ", "\n\n"): + with self.subTest(raw=repr(raw)): + self.assertEqual(parse_prometheus_text(raw), {}) + + def test_parses_histogram_buckets_sum_count(self): + text = _histogram_text( + "vllm:request_queue_time_seconds", + {"0.3": 2.0, "0.5": 3.0, "+Inf": 5.0}, + total_sum=1.75, + ) + out = parse_prometheus_text(text) + self.assertIn("vllm:request_queue_time_seconds", out) + hist = out["vllm:request_queue_time_seconds"] + self.assertEqual(hist["buckets"], {"0.3": 2.0, "0.5": 3.0, "+Inf": 5.0}) + self.assertEqual(hist["sum"], 1.75) + self.assertEqual(hist["count"], 5.0) + + def test_ignores_help_and_type_comment_lines(self): + text = "\n".join( + [ + "# HELP vllm:request_queue_time_seconds queue wait time", + "# TYPE vllm:request_queue_time_seconds histogram", + 'vllm:request_queue_time_seconds_bucket{le="0.3"} 1', + "vllm:request_queue_time_seconds_sum 0.2", + "vllm:request_queue_time_seconds_count 1", + ] + ) + out = parse_prometheus_text(text) + self.assertEqual(set(out.keys()), {"vllm:request_queue_time_seconds"}) + + def test_parses_bare_gauge_line(self): + text = "\n".join( + [ + "# TYPE vllm:num_requests_waiting gauge", + "vllm:num_requests_waiting 3", + ] + ) + out = parse_prometheus_text(text) + self.assertEqual(out["vllm:num_requests_waiting"], 3.0) + + def test_multiple_metrics_coexist(self): + text = _full_scrape_text( + _cumulative([0.1, 0.2]), 0.3, _cumulative([0.4]), 0.4 + ) + out = parse_prometheus_text(text) + self.assertIn("vllm:request_queue_time_seconds", out) + self.assertIn("vllm:request_prefill_time_seconds", out) + self.assertIn("vllm:num_requests_waiting", out) + + def test_never_raises_on_malformed_lines(self): + garbage_texts = [ + "not a valid prometheus line at all", + "vllm:request_queue_time_seconds_bucket{le=\"not_a_number_or_inf\"} abc", + "\x00\x01\x02 binary garbage", + "vllm:foo_sum not_a_float", + "vllm:foo_count", + ] + for raw in garbage_texts: + with self.subTest(raw=repr(raw)): + try: + parse_prometheus_text(raw) + except Exception as exc: # noqa: BLE001 + self.fail(f"parse_prometheus_text raised unexpectedly on {raw!r}: {exc!r}") + + def test_truncated_text_degrades_gracefully(self): + # A bucket line cut off mid-value; count line normal. + text = "vllm:request_queue_time_seconds_bucket{le=\"0.3\"} 1.\nvllm:request_queue_time_seconds_count 5" + try: + out = parse_prometheus_text(text) + except Exception as exc: # noqa: BLE001 + self.fail(f"parse_prometheus_text raised unexpectedly: {exc!r}") + # The malformed bucket line is simply skipped; count line still parses. + self.assertEqual(out.get("vllm:request_queue_time_seconds", {}).get("count"), 5.0) + + +class TestDiffHistogram(unittest.TestCase): + def test_simple_before_after_diff(self): + before = {"buckets": {"0.3": 2.0, "+Inf": 5.0}} + after = {"buckets": {"0.3": 4.0, "+Inf": 9.0}} + self.assertEqual(diff_histogram(before, after), {"0.3": 2.0, "+Inf": 4.0}) + + def test_missing_before_bucket_treated_as_zero(self): + before = {"buckets": {"+Inf": 5.0}} + after = {"buckets": {"0.3": 1.0, "+Inf": 6.0}} + self.assertEqual(diff_histogram(before, after), {"0.3": 1.0, "+Inf": 1.0}) + + def test_none_before_treated_as_all_zero(self): + after = {"buckets": {"0.3": 1.0, "+Inf": 3.0}} + self.assertEqual(diff_histogram(None, after), {"0.3": 1.0, "+Inf": 3.0}) + + def test_negative_diff_clamped_to_zero(self): + # Simulates a scrape taken across a server restart: after < before. + before = {"buckets": {"0.3": 10.0, "+Inf": 20.0}} + after = {"buckets": {"0.3": 1.0, "+Inf": 2.0}} + self.assertEqual(diff_histogram(before, after), {"0.3": 0.0, "+Inf": 0.0}) + + def test_none_after_returns_none(self): + before = {"buckets": {"0.3": 1.0, "+Inf": 1.0}} + self.assertIsNone(diff_histogram(before, None)) + + def test_empty_after_buckets_returns_none(self): + self.assertIsNone(diff_histogram({"buckets": {}}, {"buckets": {}})) + + +class TestHistogramQuantile(unittest.TestCase): + def test_zero_count_returns_none(self): + self.assertIsNone(histogram_quantile({"0.3": 0.0, "+Inf": 0.0}, 0.5)) + + def test_empty_or_none_returns_none(self): + self.assertIsNone(histogram_quantile({}, 0.5)) + self.assertIsNone(histogram_quantile(None, 0.5)) + + def test_all_mass_in_one_bucket(self): + # Every observation lands at or below 0.3s (the first bucket). + # Linear interpolation assumes uniform distribution between the + # implicit lower bound (0) and this bucket's boundary (0.3): + # target rank = 0.5*10 = 5; frac = (5-0)/(10-0) = 0.5; + # interpolated = 0 + 0.5*(0.3-0) = 0.15. + buckets = {"0.3": 10.0, "0.5": 10.0, "+Inf": 10.0} + self.assertAlmostEqual(histogram_quantile(buckets, 0.5), 0.15) + + def test_hand_computed_interpolation_p50(self): + # 0 <= x <= 0.3: 2 obs (cumulative 2); 0.3 < x <= 0.5: 8 obs (cumulative + # 10); target rank for p50 of 10 total = 5. Falls in the (0.3, 0.5] + # bucket: prev_bound=0.3 prev_count=2, bound=0.5 count=10. + # frac = (5-2)/(10-2) = 0.375; interpolated = 0.3 + 0.375*(0.5-0.3) = 0.375 + buckets = {"0.3": 2.0, "0.5": 10.0, "+Inf": 10.0} + self.assertAlmostEqual(histogram_quantile(buckets, 0.5), 0.375) + + def test_hand_computed_interpolation_p95(self): + # 20 total obs: cumulative 0.3->5, 0.5->18, 1.0->20. p95 target rank=19. + # Falls in (0.5, 1.0]: prev_bound=0.5 prev_count=18, bound=1.0 count=20. + # frac = (19-18)/(20-18) = 0.5; interpolated = 0.5 + 0.5*(1.0-0.5) = 0.75 + buckets = {"0.3": 5.0, "0.5": 18.0, "1.0": 20.0, "+Inf": 20.0} + self.assertAlmostEqual(histogram_quantile(buckets, 0.95), 0.75) + + def test_le_inf_only_bucket(self): + # Degenerate case: only the +Inf bucket present. + buckets = {"+Inf": 4.0} + self.assertEqual(histogram_quantile(buckets, 0.5), float("inf")) + + def test_never_raises_on_malformed_le_values(self): + try: + out = histogram_quantile({"not_a_number": 1.0, "+Inf": 2.0}, 0.5) + except Exception as exc: # noqa: BLE001 + self.fail(f"histogram_quantile raised unexpectedly: {exc!r}") + else: + self.assertIsNone(out) + + +class TestToPromMetrics(unittest.TestCase): + def test_all_prom_metrics_keys_present_shape(self): + expected_keys = {f"prom.{short}" for short, _unit in PROM_METRICS} + out = to_prom_metrics(None, None) + self.assertEqual(set(out.keys()), expected_keys) + + def test_none_before_or_after_yields_all_none(self): + after_text = _full_scrape_text(_cumulative([0.1]), 0.1, _cumulative([0.2]), 0.2) + for before, after in ((None, after_text), (after_text, None), (None, None)): + with self.subTest(before=before, after=after): + out = to_prom_metrics(before, after) + for k in out: + self.assertIsNone(out[k]) + + def test_unparseable_text_yields_all_none_not_raise(self): + try: + out = to_prom_metrics("garbage before", "garbage after") + except Exception as exc: # noqa: BLE001 + self.fail(f"to_prom_metrics raised unexpectedly: {exc!r}") + for k in out: + self.assertIsNone(out[k]) + + def test_end_to_end_realistic_before_after_pair(self): + # "before" scrape: server already served 3 queue-wait observations + # from a prior cell (0.1, 0.2, 0.4s) -- the reused-server baseline. + before_text = _full_scrape_text( + _cumulative([0.1, 0.2, 0.4]), 0.7, _cumulative([0.2, 0.3]), 0.5 + ) + # "after" scrape: this cell added 2 more queue-wait obs (0.6, 0.9s) + # and 1 more prefill obs (1.2s) on top of the same server's counters. + after_text = _full_scrape_text( + _cumulative([0.1, 0.2, 0.4, 0.6, 0.9]), + 2.2, + _cumulative([0.2, 0.3, 1.2]), + 1.7, + ) + out = to_prom_metrics(before_text, after_text) + # This cell's isolated queue-wait observations are exactly [0.6, 0.9] + # (0.6 falls in bucket 0.8, 0.9 falls in bucket 1.0); p50 of 2 obs + # falls in/around the first of the two remaining buckets. + self.assertIsNotNone(out["prom.queue_time_p50_ms"]) + self.assertIsNotNone(out["prom.queue_time_p95_ms"]) + self.assertIsNotNone(out["prom.prefill_time_p50_ms"]) + self.assertIsNotNone(out["prom.prefill_time_p95_ms"]) + # Values are in ms (seconds * 1000), and in the right ballpark given + # only [0.6, 0.9] contributed post-diff (600-1000ms range). + self.assertGreater(out["prom.queue_time_p50_ms"], 500) + self.assertLess(out["prom.queue_time_p50_ms"], 1100) + + def test_prom_metric_units_cover_every_metric(self): + for short, unit in PROM_METRICS: + with self.subTest(short=short): + self.assertEqual(PROM_METRIC_UNITS[short], unit) + + +class TestScrapeVllmMetrics(unittest.TestCase): + """I/O-boundary test for scrape_vllm_metrics (lives in vllm_job.py). + + Mirrors TestCaptureGpuMetrics's assert_called_once_with style: mock orch, + pin the exact command string, verify degrade-on-failure never raises. + """ + + def test_happy_path_returns_raw_text(self): + orch = MagicMock() + orch.exec_on_head.return_value = {"node0": "vllm:num_requests_waiting 0\n"} + out = scrape_vllm_metrics(orch, "http://0.0.0.0", "8888") + self.assertEqual(out, "vllm:num_requests_waiting 0\n") + orch.exec_on_head.assert_called_once_with("curl -sf http://0.0.0.0:8888/metrics") + + def test_timeout_kwarg_passed_through_when_given(self): + orch = MagicMock() + orch.exec_on_head.return_value = {"node0": "vllm:num_requests_waiting 0\n"} + scrape_vllm_metrics(orch, "http://0.0.0.0", "8888", timeout_s=30) + orch.exec_on_head.assert_called_once_with("curl -sf http://0.0.0.0:8888/metrics", timeout=30) + + def test_curl_failure_exception_degrades_to_none(self): + orch = MagicMock() + orch.exec_on_head.side_effect = RuntimeError("connection refused") + try: + out = scrape_vllm_metrics(orch, "http://0.0.0.0", "8888") + except Exception as exc: # noqa: BLE001 + self.fail(f"scrape_vllm_metrics raised unexpectedly: {exc!r}") + self.assertIsNone(out) + + def test_empty_output_degrades_to_none(self): + orch = MagicMock() + orch.exec_on_head.return_value = {"node0": ""} + self.assertIsNone(scrape_vllm_metrics(orch, "http://0.0.0.0", "8888")) + + def test_no_hosts_in_output_degrades_to_none(self): + orch = MagicMock() + orch.exec_on_head.return_value = {} + self.assertIsNone(scrape_vllm_metrics(orch, "http://0.0.0.0", "8888")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cvs/lib/inference/utils/vllm_config_loader.py b/cvs/lib/inference/utils/vllm_config_loader.py index f2cd7306..e9c07a8a 100644 --- a/cvs/lib/inference/utils/vllm_config_loader.py +++ b/cvs/lib/inference/utils/vllm_config_loader.py @@ -34,10 +34,17 @@ from typing_extensions import Literal from cvs.lib.inference.utils.vllm_parsing import GATED_METRICS +from cvs.lib.inference.utils.vllm_server_metrics import PROM_METRICS from cvs.lib.utils.config_loader import substitute_config from cvs.lib.utils.gpu import GPU_METRICS GATED_GPU_METRICS = {k for k, _unit in GPU_METRICS} +# A fully separate, parallel gated family, following GPU_METRICS's precedent +# rather than joining vllm_parsing.GATED_METRICS/METRIC_TIERS -- see +# VLLM_PROMETHEUS_METRICS_SPEC.md Sec 6 for why prom.* must not be mixed into +# the client.* tiering machinery (a locked invariant test partitions that set +# exactly). +GATED_PROM_METRICS = {k for k, _unit in PROM_METRICS} class _Forbid(BaseModel): @@ -245,9 +252,11 @@ def _check_thresholds_cover_sweep(self): problems.append(f"sweep cells with no threshold entry: {missing}") if extra: problems.append(f"threshold keys matching no sweep cell (typo?): {extra}") - gated_keys = [f"client.{m}" for m in sorted(GATED_METRICS)] + [ - f"gpu.{m}" for m in sorted(GATED_GPU_METRICS) - ] + gated_keys = ( + [f"client.{m}" for m in sorted(GATED_METRICS)] + + [f"gpu.{m}" for m in sorted(GATED_GPU_METRICS)] + + [f"prom.{m}" for m in sorted(GATED_PROM_METRICS)] + ) gated_gaps = {} for cell in sorted(expected & present): specs = self.thresholds.get(cell) or {} diff --git a/cvs/lib/inference/utils/vllm_server_metrics.py b/cvs/lib/inference/utils/vllm_server_metrics.py new file mode 100644 index 00000000..b48ef1ca --- /dev/null +++ b/cvs/lib/inference/utils/vllm_server_metrics.py @@ -0,0 +1,195 @@ +''' +Copyright 2025 Advanced Micro Devices, Inc. +All rights reserved. + +Pure parsers for vLLM's engine-side Prometheus `/metrics` endpoint. + +This module owns the *vocabulary and math* of the `prom.*` namespace -- the +mapping from two raw Prometheus text-exposition scrapes (one taken before a +sweep cell's client run, one taken after) to the namespaced metric dict that +downstream code (threshold files, the per-metric HTML rows, `evaluate_all`) +keys on. Deliberately free of I/O and orchestration, matching +`vllm_parsing.py`'s split: callers (`vllm_job.py`) fetch the scrape text, +this module turns it into numbers. + +Namespacing contract: `prom.*` -- percentile metrics interpolated from +Prometheus Histograms scraped off the live vLLM server, distinct from +`client.*` (measured by the load generator) and `gpu.*` (amd-smi snapshots). +See VLLM_PROMETHEUS_METRICS_SPEC.md Sec 1 for why this is its own namespace +rather than joining either of those. + +vLLM's histogram buckets are cumulative per scrape (each `le` bucket already +counts everything at or below it), but the *counters themselves* are +cumulative across the server process's lifetime, not per-request-run. Since a +server is reused across concurrency-only-differing sweep cells +(`server_signature()`), isolating one cell's observations requires diffing +two scrapes taken immediately before and after that cell's client run -- +never a single scrape. +''' + +from __future__ import annotations + +import re + +# Human-readable derived metrics exposed as HTML rows (one row per entry per +# cell), mirroring gpu.py's GPU_METRICS shape. These are the metrics this +# spec's Phase 1 closes a gap for -- see VLLM_PROMETHEUS_METRICS_SPEC.md Sec 5 +# for the primary/secondary/out-of-scope breakdown. +PROM_METRICS: list[tuple[str, str]] = [ + ("queue_time_p50_ms", "ms"), + ("queue_time_p95_ms", "ms"), + ("prefill_time_p50_ms", "ms"), + ("prefill_time_p95_ms", "ms"), +] +PROM_METRIC_UNITS: dict[str, str] = {k: u for k, u in PROM_METRICS} + +# vLLM Prometheus histogram names this module reads, and the (short_name +# prefix, quantile) pairs each feeds into PROM_METRICS above. +_QUEUE_TIME_METRIC = "vllm:request_queue_time_seconds" +_PREFILL_TIME_METRIC = "vllm:request_prefill_time_seconds" + +_QUANTILES: dict[str, float] = { + "p50": 0.50, + "p95": 0.95, +} + +_BUCKET_LINE_RE = re.compile( + r'^(?P[A-Za-z_:][A-Za-z0-9_:]*)_bucket\{[^}]*le="(?P[^"]+)"[^}]*\}\s+(?P[0-9.eE+-]+)\s*$' +) +_SUM_LINE_RE = re.compile(r"^(?P[A-Za-z_:][A-Za-z0-9_:]*)_sum(\{[^}]*\})?\s+(?P[0-9.eE+-]+)\s*$") +_COUNT_LINE_RE = re.compile(r"^(?P[A-Za-z_:][A-Za-z0-9_:]*)_count(\{[^}]*\})?\s+(?P[0-9.eE+-]+)\s*$") +_GAUGE_LINE_RE = re.compile(r"^(?P[A-Za-z_:][A-Za-z0-9_:]*)(\{[^}]*\})?\s+(?P[0-9.eE+-]+)\s*$") + + +def parse_prometheus_text(raw: "str | None") -> dict: + """Hand-rolled Prometheus text-exposition-format parser. + + Returns {metric_name: {"buckets": {le: cumulative_count}, "sum": float, + "count": float}} for every histogram found (le values are strings, + including "+Inf"), plus {metric_name: float} for any bare gauge/counter + line not part of a histogram. Ignores `# HELP`/`# TYPE` comment lines and + any line it can't parse. Degrades to {} on None/empty/unparseable input + -- never raises, matching gpu.py's `_try_parse` convention. + """ + if not raw: + return {} + histograms: dict[str, dict] = {} + gauges: dict[str, float] = {} + try: + for line in raw.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + m = _BUCKET_LINE_RE.match(line) + if m: + name = m.group("name") + hist = histograms.setdefault(name, {"buckets": {}, "sum": None, "count": None}) + hist["buckets"][m.group("le")] = float(m.group("count")) + continue + m = _SUM_LINE_RE.match(line) + if m: + name = m.group("name") + hist = histograms.setdefault(name, {"buckets": {}, "sum": None, "count": None}) + hist["sum"] = float(m.group("value")) + continue + m = _COUNT_LINE_RE.match(line) + if m: + name = m.group("name") + hist = histograms.setdefault(name, {"buckets": {}, "sum": None, "count": None}) + hist["count"] = float(m.group("value")) + continue + m = _GAUGE_LINE_RE.match(line) + if m: + gauges[m.group("name")] = float(m.group("value")) + except (ValueError, TypeError): + return {} + result: dict = dict(histograms) + for name, val in gauges.items(): + if name not in result: + result[name] = val + return result + + +def diff_histogram(before: "dict | None", after: "dict | None") -> "dict[str, float] | None": + """Per-bucket subtraction isolating one sweep cell's observations out of + a server-process-lifetime-cumulative Prometheus histogram. + + before/after: histogram dicts as returned by parse_prometheus_text() for + one metric name (i.e. {"buckets": {le: count}, "sum": ..., "count": ...}). + Returns {le: after_count - before_count} for every `le` present in + `after` (a bucket boundary can only appear once the server has been + running long enough to register it; `before` may be missing a boundary + `after` has if this is the server's first-ever scrape). Missing `before` + buckets are treated as 0. Negative diffs (e.g. a server restart between + scrapes resetting the counters) are clamped to 0 rather than propagated, + per VLLM_PROMETHEUS_METRICS_SPEC.md Sec 3.2/8.3. + + Returns None if `after` is missing/empty (nothing to diff against). + """ + if not after or not after.get("buckets"): + return None + before_buckets = (before or {}).get("buckets", {}) + after_buckets = after["buckets"] + return {le: max(0.0, count - before_buckets.get(le, 0.0)) for le, count in after_buckets.items()} + + +def histogram_quantile(buckets: "dict[str, float] | None", q: float) -> "float | None": + """Linear interpolation between cumulative bucket boundaries -- the same + algorithm PromQL's `histogram_quantile()` uses. + + buckets: {le: cumulative_count}, `le` values are numeric strings or + "+Inf". Returns None if buckets is empty/missing or total count (the + "+Inf" bucket) is 0 -- mirrors vllm_parsing.py's `_safe_div` None-safe + convention, never a ZeroDivisionError. + """ + if not buckets: + return None + try: + parsed = sorted(((float("inf") if le == "+Inf" else float(le), count) for le, count in buckets.items())) + except (TypeError, ValueError): + return None + total = parsed[-1][1] + if total <= 0: + return None + target = q * total + prev_bound, prev_count = 0.0, 0.0 + for bound, count in parsed: + if count >= target: + if bound == prev_bound or count == prev_count: + return bound + frac = (target - prev_count) / (count - prev_count) + return prev_bound + frac * (bound - prev_bound) + prev_bound, prev_count = bound, count + return prev_bound + + +def _quantile_ms(before_metrics: dict, after_metrics: dict, metric_name: str, q: float) -> "float | None": + diffed = diff_histogram(before_metrics.get(metric_name), after_metrics.get(metric_name)) + seconds = histogram_quantile(diffed, q) + return None if seconds is None else seconds * 1000.0 + + +def to_prom_metrics(before_text: "str | None", after_text: "str | None") -> dict: + """Composed entry point: two raw scrape texts -> the `prom.*` metric dict. + + Analogous to vllm_parsing.py's to_client_metrics(). Returns an all-None + dict (never a partial one, never a raise) if either scrape is + missing/unparseable -- see VLLM_PROMETHEUS_METRICS_SPEC.md Sec 3.4: a + transport failure must degrade every prom.* key for the cell, not crash it. + """ + all_none = {f"prom.{short}": None for short, _unit in PROM_METRICS} + if not before_text or not after_text: + return all_none + + before_metrics = parse_prometheus_text(before_text) + after_metrics = parse_prometheus_text(after_text) + if not before_metrics or not after_metrics: + return all_none + + result = dict(all_none) + for qname, q in _QUANTILES.items(): + result[f"prom.queue_time_{qname}_ms"] = _quantile_ms(before_metrics, after_metrics, _QUEUE_TIME_METRIC, q) + result[f"prom.prefill_time_{qname}_ms"] = _quantile_ms( + before_metrics, after_metrics, _PREFILL_TIME_METRIC, q + ) + return result diff --git a/cvs/lib/inference/vllm_job.py b/cvs/lib/inference/vllm_job.py index 0c298d31..d2d6e7cb 100644 --- a/cvs/lib/inference/vllm_job.py +++ b/cvs/lib/inference/vllm_job.py @@ -47,6 +47,34 @@ log = globals.log +def scrape_vllm_metrics(orch, base_url: str, port_no: str, timeout_s: "float | None" = None) -> "str | None": + """One-shot scrape of vLLM's `/metrics` Prometheus endpoint, head-only. + + Mirrors capture_gpu_metrics()'s one-shot-exec shape (gpu.py): a single, + synchronous, main-thread orch.exec_on_head call, never backgrounded. Two + calls to this function (one before, one after a cell's client run) bracket + test_vllm_inference the same way start_gpu_poller/stop_and_collect_gpu_poller + do, but this only needs point-in-time reads, not a continuous poll -- see + VLLM_PROMETHEUS_METRICS_SPEC.md Sec 3.1 for why a background thread/poller + must never be used here (the SSH-session race that started that spec's + concurrency-safety note applies to this bracket too). + + Returns the raw exposition-format text, or None if the curl fails + (endpoint down, timeout, non-2xx) or comes back empty -- never raises. + """ + kwargs = {"timeout": timeout_s} if timeout_s is not None else {} + try: + out = orch.exec_on_head(f"curl -sf {base_url}:{port_no}/metrics", **kwargs) + except Exception as exc: + log.warning("scrape_vllm_metrics: exec_on_head failed: %s", exc) + return None + text = next(iter(out.values()), None) if out else None + if not text or not str(text).strip(): + log.warning("scrape_vllm_metrics: empty/failed scrape from %s:%s/metrics", base_url, port_no) + return None + return text + + class VllmJob: """Unified vLLM benchmark job for single-node and multinode distributed runs. diff --git a/cvs/tests/inference/vllm/vllm.py b/cvs/tests/inference/vllm/vllm.py index 45e7d1bd..4fc84d4e 100644 --- a/cvs/tests/inference/vllm/vllm.py +++ b/cvs/tests/inference/vllm/vllm.py @@ -36,7 +36,12 @@ ) from cvs.lib.utils.verdict import evaluate_all from cvs.lib.inference.utils.vllm_parsing import CLIENT_METRICS as _METRICS, CLIENT_METRIC_UNITS as _METRIC_UNITS -from cvs.lib.inference.vllm_job import VllmJob +from cvs.lib.inference.utils.vllm_server_metrics import ( + PROM_METRICS, + PROM_METRIC_UNITS, + to_prom_metrics, +) +from cvs.lib.inference.vllm_job import VllmJob, scrape_vllm_metrics import importlib.util as _ilu import pathlib as _pl @@ -105,6 +110,15 @@ def pytest_generate_tests(metafunc): gpu_metric_cases.append((combo, c, short)) gpu_metric_ids.append(cid + "-" + short) metafunc.parametrize("seq_combo,concurrency,gpu_metric", gpu_metric_cases, ids=gpu_metric_ids) + elif "prom_metric" in metafunc.fixturenames: + if cases: + prom_metric_cases = [] + prom_metric_ids = [] + for (combo, c), cid in zip(cases, ids): + for short, _unit in PROM_METRICS: + prom_metric_cases.append((combo, c, short)) + prom_metric_ids.append(cid + "-" + short) + metafunc.parametrize("seq_combo,concurrency,prom_metric", prom_metric_cases, ids=prom_metric_ids) elif "seq_combo" in metafunc.fixturenames and "concurrency" in metafunc.fixturenames and cases: metafunc.parametrize("seq_combo,concurrency", cases, ids=ids) @@ -350,6 +364,13 @@ def test_vllm_inference(orch, variant_config, hf_token, seq_combo, concurrency, run_id=f"{request.node.nodeid}_{isl}_{osl}_{concurrency}", nodes=None if int(variant_config.params.nnodes) == 1 else list(job.orch.hosts), ) + # One-shot scrape of vLLM's own /metrics endpoint, immediately before + # the client run -- not before the server-reuse branch above, since a + # reused server has no guaranteed-zero baseline (it may have already + # served the smoke test and/or prior cells). Two single, sequential, + # main-thread exec calls -- see VLLM_PROMETHEUS_METRICS_SPEC.md Sec 3.1 + # for why this is safe from the SSH-session race the GPU poller hit. + prom_before = scrape_vllm_metrics(orch, job.base_url, job.port_no) try: job.run_client() job.wait_client_complete() @@ -361,6 +382,7 @@ def test_vllm_inference(orch, variant_config, hf_token, seq_combo, concurrency, model_load_s=load_s, model_load_memory_mb=load_mb, ) + prom_after = scrape_vllm_metrics(orch, job.base_url, job.port_no) results = job.parse_results() except Exception: lifecycle.failed = True @@ -377,8 +399,10 @@ def test_vllm_inference(orch, variant_config, hf_token, seq_combo, concurrency, "gpu.gpu_bandwidth_util_pct": agg.get("gpu_bandwidth_util_pct"), "gpu.gpu_compute_util_pct": agg.get("gpu_compute_util_pct"), } + prom_results = to_prom_metrics(prom_before, prom_after) for host_actuals in results.values(): host_actuals.update(gpu_results) + host_actuals.update(prom_results) key = ( variant_config.model.id, @@ -460,6 +484,42 @@ def test_gpu_metric(seq_combo, concurrency, gpu_metric, inf_res_dict, variant_co evaluate_all(actuals, {full: spec}) +def test_prom_metric(seq_combo, concurrency, prom_metric, inf_res_dict, variant_config, lifecycle, request): + """One pytest test (= one HTML row) per vLLM /metrics-derived metric per cell.""" + if lifecycle.failed: + pytest.skip("a prior lifecycle stage failed") + isl = seq_combo["isl"] + osl = seq_combo["osl"] + key = ( + variant_config.model.id, + variant_config.gpu_arch, + isl, + osl, + seq_combo.get("name", "default"), + concurrency, + ) + if key not in inf_res_dict: + pytest.skip(f"no recorded results for cell {key!r} (inference did not run)") + host_dict = inf_res_dict[key] + _host, actuals = next(iter(host_dict.items())) + full = "prom." + prom_metric + value = actuals.get(full) + unit = PROM_METRIC_UNITS.get(prom_metric, "-") + request.node.user_properties.append(("metric_value", value)) + request.node.user_properties.append(("metric_unit", unit)) + + if value is None: + pytest.skip(f"{full}: no value recorded (/metrics scrape unavailable or unparseable)") + + if not variant_config.enforce_thresholds: + return + cell = variant_config.cell_key(isl, osl, concurrency) + spec = (variant_config.thresholds.get(cell) or {}).get(full) + if spec is None: + return + evaluate_all(actuals, {full: spec}) + + def test_teardown(orch, lifecycle, request): """Final stage: explicit container teardown, timed, asserting it is gone.""" name = orch.get_container_name(orch.container_config, orch.container_config["image"])