diff --git a/src/hyperloom/inference_optimizer/breakdown/session_package.py b/src/hyperloom/inference_optimizer/breakdown/session_package.py index b7fb3f0e18..fc02cdc902 100644 --- a/src/hyperloom/inference_optimizer/breakdown/session_package.py +++ b/src/hyperloom/inference_optimizer/breakdown/session_package.py @@ -71,6 +71,7 @@ "reports/conc_sweep_summary.json", "runs/**/kv_metrics.json", "runs/**/agentx_timeline.jsonl", + "runs/**/gpu_metrics.json", "reports/sbd_v6/timeline/*.json", "reports/sbd_v6/write_warnings.jsonl", "reports/trace/*.jsonl", diff --git a/src/hyperloom/inference_optimizer/tests/test_gpu_metrics.py b/src/hyperloom/inference_optimizer/tests/test_gpu_metrics.py new file mode 100644 index 0000000000..3297257ca4 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_gpu_metrics.py @@ -0,0 +1,291 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for the per-round GPU telemetry artifact. + +The blocks below are the two shapes that actually occur, taken from production +``benchmark_report.json`` files rather than invented: Magpie's pre-aggregated +``{min,max,avg}`` form, and the multi-node harvester's flat per-sample form. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any + +from hyperloom.orchestrator.actions.executors._gpu_metrics import ( + GPU_ARTIFACT_NAME, + gpu_metrics_from_report, + write_gpu_metrics, +) + +# Verbatim from a production single-node round. Note what is *not* here: no +# utilization and no VRAM -- Magpie's monitor samples neither. +MAGPIE_BLOCK: dict[str, Any] = { + "sample_count": 35, + "duration_sec": 70.55, + "temperature_c": {"min": 56.0, "max": 63.0, "avg": 58.6}, + "gpu_clock_mhz": {"min": 94, "max": 2402, "avg": 1030.1}, + "mem_clock_mhz": {"min": 2000, "max": 2000, "avg": 2000.0}, + "power_watts": {"min": 253.0, "max": 943.0, "avg": 476.5}, +} + +# The second block from that same session, so the weighting below is real. +MAGPIE_BLOCK_2: dict[str, Any] = { + "sample_count": 126, + "duration_sec": 259.37, + "temperature_c": {"min": 56.0, "max": 63.0, "avg": 58.8}, + "gpu_clock_mhz": {"min": 94, "max": 2400, "avg": 1682.3}, + "mem_clock_mhz": {"min": 2000, "max": 2000, "avg": 2000.0}, + "power_watts": {"min": 253.0, "max": 848.0, "avg": 471.6}, +} + + +def _round(tmp_path: Path, gpu_monitor: Any) -> Path: + """A round workspace carrying a benchmark report.""" + (tmp_path / "benchmark_report.json").write_text(json.dumps({"gpu_monitor": gpu_monitor}), encoding="utf-8") + return tmp_path + + +def test_the_magpie_block_yields_real_numbers(): + """The regression this started from: this exact shape read as all zeros.""" + out = gpu_metrics_from_report({"gpu_monitor": MAGPIE_BLOCK}) + + assert out["avg_power_w"] == 476.5 + assert out["max_power_w"] == 943.0 + assert out["avg_temp_c"] == 58.6 + assert out["max_temp_c"] == 63.0 + assert out["avg_clock_mhz"] == 1030.1 + assert out["avg_mem_clock_mhz"] == 2000.0 + assert out["samples"] == 35 + assert out["blocks"] == 1 + + +def test_means_are_weighted_by_sample_count(): + """A 35-sample block must not pull the round's mean as hard as a 126-sample one.""" + out = gpu_metrics_from_report({"gpu_monitor": [MAGPIE_BLOCK, MAGPIE_BLOCK_2]}) + + # (476.5*35 + 471.6*126) / 161 + assert out["avg_power_w"] == 472.67 + assert out["avg_temp_c"] == 58.76 + assert out["max_power_w"] == 943.0 + assert out["samples"] == 161 + assert out["blocks"] == 2 + + +def test_magpie_reports_no_utilization_or_vram(): + """Single-node reality, and the reason those fields are tri-state. + + Null, not 0.0: a 0% utilization reading asserts the GPU sat idle through the + round, which is the opposite of what an unsampled metric means. + """ + out = gpu_metrics_from_report({"gpu_monitor": MAGPIE_BLOCK}) + + assert out["avg_gpu_util_pct"] is None + assert out["max_gpu_util_pct"] is None + assert out["avg_vram_pct"] is None + assert out["max_vram_pct"] is None + + +def test_multi_node_flat_samples_carry_utilization_and_vram(): + """``_row_to_gpu_sample``'s own field names, on the path that has them.""" + flat = [ + {"power_w": 100.0, "temperature_c": 50.0, "clock_mhz": 1000.0, "gpu_util_pct": 80.0, "vram_pct": 40.0}, + {"power_w": 200.0, "temperature_c": 60.0, "clock_mhz": 2000.0, "gpu_util_pct": 90.0, "vram_pct": 50.0}, + ] + out = gpu_metrics_from_report({"gpu_monitor": flat}) + + assert out["avg_gpu_util_pct"] == 85.0 + assert out["max_gpu_util_pct"] == 90.0 + assert out["avg_vram_pct"] == 45.0 + assert out["max_vram_pct"] == 50.0 + # A flat scalar is that sample's mean and its peak alike. + assert out["avg_power_w"] == 150.0 + assert out["max_power_w"] == 200.0 + assert out["samples"] == 2 + + +def test_a_measured_zero_is_not_a_missing_reading(): + """The original defect in one line: ``_avg(a) or _avg(b)`` could not say this.""" + out = gpu_metrics_from_report({"gpu_monitor": {"power_w": 0.0, "power": 500.0}}) + + assert out["avg_power_w"] == 0.0 + assert out["max_power_w"] == 0.0 + + +def test_one_alias_wins_per_block_so_the_peak_cannot_fall_below_the_mean(): + """Resolving per statistic lets the mean come from one key and the peak from + a stale sibling, which reports a maximum below the average.""" + out = gpu_metrics_from_report({"gpu_monitor": {"power_watts": {"avg": 300.0, "max": 316.0}, "power_w": 12.0}}) + + assert out["avg_power_w"] == 300.0 + assert out["max_power_w"] == 316.0 + + +def test_a_statistic_the_winning_alias_omits_stays_null(): + """Not borrowed from another key: the producer did not report it.""" + out = gpu_metrics_from_report({"gpu_monitor": {"power_watts": {"avg": 300.0}, "power_w": 12.0}}) + + assert out["avg_power_w"] == 300.0 + assert out["max_power_w"] is None + + +def test_absolute_vram_is_not_folded_into_the_percent_field(): + """A MiB reading under a ``_pct`` name would be a units error, not a fallback.""" + out = gpu_metrics_from_report({"gpu_monitor": {"vram_used_mb": 81920.0, "memory_used_bytes": 8.6e10}}) + + assert out["avg_vram_pct"] is None + assert out["max_vram_pct"] is None + # Nothing readable, so nothing is credited as measured. + assert out["blocks"] == 1 + assert out["samples"] == 0 + + +def test_a_block_that_measured_nothing_is_not_counted_as_a_sample(): + """``sample_count`` beside no recognised metric would put a large count next + to a row of nulls.""" + out = gpu_metrics_from_report({"gpu_monitor": {"sample_count": 27000, "duration_sec": 10.0}}) + + assert out["blocks"] == 1 + assert out["samples"] == 0 + assert out["avg_power_w"] is None + + +def test_zero_sample_count_is_not_promoted_to_one(): + """A monitor that started and sampled nothing reported zero, not one.""" + out = gpu_metrics_from_report({"gpu_monitor": {"sample_count": 0, "power_watts": {"avg": 300.0, "max": 300.0}}}) + + assert out["samples"] == 0 + + +def test_a_report_without_gpu_monitor_yields_nothing(): + """Absence of a block is a different answer from a block full of nulls.""" + assert gpu_metrics_from_report({"throughput": {"avg": 1.0}}) == {} + assert gpu_metrics_from_report(None) == {} + + +def test_the_artifact_is_written_into_the_round(tmp_path): + """Per round, beside the report it was read from.""" + ws = _round(tmp_path, MAGPIE_BLOCK) + + path = write_gpu_metrics(ws) + + assert path is not None + payload = json.loads((ws / GPU_ARTIFACT_NAME).read_text(encoding="utf-8")) + assert payload["avg_power_w"] == 476.5 + assert payload["schema_version"] == 1 + assert payload["source"] == "benchmark_report.json" + + +def test_no_report_writes_no_artifact(tmp_path): + """Every round that never produced a benchmark report is this case.""" + assert write_gpu_metrics(tmp_path) is None + assert not (tmp_path / GPU_ARTIFACT_NAME).exists() + + +def test_a_report_without_telemetry_writes_no_artifact(tmp_path): + """An empty artifact would claim the question was asked and answered.""" + (tmp_path / "benchmark_report.json").write_text(json.dumps({"throughput": {}}), encoding="utf-8") + + assert write_gpu_metrics(tmp_path) is None + assert not (tmp_path / GPU_ARTIFACT_NAME).exists() + + +def test_an_unreadable_report_does_not_raise(tmp_path): + """Telemetry must never fail a round.""" + (tmp_path / "benchmark_report.json").write_text("{not json", encoding="utf-8") + + assert write_gpu_metrics(tmp_path) is None + + +def test_the_harvest_writes_it_for_every_round(tmp_path, monkeypatch): + """The one hook: wherever a round's artifacts are finalised.""" + from hyperloom.orchestrator.actions.executors import benchmark_result as br + + monkeypatch.setattr(br, "harvest_mn_gpu_metrics", lambda *_a, **_k: {}) + ws = _round(tmp_path, MAGPIE_BLOCK) + + br.harvest_leaked_artifacts(ws) + + assert json.loads((ws / GPU_ARTIFACT_NAME).read_text(encoding="utf-8"))["avg_power_w"] == 476.5 + + +def test_a_block_offering_only_min_contributes_nothing(tmp_path): + """``min`` is not among the statistics reported, so a block with only ``min`` measured nothing reportable. + + Crediting its ``sample_count`` would put 27000 samples beside a row of nulls, + which reads as a large, well-sampled round that somehow measured nothing. + """ + out = gpu_metrics_from_report({"gpu_monitor": {"sample_count": 27000, "power_watts": {"min": 100.0}}}) + + assert out["samples"] == 0 + assert out["avg_power_w"] is None + assert out["max_power_w"] is None + + +def test_a_min_only_block_does_not_inflate_a_real_one(tmp_path): + """The weighting must come from the blocks that actually reported.""" + out = gpu_metrics_from_report( + {"gpu_monitor": [MAGPIE_BLOCK, {"sample_count": 27000, "power_watts": {"min": 100.0}}]} + ) + + assert out["samples"] == 35 + assert out["avg_power_w"] == 476.5 + + +def test_a_failed_write_is_reported_not_swallowed(tmp_path, monkeypatch, caplog): + """Telemetry that was read but could not be written is a defect, and has to be audible. + + The bug this artifact exists to fix went unnoticed for 101 sessions because + nothing said anything; a silent writer would repeat exactly that. + """ + from hyperloom.common import io as common_io + + def explode(*_a, **_k): + raise OSError("read-only workspace") + + monkeypatch.setattr(common_io, "atomic_write_json", explode) + ws = _round(tmp_path, MAGPIE_BLOCK) + + with caplog.at_level(logging.WARNING): + assert write_gpu_metrics(ws) is None + assert "could not be written" in caplog.text + + +def test_a_round_without_telemetry_stays_quiet(tmp_path, caplog): + """The ordinary case must not cry wolf, or the warning above stops meaning anything.""" + (tmp_path / "benchmark_report.json").write_text(json.dumps({"throughput": {}}), encoding="utf-8") + + with caplog.at_level(logging.WARNING): + assert write_gpu_metrics(tmp_path) is None + assert caplog.text == "" + + +def test_a_report_that_lands_after_the_harvest_is_still_written(tmp_path): + """Harvest runs before the report is guaranteed to exist, so it cannot be the only chance. + + Magpie can finish writing ``benchmark_report.json`` after the subprocess is + reaped; the round settles on a report the harvest never saw. + """ + from hyperloom.orchestrator.actions.executors import benchmark_result as br + from hyperloom.orchestrator.actions.executors._gpu_metrics import write_gpu_metrics_from_report + + # Harvest first, with no report on disk yet: nothing to write. + br.harvest_leaked_artifacts(tmp_path) + assert not (tmp_path / GPU_ARTIFACT_NAME).exists() + + # The report settles afterwards, which is where the second attempt reads it. + report = {"gpu_monitor": MAGPIE_BLOCK} + (tmp_path / "benchmark_report.json").write_text(json.dumps(report), encoding="utf-8") + write_gpu_metrics_from_report(tmp_path, report, source="benchmark_report.json") + + assert json.loads((tmp_path / GPU_ARTIFACT_NAME).read_text(encoding="utf-8"))["avg_power_w"] == 476.5 + + +def test_the_artifact_is_in_the_package_globs(): + """It lives in the round workspace, which the bundle does not otherwise reach.""" + from hyperloom.inference_optimizer.breakdown.session_package import PACKAGE_GLOBS + + assert "runs/**/gpu_metrics.json" in PACKAGE_GLOBS diff --git a/src/hyperloom/inference_optimizer/tests/test_target_analysis_executor.py b/src/hyperloom/inference_optimizer/tests/test_target_analysis_executor.py index 42d669f920..423848d506 100644 --- a/src/hyperloom/inference_optimizer/tests/test_target_analysis_executor.py +++ b/src/hyperloom/inference_optimizer/tests/test_target_analysis_executor.py @@ -40,6 +40,20 @@ def _ctx(session_dir: Path, params: dict[str, Any] | None = None) -> _Ctx: ) +@pytest.fixture(autouse=True) +def _clear_query_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Answer the executor's query from the arguments each test sets, not from the process. + + ``TargetAnalysisExecutor`` resolves every query field as ``params`` > environment > + ``SharedState``, so a variable left behind by anything that ran earlier in the same + worker silently outranks the state a test builds. That is not hypothetical: a stale + ``PRECISION`` turned an expected match into ``no_match``, and only for whichever + xdist worker happened to inherit it. + """ + for name in ("PRECISION", "FRAMEWORK", "MODEL_PATH", "ISL", "OSL"): + monkeypatch.delenv(name, raising=False) + + @pytest.fixture def session_dir(tmp_path: Path) -> Path: sd = tmp_path / "sess" diff --git a/src/hyperloom/orchestrator/actions/executors/_gpu_metrics.py b/src/hyperloom/orchestrator/actions/executors/_gpu_metrics.py new file mode 100644 index 0000000000..59fda2215a --- /dev/null +++ b/src/hyperloom/orchestrator/actions/executors/_gpu_metrics.py @@ -0,0 +1,232 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Per-round GPU telemetry, normalised out of the round's own benchmark report. + +The numbers are already collected -- Magpie's ``GPUMonitor`` on a single node, the pod-side ``rocm-smi`` sampler on +several -- and land in ``benchmark_report.json`` under ``gpu_monitor``. What was missing is a reading of them that a +consumer can trust, per round, without knowing which producer wrote the block. + +Two things make that non-trivial, and both have burned this data before: + +* **The producers disagree on names and on shape.** Magpie writes ``power_watts`` as a pre-aggregated + ``{min, max, avg}`` block; the multi-node harvester writes ``power_w`` as a flat scalar per sample. Reading only the + short spelling is what left every single-node session's GPU numbers reading 0.0 -- not absent, *zero*, which is a + plausible-looking lie. +* **Absent and zero are different findings.** A card that drew no power and a card nobody sampled must not produce the + same number. Every metric here is ``float | None`` and is never coerced. + +Written per round rather than aggregated per session on purpose: a session mixes baseline, explore and roofline rounds +whose power and thermal behaviour have nothing to do with each other, and averaging them describes no round that +actually ran. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any + +log = logging.getLogger(__name__) + + +__all__ = [ + "GPU_ARTIFACT_NAME", + "gpu_metrics_from_report", + "write_gpu_metrics", + "write_gpu_metrics_from_report", +] + +#: Artifact written into the round's workspace, beside ``benchmark_report.json``. +GPU_ARTIFACT_NAME = "gpu_metrics.json" + +# ``gpu_monitor`` metric aliases, current producer name first. Magpie emits ``power_watts`` / ``temperature_c`` / +# ``gpu_clock_mhz`` / ``mem_clock_mhz``; the shorter spellings are older shapes, kept so archived reports still parse. +_POWER_KEYS = ("power_watts", "power_w", "power") +_TEMP_KEYS = ("temperature_c", "temp_c", "temperature") +_CLOCK_KEYS = ("gpu_clock_mhz", "clock_mhz", "sclk_mhz") +_MEM_CLOCK_KEYS = ("mem_clock_mhz", "memory_clock_mhz", "mclk_mhz") + +# Occupancy, in percent. ``gpu_util_pct`` / ``vram_pct`` are what the multi-node harvester writes (see +# ``benchmark_result._row_to_gpu_sample``); the rest are spellings of the same percentage a producer might use. +# +# Percent-named aliases only, deliberately. An absolute reading -- ``vram_used_mb``, ``memory_used_bytes`` -- is a +# different quantity, and folding one into a field called ``_pct`` would put 81920 where a percentage belongs. +_UTIL_KEYS = ("gpu_util_pct", "gpu_utilization_pct", "gpu_use_pct", "utilization_pct") +_VRAM_KEYS = ("vram_pct", "vram_usage_pct", "vram_used_pct", "memory_used_pct") + +#: Every metric this reader knows. A block counts as contributing when it yields any one of them. +_ALL_KEYS = (_POWER_KEYS, _TEMP_KEYS, _CLOCK_KEYS, _MEM_CLOCK_KEYS, _UTIL_KEYS, _VRAM_KEYS) + + +def _to_float(raw: Any) -> float | None: + """Coerce a reading to a float, or ``None`` when it is not one.""" + if isinstance(raw, bool) or not isinstance(raw, (int, float)): + return None + value = float(raw) + return value if value == value and value not in (float("inf"), float("-inf")) else None + + +#: Statistics this reader actually emits. A block that carries only the others has nothing to contribute. +_REPORTED_STATS = ("avg", "max") + + +def _source(block: dict[str, Any], keys: tuple[str, ...]) -> Any: + """Pick the one alias in this block that carries a reading this reader can report. + + Resolved once per block rather than once per statistic. Choosing per statistic lets the mean come from + ``power_watts`` and the peak from a stale ``power_w`` in the same block, which can report a maximum below the + average -- a self-contradictory number of exactly the kind this exists to stop emitting. + + Usable means usable *here*: only ``avg`` and ``max`` are emitted, so a block offering nothing but ``min`` measured + nothing this artifact can carry. Accepting it would credit its ``sample_count`` toward a row of nulls. + """ + for key in keys: + if key not in block: + continue + raw = block[key] + if isinstance(raw, dict): + if any(_to_float(raw.get(stat)) is not None for stat in _REPORTED_STATS): + return raw + elif _to_float(raw) is not None: + return raw + return None + + +def _metric(block: dict[str, Any], keys: tuple[str, ...], field: str) -> float | None: + """Read one metric out of a single ``gpu_monitor`` block. + + Both producer shapes are read here: a flat scalar per sample, where the scalar is that sample's mean and its peak + alike, and a pre-aggregated ``{min, max, avg}`` block, where ``field`` picks the statistic. ``None`` rather than + ``0.0`` when the winning alias omits this particular statistic -- a metric that was never sampled has to stay + distinguishable from one that measured zero. + """ + raw = _source(block, keys) + if raw is None: + return None + return _to_float(raw.get(field)) if isinstance(raw, dict) else _to_float(raw) + + +def _blocks(report: Any) -> list[dict[str, Any]]: + """Every ``gpu_monitor`` entry in a report, whichever shape it was written in.""" + if not isinstance(report, dict): + return [] + monitor = report.get("gpu_monitor") + if isinstance(monitor, list): + return [b for b in monitor if isinstance(b, dict)] + return [monitor] if isinstance(monitor, dict) else [] + + +def gpu_metrics_from_report(report: Any) -> dict[str, Any]: + """Normalise one round's GPU telemetry. ``{}`` when the report carried none. + + Means are weighted by each block's ``sample_count``: a Magpie block summarises many samples while a flat per-sample + block is one, and unweighted a 10-sample block would pull the round's mean as hard as a 10,000-sample one. + """ + blocks = _blocks(report) + if not blocks: + return {} + + def _weight(block: dict[str, Any]) -> float: + """Underlying samples behind one block; 1.0 when it does not say. + + ``or 1.0`` would promote a monitor that started and sampled nothing into one sample -- the same conflation of + "no reading" with "a reading of zero" this module exists to avoid. + """ + declared = _to_float(block.get("sample_count")) + return 1.0 if declared is None else max(0.0, declared) + + weights = [_weight(b) for b in blocks] + # Only blocks that yielded a metric count toward ``samples``. A block carrying ``sample_count: 27000`` and no + # recognised key measured nothing this reader can report, and crediting it would put a large sample count beside a + # row of nulls. + contributing = [w for b, w in zip(blocks, weights) if any(_source(b, keys) is not None for keys in _ALL_KEYS)] + + def _avg(keys: tuple[str, ...]) -> float | None: + """Sample-count-weighted mean of one metric, or ``None`` if unread.""" + total = 0.0 + weight_sum = 0.0 + for block, weight in zip(blocks, weights): + value = _metric(block, keys, "avg") + if value is None: + continue + total += value * weight + weight_sum += weight + return round(total / weight_sum, 2) if weight_sum else None + + def _max(keys: tuple[str, ...]) -> float | None: + """Peak of one metric across all blocks, or ``None`` if unread.""" + values = [v for b in blocks if (v := _metric(b, keys, "max")) is not None] + return round(max(values), 2) if values else None + + return { + "schema_version": 1, + "samples": round(sum(contributing)), + "blocks": len(blocks), + "avg_power_w": _avg(_POWER_KEYS), + "max_power_w": _max(_POWER_KEYS), + "avg_temp_c": _avg(_TEMP_KEYS), + "max_temp_c": _max(_TEMP_KEYS), + "avg_clock_mhz": _avg(_CLOCK_KEYS), + "max_clock_mhz": _max(_CLOCK_KEYS), + "avg_mem_clock_mhz": _avg(_MEM_CLOCK_KEYS), + # Compute occupancy and memory pressure. Only the multi-node harvester reports these today -- Magpie's monitor + # samples neither -- so a single-node round reports null for both. Null, not 0.0: "the GPU sat idle" and + # "nobody asked" lead to opposite conclusions about a round that looks slow. + "avg_gpu_util_pct": _avg(_UTIL_KEYS), + "max_gpu_util_pct": _max(_UTIL_KEYS), + "avg_vram_pct": _avg(_VRAM_KEYS), + "max_vram_pct": _max(_VRAM_KEYS), + } + + +def write_gpu_metrics_from_report(workspace: Path | str, report: Any, *, source: str) -> str | None: + """Write ``gpu_metrics.json`` from a report already in hand. Returns the path, or ``None``. + + This is the whole of the failure policy, and it draws one distinction the caller cannot draw for itself: a round + that carried no telemetry is an ordinary outcome and stays quiet, while a round that carried telemetry this failed + to write is a defect and says so. Reporting both at the same volume is how a broken writer goes unnoticed for as + long as this data did. + + It never raises. A round that produced a good benchmark number and no GPU artifact is a far better outcome than a + round failed by its own telemetry, so the one guarantee lives here rather than being repeated by each caller. + """ + root = Path(workspace) + try: + payload = gpu_metrics_from_report(report) + except Exception: # noqa: BLE001 - a malformed block must not fail the round + log.warning("gpu_metrics: could not normalise telemetry for %s", root, exc_info=True) + return None + if not payload: + return None + payload["source"] = source + out = root / GPU_ARTIFACT_NAME + try: + from hyperloom.common.io import atomic_write_json + + atomic_write_json(out, payload) + except Exception: # noqa: BLE001 - the round's measurement outranks its description + log.warning("gpu_metrics: telemetry was read but could not be written to %s", out, exc_info=True) + return None + return str(out) + + +def write_gpu_metrics(workspace: Path | str) -> str | None: + """Normalise the round's GPU telemetry into ``gpu_metrics.json``. Returns the path, or ``None``. + + A missing report is not a failure here: harvest runs before the report is guaranteed to have settled, so this is + called again once one is in hand. + """ + root = Path(workspace) + report_path = root / "benchmark_report.json" + if not report_path.is_file(): + return None + try: + report = json.loads(report_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + # A report still being written is the expected case at harvest time, not a fault worth warning about; the + # settled call that follows is what reports a genuinely unreadable one. + log.debug("gpu_metrics: %s is not readable yet", report_path, exc_info=True) + return None + return write_gpu_metrics_from_report(root, report, source=report_path.name) diff --git a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py b/src/hyperloom/orchestrator/actions/executors/_grid_runner.py index c7b4075e8f..52cdf71c0a 100644 --- a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py +++ b/src/hyperloom/orchestrator/actions/executors/_grid_runner.py @@ -63,6 +63,7 @@ select_run_workspace, snapshot_workspaces, ) +from ._gpu_metrics import write_gpu_metrics_from_report from .benchmark_backend import build_benchmark_command from ._inferencex_patcher import ( ensure_benchmark_lib_eval_start_patched, @@ -556,6 +557,11 @@ async def _settled_measurement( "the report was still being written when the subprocess was reaped", attempts, ) + # Harvest already tried this, but it runs before the report is guaranteed to exist: a report Magpie + # finishes writing after the subprocess is reaped would otherwise leave the round with no GPU artifact at + # all. Here the report is in hand, so write it from that rather than reading the file a second time. + if report is not None: + write_gpu_metrics_from_report(workspace, report, source="benchmark_report.json") return report, measurement await asyncio.sleep(max(0.01, float(poll_seconds))) diff --git a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py b/src/hyperloom/orchestrator/actions/executors/benchmark_result.py index 73ca9899c0..70d1f0bf08 100644 --- a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py +++ b/src/hyperloom/orchestrator/actions/executors/benchmark_result.py @@ -18,6 +18,8 @@ from hyperloom.common.coerce import first_float, first_int, to_float, to_int from hyperloom.common.jsonio import read_json +from ._gpu_metrics import write_gpu_metrics + log = logging.getLogger(__name__) @@ -259,6 +261,13 @@ def harvest_leaked_artifacts( harvest_mn_gpu_metrics(destination, subprocess_started_unix=subprocess_started_unix) except Exception as exc: log.warning("benchmark_result.harvest: MN GPU-metrics harvest failed: %s", exc) + # Whatever wrote the round's ``gpu_monitor`` block -- Magpie on one node, the harvest above on several -- normalise + # it into an artifact of its own now, while the round's own workspace is the subject. Aggregating it per session + # instead averaged baseline, explore and roofline rounds together and described none of them. + # Best effort, and only that: the report may still be settling, in which case there is nothing to read yet and the + # settled path writes it instead. ``write_gpu_metrics`` owns the guarantee that it never raises, so wrapping it + # again here would only add a second, unreachable handler over the one that reports what actually went wrong. + write_gpu_metrics(destination) return harvested