From b4c835a71fe5099b60031531a524b5d44728d6b8 Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 23 Jul 2026 16:04:57 -0700 Subject: [PATCH 1/3] feat(power): validate single-node GPU energy metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrate per-device telemetry over the formal benchmark window, enforce expected GPU topology, emit auditable validity artifacts, and add best-effort and strict power modes with CPU coverage. 中文：校验单节点 GPU 能耗指标。基于正式基准测试窗口对逐卡遥测数据进行积分，校验预期 GPU 拓扑，输出可审计的有效性产物，并为默认尽力模式和严格功耗模式补充 CPU 测试覆盖。 --- .github/workflows/benchmark-tmpl.yml | 18 + .github/workflows/e2e-tests.yml | 11 + .github/workflows/test-process-result.yml | 10 +- benchmarks/benchmark_lib.sh | 20 +- utils/aggregate_power.py | 658 +++++++++++++++++++--- utils/process_result.py | 123 +++- utils/test_aggregate_power.py | 423 +++++++++++++- utils/test_process_result.py | 309 +++++++++- 8 files changed, 1467 insertions(+), 105 deletions(-) diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml index 0e32953e8a..d9364ab04b 100644 --- a/.github/workflows/benchmark-tmpl.yml +++ b/.github/workflows/benchmark-tmpl.yml @@ -129,6 +129,11 @@ on: required: false type: string default: '3600' + require-power: + description: "Fail fixed-sequence result processing when GPU power is invalid" + required: false + type: boolean + default: false eval-limit: description: "Eval instance count: empty/full = whole split (default); N = first-N smoke slice" required: false @@ -174,6 +179,7 @@ env: KV_P2P_TRANSFER: ${{ inputs.kv-p2p-transfer }} TOTAL_CPU_DRAM_GB: ${{ inputs.total-cpu-dram-gb }} DURATION: ${{ inputs.duration }} + REQUIRE_POWER: ${{ inputs.require-power && '1' || '0' }} EVAL_LIMIT: ${{ inputs.eval-limit }} SWEBENCH_GEN_MODE: ${{ inputs.swebench-gen-mode }} AIPERF_FAILED_REQUEST_THRESHOLD: '0.10' @@ -353,6 +359,18 @@ jobs: path: gpu_metrics.csv if-no-files-found: ignore + - name: Upload power audit bundle + if: ${{ always() && !inputs.eval-only && inputs.scenario-type != 'agentic-coding' }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: power_audit_${{ env.RESULT_FILENAME }} + path: | + ${{ env.RESULT_FILENAME }}.json + agg_${{ env.RESULT_FILENAME }}.json + gpu_metrics.csv + power_validation_${{ env.RESULT_FILENAME }}.json + if-no-files-found: ignore + - name: Upload eval results (if any) if: ${{ always() && (env.RUN_EVAL == 'true' || inputs.eval-only) }} uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 8ab245ae26..a436693927 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -21,6 +21,11 @@ on: required: false type: string default: "" + require-power: + description: "Fail single-node fixed-sequence jobs when GPU power is invalid" + required: false + type: boolean + default: false eval-limit: description: "Eval instance count: empty/full = whole split (default); N = first-N smoke slice" required: false @@ -50,6 +55,11 @@ on: required: false type: string default: "" + require-power: + description: "Fail single-node fixed-sequence jobs when GPU power is invalid" + required: false + type: boolean + default: false eval-limit: description: "Eval instance count: empty/full = whole split (default); N = first-N smoke slice" required: false @@ -404,6 +414,7 @@ jobs: spec-decoding: ${{ matrix.config.spec-decoding }} disagg: ${{ matrix.config.disagg }} run-eval: false + require-power: ${{ inputs.require-power }} ref: ${{ inputs.ref }} test-sweep-evals: diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml index 833e602528..d5579238b8 100644 --- a/.github/workflows/test-process-result.yml +++ b/.github/workflows/test-process-result.yml @@ -3,7 +3,14 @@ name: Test Process Result on: pull_request: paths: + - '.github/workflows/benchmark-tmpl.yml' + - '.github/workflows/e2e-tests.yml' + - '.github/workflows/test-process-result.yml' + - 'benchmarks/benchmark_lib.sh' + - 'utils/aggregate_power.py' + - 'utils/bench_serving/benchmark_serving.py' - 'utils/process_result.py' + - 'utils/test_aggregate_power.py' - 'utils/test_process_result.py' permissions: @@ -11,7 +18,6 @@ permissions: jobs: test: - if: github.event.pull_request.draft != true runs-on: ubuntu-latest permissions: contents: read @@ -33,4 +39,4 @@ jobs: - name: Run pytest run: | cd utils - pytest test_process_result.py -v + python -m pytest test_aggregate_power.py test_process_result.py -v diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index bda03e8b6e..efb55cb967 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -109,7 +109,9 @@ unset _benchmark_caller # -------------------------------- GPU_MONITOR_PID="" +GPU_MONITOR_VENDOR="" GPU_METRICS_CSV="/workspace/gpu_metrics.csv" +NVIDIA_GPU_MONITOR_QUERY="timestamp,index,power.draw,temperature.gpu,clocks.current.sm,clocks.current.memory,utilization.gpu,utilization.memory" export GPU_METRICS_CSV # Start background GPU monitoring that logs metrics every second to CSV. @@ -131,11 +133,13 @@ start_gpu_monitor() { export GPU_METRICS_CSV if command -v nvidia-smi &>/dev/null; then - nvidia-smi --query-gpu=timestamp,index,power.draw,temperature.gpu,clocks.current.sm,clocks.current.memory,utilization.gpu,utilization.memory \ + GPU_MONITOR_VENDOR="nvidia" + nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ --format=csv -l "$interval" > "$output" 2>/dev/null & GPU_MONITOR_PID=$! echo "[GPU Monitor] Started NVIDIA (PID=$GPU_MONITOR_PID, interval=${interval}s, output=$output)" elif command -v amd-smi &>/dev/null; then + GPU_MONITOR_VENDOR="amd" # Use amd-smi native watch mode (-w) which includes timestamps automatically. # Pipe through awk to: skip preamble lines, keep first CSV header, skip repeated headers. amd-smi metric -p -c -t -u -w "$interval" --csv 2>/dev/null \ @@ -143,6 +147,7 @@ start_gpu_monitor() { GPU_MONITOR_PID=$! echo "[GPU Monitor] Started AMD (PID=$GPU_MONITOR_PID, interval=${interval}s, output=$output)" else + GPU_MONITOR_VENDOR="" echo "[GPU Monitor] No GPU monitoring tool found (nvidia-smi or amd-smi), skipping" return 0 fi @@ -153,6 +158,18 @@ stop_gpu_monitor() { if [[ -n "$GPU_MONITOR_PID" ]] && kill -0 "$GPU_MONITOR_PID" 2>/dev/null; then kill "$GPU_MONITOR_PID" 2>/dev/null wait "$GPU_MONITOR_PID" 2>/dev/null || true + # benchmark_end_time_unix is recorded shortly before the benchmark + # process exits. For the NVIDIA PR1 canary, append one post-exit sample + # so boundary interpolation is deterministic even when the run ends + # between 1 Hz monitor ticks. AMD one-shot CSV schemas vary by amd-smi + # version, so strict AMD lifecycle validation remains follow-up work. + case "$GPU_MONITOR_VENDOR" in + nvidia) + nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ + --format=csv,noheader >> "$GPU_METRICS_CSV" 2>/dev/null || + echo "[GPU Monitor] Warning: final NVIDIA sample failed" >&2 + ;; + esac echo "[GPU Monitor] Stopped (PID=$GPU_MONITOR_PID)" if [[ -f "$GPU_METRICS_CSV" ]]; then local lines @@ -161,6 +178,7 @@ stop_gpu_monitor() { fi fi GPU_MONITOR_PID="" + GPU_MONITOR_VENDOR="" } # Return success only while a PID exists and is not a zombie waiting to be diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index 3c204085a2..71aa9daca4 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -1,44 +1,87 @@ -"""Aggregate measured GPU power from a vendor SMI CSV into the agg result JSON. - -Reads a GPU-metrics CSV produced by `start_gpu_monitor` (nvidia-smi or amd-smi), -filters samples to the benchmark load window using start/end Unix timestamps -written by benchmark_serving.py, and patches two keys into the aggregated -result JSON consumed by InferenceX-app's ETL: - - - avg_power_w: mean per-GPU power draw (W) during the load window - - joules_per_output_token: (avg_power_w * num_gpus * duration_s) / total_output_tokens - -The ETL (`packages/db/src/etl/benchmark-mapper.ts`) auto-captures any numeric -field in the agg JSON into the `metrics` JSONB column, so no schema migration -is required. - -Vendor schema detection is regex-based: any timestamp-like column + any column -whose name contains "power" (excluding "limit"/"cap"/"max") is picked up. -NVIDIA emits "power.draw [W]"; AMD's amd-smi varies by version. Both are -handled. - -This script is best-effort. Missing or malformed CSV exits 0 without patching -so a monitoring hiccup never breaks the benchmark upload. +"""Validate and aggregate single-node GPU-board power into benchmark results. + +The formal benchmark window comes from ``benchmark_serving.py``. Power is +integrated independently for every observed GPU with trapezoidal integration +and linear interpolation at the window boundaries. Unqualified energy metrics +therefore always describe the whole deployment, never a prefill/decode role. + +Ordinary benchmark runs are best-effort: invalid telemetry is recorded in the +aggregate and a validation sidecar, but does not fail the benchmark. Power +studies can set ``REQUIRE_POWER=1`` to fail after those audit artifacts exist. +The aggregate carries numeric ``power_valid`` (1/0) for metric ingestion; the +sidecar is the canonical source for boolean validity and reason codes. """ from __future__ import annotations import argparse +import bisect import csv import json +import math +import os import re import sys +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from statistics import mean - _POWER_COL_RE = re.compile(r"power", re.IGNORECASE) _POWER_EXCLUDE_RE = re.compile(r"limit|cap|max|min", re.IGNORECASE) _TIMESTAMP_COL_RE = re.compile(r"time", re.IGNORECASE) _GPU_INDEX_COL_RE = re.compile(r"^(index|gpu|gpu_id|gpu_index|card|device)$", re.IGNORECASE) _NUMBER_RE = re.compile(r"-?\d+(?:\.\d+)?") +_INTEGRATION_METHOD = "per_device_trapezoidal_with_linear_boundary_interpolation" +_DEFAULT_MAX_SAMPLE_GAP_S = 3.0 +_POWER_METRIC_KEYS = { + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", +} + + +@dataclass(frozen=True) +class PowerIntegration: + """Auditable result of integrating one single-node telemetry stream.""" + + power_valid: bool + invalid_reasons: tuple[str, ...] + expected_num_gpus: int | None + observed_gpu_ids: tuple[str, ...] + per_gpu_sample_counts: dict[str, int] + per_gpu_max_sample_gap_s: dict[str, float] + per_gpu_energy_j: dict[str, float] + device_issues: dict[str, list[str]] + avg_power_w: float | None = None + avg_total_gpu_power_w: float | None = None + total_gpu_energy_j: float | None = None + + @property + def observed_num_gpus(self) -> int: + return len(self.observed_gpu_ids) + + +@dataclass(frozen=True) +class BenchmarkData: + """Raw benchmark fields required for energy normalization.""" + + start_unix: float + end_unix: float + reported_duration_s: float + completed: int + total_input_tokens: int + total_output_tokens: int + + @property + def integration_duration_s(self) -> float: + return self.end_unix - self.start_unix + def _parse_timestamp(value: str) -> float | None: """Best-effort timestamp parse to Unix epoch seconds (local wall clock). @@ -113,10 +156,10 @@ def aggregate_power( start_unix: float, end_unix: float, ) -> tuple[float, int] | None: - """Return (per_gpu_avg_power_w, num_gpus) for samples in [start, end]. + """Legacy arithmetic mean retained for callers of the pre-PR1 helper. - Returns None if the CSV is missing, empty, has no detectable power column, - or no rows fall in the window. + Published PR1 metrics use :func:`integrate_power`; this compatibility API + must not be used for energy calculations. """ if not csv_path.is_file() or csv_path.stat().st_size == 0: return None @@ -190,6 +233,267 @@ def aggregate_power( return mean(per_sample_mean_per_gpu), num_gpus +def _append_reason(reasons: list[str], reason: str) -> None: + if reason not in reasons: + reasons.append(reason) + + +def _gpu_sort_key(gpu_id: str) -> tuple[int, int | str]: + return (0, int(gpu_id)) if gpu_id.isdigit() else (1, gpu_id) + + +def _empty_integration( + *, + expected_num_gpus: int | None, + reasons: list[str], +) -> PowerIntegration: + return PowerIntegration( + power_valid=False, + invalid_reasons=tuple(reasons), + expected_num_gpus=expected_num_gpus, + observed_gpu_ids=(), + per_gpu_sample_counts={}, + per_gpu_max_sample_gap_s={}, + per_gpu_energy_j={}, + device_issues={}, + ) + + +def _interpolate_power(samples: list[tuple[float, float]], timestamp: float) -> float: + """Linearly interpolate power at a timestamp bracketed by ``samples``.""" + times = [sample_time for sample_time, _ in samples] + right_index = bisect.bisect_left(times, timestamp) + if right_index < len(samples) and math.isclose( + samples[right_index][0], timestamp, rel_tol=0.0, abs_tol=1e-9 + ): + return samples[right_index][1] + left_index = right_index - 1 + if left_index < 0 or right_index >= len(samples): + raise ValueError("timestamp is not bracketed") + left_time, left_power = samples[left_index] + right_time, right_power = samples[right_index] + fraction = (timestamp - left_time) / (right_time - left_time) + return left_power + fraction * (right_power - left_power) + + +def _integrate_device( + samples: list[tuple[float, float]], + *, + start_unix: float, + end_unix: float, +) -> float: + """Integrate one device over ``[start_unix, end_unix]``.""" + start_power = _interpolate_power(samples, start_unix) + end_power = _interpolate_power(samples, end_unix) + clipped = [(start_unix, start_power)] + clipped.extend( + (timestamp, power) + for timestamp, power in samples + if start_unix < timestamp < end_unix + ) + clipped.append((end_unix, end_power)) + + energy_j = 0.0 + for (left_time, left_power), (right_time, right_power) in zip( + clipped, clipped[1:] + ): + energy_j += (right_time - left_time) * (left_power + right_power) / 2.0 + return energy_j + + +def integrate_power( + csv_path: Path, + *, + start_unix: float, + end_unix: float, + expected_num_gpus: int | None = None, + max_sample_gap_s: float = _DEFAULT_MAX_SAMPLE_GAP_S, +) -> PowerIntegration: + """Validate and integrate per-device GPU power over the formal window. + + A valid stream must expose stable GPU identities, match the expected + topology when supplied, bracket both window boundaries for every device, + and have no in-window sampling gap larger than ``max_sample_gap_s``. + """ + reasons: list[str] = [] + if ( + not math.isfinite(start_unix) + or not math.isfinite(end_unix) + or end_unix <= start_unix + ): + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=["invalid_benchmark_window"], + ) + if expected_num_gpus is not None and expected_num_gpus <= 0: + _append_reason(reasons, "invalid_expected_gpu_count") + if not math.isfinite(max_sample_gap_s) or max_sample_gap_s <= 0: + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=["invalid_max_sample_gap"], + ) + + if not csv_path.is_file(): + _append_reason(reasons, "telemetry_file_missing") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + try: + if csv_path.stat().st_size == 0: + _append_reason(reasons, "telemetry_file_empty") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + except OSError: + _append_reason(reasons, "telemetry_file_unreadable") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + # Values are first grouped by GPU and exact timestamp. Some SMI versions + # expose timestamps at lower resolution than their sampling cadence, so + # duplicate-timestamp readings are averaged rather than treated as corrupt. + raw_samples: dict[str, dict[float, list[float]]] = {} + saw_missing_gpu_identity = False + try: + with csv_path.open("r", newline="", encoding="utf-8", errors="replace") as f: + reader = csv.DictReader(f, skipinitialspace=True) + header = [column.strip() for column in (reader.fieldnames or [])] + reader.fieldnames = header + timestamp_col, power_col, gpu_col = _detect_columns(header) + missing_required_columns = False + if not timestamp_col: + _append_reason(reasons, "timestamp_column_missing") + missing_required_columns = True + if not power_col: + _append_reason(reasons, "power_column_missing") + missing_required_columns = True + if not gpu_col: + _append_reason(reasons, "gpu_identity_column_missing") + missing_required_columns = True + if missing_required_columns: + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + for row in reader: + timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip()) + if timestamp is None or not math.isfinite(timestamp): + continue + # Only the formal window and its possible boundary neighbors + # can affect integration. Bad warmup/eval rows farther away + # must not invalidate an otherwise sound measurement. + if ( + timestamp < start_unix - max_sample_gap_s + or timestamp > end_unix + max_sample_gap_s + ): + continue + + power = _parse_power((row.get(power_col) or "").strip()) + gpu_id = (row.get(gpu_col) or "").strip() + if power is None: + continue + if not math.isfinite(power) or power < 0: + _append_reason(reasons, "invalid_power_sample") + continue + if not gpu_id: + saw_missing_gpu_identity = True + continue + values = raw_samples.setdefault(gpu_id, {}).setdefault(timestamp, []) + values.append(power) + except (OSError, csv.Error): + _append_reason(reasons, "telemetry_file_unreadable") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + if saw_missing_gpu_identity: + _append_reason(reasons, "gpu_identity_missing") + if not raw_samples: + _append_reason(reasons, "no_usable_power_samples") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + observed_gpu_ids = tuple(sorted(raw_samples, key=_gpu_sort_key)) + if ( + expected_num_gpus is not None + and expected_num_gpus > 0 + and len(observed_gpu_ids) != expected_num_gpus + ): + _append_reason(reasons, "expected_gpu_count_mismatch") + + per_gpu_sample_counts: dict[str, int] = {} + per_gpu_max_sample_gap_s: dict[str, float] = {} + per_gpu_energy_j: dict[str, float] = {} + device_issues: dict[str, list[str]] = {} + + for gpu_id in observed_gpu_ids: + timestamp_values = raw_samples[gpu_id] + samples = sorted( + (timestamp, mean(values)) for timestamp, values in timestamp_values.items() + ) + per_gpu_sample_counts[gpu_id] = len(samples) + issues: list[str] = [] + + if len(samples) < 2: + issues.append("insufficient_power_samples") + _append_reason(reasons, "insufficient_power_samples") + elif samples[0][0] > start_unix or samples[-1][0] < end_unix: + issues.append("benchmark_window_not_bracketed") + _append_reason(reasons, "benchmark_window_not_bracketed") + else: + times = [timestamp for timestamp, _ in samples] + left_index = bisect.bisect_right(times, start_unix) - 1 + right_index = bisect.bisect_left(times, end_unix) + relevant = samples[left_index : right_index + 1] + gaps = [ + right[0] - left[0] for left, right in zip(relevant, relevant[1:]) + ] + max_gap = max(gaps, default=0.0) + per_gpu_max_sample_gap_s[gpu_id] = max_gap + if max_gap > max_sample_gap_s: + issues.append("sampling_gap_exceeded") + _append_reason(reasons, "sampling_gap_exceeded") + per_gpu_energy_j[gpu_id] = _integrate_device( + samples, + start_unix=start_unix, + end_unix=end_unix, + ) + + if issues: + device_issues[gpu_id] = issues + + duration_s = end_unix - start_unix + total_gpu_energy_j: float | None = None + avg_total_gpu_power_w: float | None = None + avg_power_w: float | None = None + if len(per_gpu_energy_j) == len(observed_gpu_ids): + total_gpu_energy_j = sum(per_gpu_energy_j.values()) + avg_total_gpu_power_w = total_gpu_energy_j / duration_s + avg_power_w = avg_total_gpu_power_w / len(observed_gpu_ids) + + return PowerIntegration( + power_valid=not reasons, + invalid_reasons=tuple(reasons), + expected_num_gpus=expected_num_gpus, + observed_gpu_ids=observed_gpu_ids, + per_gpu_sample_counts=per_gpu_sample_counts, + per_gpu_max_sample_gap_s=per_gpu_max_sample_gap_s, + per_gpu_energy_j=per_gpu_energy_j, + device_issues=device_issues, + avg_power_w=avg_power_w, + avg_total_gpu_power_w=avg_total_gpu_power_w, + total_gpu_energy_j=total_gpu_energy_j, + ) + + def _load_bench_window( bench_result_path: Path, ) -> tuple[float, float, float, int, int] | None: @@ -218,6 +522,69 @@ def _load_bench_window( return float(start), float(end), float(duration), int(total_output), int(total_input) +def _load_benchmark_data( + bench_result_path: Path, +) -> tuple[BenchmarkData | None, list[str]]: + """Load the strict energy-normalization contract from raw benchmark JSON.""" + try: + bench = json.loads(bench_result_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None, ["invalid_benchmark_result"] + + start = bench.get("benchmark_start_time_unix") + end = bench.get("benchmark_end_time_unix") + duration = bench.get("duration") + numeric_window = all( + isinstance(value, (int, float)) and not isinstance(value, bool) + for value in (start, end, duration) + ) + if not numeric_window: + return None, ["invalid_benchmark_window"] + + start = float(start) + end = float(end) + duration = float(duration) + if ( + not all(math.isfinite(value) for value in (start, end, duration)) + or end <= start + or duration <= 0 + ): + return None, ["invalid_benchmark_window"] + + reasons: list[str] = [] + integration_duration = end - start + duration_tolerance = max(0.5, integration_duration * 0.01) + if abs(duration - integration_duration) > duration_tolerance: + _append_reason(reasons, "benchmark_duration_mismatch") + + completed = bench.get("completed") + if not isinstance(completed, int) or isinstance(completed, bool) or completed <= 0: + _append_reason(reasons, "invalid_successful_query_count") + completed = 0 + + total_input = bench.get("total_input_tokens") + if not isinstance(total_input, int) or isinstance(total_input, bool) or total_input <= 0: + _append_reason(reasons, "invalid_input_token_count") + total_input = 0 + + total_output = bench.get("total_output_tokens") + if not isinstance(total_output, int) or isinstance(total_output, bool) or total_output <= 0: + _append_reason(reasons, "invalid_output_token_count") + total_output = 0 + + return ( + BenchmarkData( + start_unix=start, + end_unix=end, + reported_duration_s=duration, + completed=completed, + total_input_tokens=total_input, + total_output_tokens=total_output, + ), + reasons, + ) + + def patch_agg_result( agg_path: Path, avg_power_w: float, @@ -234,57 +601,192 @@ def patch_agg_result( tmp_path.replace(agg_path) -def run(csv_path: Path, bench_result: Path, agg_result: Path) -> int: - window = _load_bench_window(bench_result) - if window is None: - print( - f"[aggregate_power] No bench window in {bench_result} — skipping power aggregation", - file=sys.stderr, +def _write_json_atomic(path: Path, data: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + tmp_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + tmp_path.replace(path) + + +def _derived_metrics( + integration: PowerIntegration, + benchmark: BenchmarkData, +) -> dict[str, float]: + """Return whole-deployment energy metrics for a valid measurement.""" + if ( + integration.avg_power_w is None + or integration.avg_total_gpu_power_w is None + or integration.total_gpu_energy_j is None + ): + raise ValueError("valid power integration has incomplete power metrics") + avg_power_w = integration.avg_power_w + avg_total_gpu_power_w = integration.avg_total_gpu_power_w + energy = integration.total_gpu_energy_j + total_tokens = benchmark.total_input_tokens + benchmark.total_output_tokens + return { + "avg_power_w": avg_power_w, + "avg_total_gpu_power_w": avg_total_gpu_power_w, + "total_gpu_energy_j": energy, + "joules_per_successful_query": energy / benchmark.completed, + "joules_per_input_token": energy / benchmark.total_input_tokens, + "joules_per_output_token": energy / benchmark.total_output_tokens, + "joules_per_total_token": energy / total_tokens, + } + + +def _patch_power_result( + agg_path: Path, + *, + power_valid: bool, + metrics: dict[str, float], +) -> None: + data = json.loads(agg_path.read_text(encoding="utf-8")) + for key in _POWER_METRIC_KEYS: + data.pop(key, None) + # Keep the canonical aggregate numeric-only for InferenceX-app's metric + # auto-capture. Detailed reason codes live in the validation sidecar. + data["power_valid"] = int(power_valid) + data.pop("power_invalid_reasons", None) + if power_valid: + for key, value in metrics.items(): + if value is None or not math.isfinite(value): + raise ValueError(f"non-finite power metric: {key}") + precision = 3 if key.endswith(("_w", "_j")) else 6 + data[key] = round(value, precision) + _write_json_atomic(agg_path, data) + + +def _validation_payload( + *, + csv_path: Path, + bench_result: Path, + benchmark: BenchmarkData | None, + integration: PowerIntegration, + power_valid: bool, + reasons: list[str], + metrics: dict[str, float], +) -> dict: + benchmark_window = None + if benchmark is not None: + benchmark_window = { + "start_time_unix": benchmark.start_unix, + "end_time_unix": benchmark.end_unix, + "reported_duration_s": benchmark.reported_duration_s, + "integration_duration_s": benchmark.integration_duration_s, + "completed": benchmark.completed, + "total_input_tokens": benchmark.total_input_tokens, + "total_output_tokens": benchmark.total_output_tokens, + } + return { + "schema_version": 1, + "power_valid": power_valid, + "reasons": reasons, + "telemetry_source": str(csv_path), + "benchmark_result": str(bench_result), + "benchmark_window": benchmark_window, + "integration_method": _INTEGRATION_METHOD, + "expected_gpu_count": integration.expected_num_gpus, + "observed_gpu_count": integration.observed_num_gpus, + "observed_gpu_ids": list(integration.observed_gpu_ids), + "per_gpu_sample_counts": integration.per_gpu_sample_counts, + "per_gpu_max_sample_gap_s": integration.per_gpu_max_sample_gap_s, + "per_gpu_energy_j": integration.per_gpu_energy_j, + "device_issues": integration.device_issues, + "metrics": { + key: round(value, 6) + for key, value in metrics.items() + if value is not None and math.isfinite(value) + }, + } + + +def run( + csv_path: Path, + bench_result: Path, + agg_result: Path, + *, + expected_num_gpus: int | None = None, + validation_result: Path | None = None, + require_power: bool = False, +) -> int: + """Aggregate power, always preserving validity, and optionally fail closed.""" + validation_result = validation_result or bench_result.with_name( + f"power_validation_{bench_result.stem}.json" + ) + benchmark, benchmark_reasons = _load_benchmark_data(bench_result) + if benchmark is None: + integration = _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=benchmark_reasons.copy(), ) - return 0 - start, end, duration, total_output, total_input = window - - result = aggregate_power(csv_path, start, end) - if result is None: - print( - f"[aggregate_power] No usable power samples in {csv_path} for " - f"window [{start}, {end}] — skipping", - file=sys.stderr, + else: + integration = integrate_power( + csv_path, + start_unix=benchmark.start_unix, + end_unix=benchmark.end_unix, + expected_num_gpus=expected_num_gpus, ) - return 0 - avg_power_w, num_gpus = result - - # Joules consumed by the system during the bench window, divided by either - # output tokens (for generation-cost metrics) or all tokens (for whole- - # workload efficiency). - total_system_energy_j = avg_power_w * num_gpus * duration - joules_per_output_token = total_system_energy_j / total_output - total_tokens = total_output + total_input - joules_per_total_token = ( - total_system_energy_j / total_tokens if total_tokens > 0 else joules_per_output_token - ) + + reasons = benchmark_reasons.copy() + for reason in integration.invalid_reasons: + _append_reason(reasons, reason) + + metrics: dict[str, float] = {} + power_valid = not reasons + if power_valid and benchmark is not None: + metrics = _derived_metrics(integration, benchmark) if not agg_result.is_file(): + _append_reason(reasons, "aggregate_result_missing") + power_valid = False + metrics = {} + else: + try: + _patch_power_result( + agg_result, + power_valid=power_valid, + metrics=metrics, + ) + except (OSError, json.JSONDecodeError, ValueError) as exc: + _append_reason(reasons, "aggregate_result_unwritable") + power_valid = False + metrics = {} + print(f"[aggregate_power] Failed to patch {agg_result}: {exc}", file=sys.stderr) + + try: + _write_json_atomic( + validation_result, + _validation_payload( + csv_path=csv_path, + bench_result=bench_result, + benchmark=benchmark, + integration=integration, + power_valid=power_valid, + reasons=reasons, + metrics=metrics, + ), + ) + except OSError as exc: print( - f"[aggregate_power] Agg result {agg_result} missing — cannot patch", + f"[aggregate_power] Failed to write validation artifact " + f"{validation_result}: {exc}", file=sys.stderr, ) - return 0 + return 1 if require_power else 0 - try: - patch_agg_result( - agg_result, avg_power_w, joules_per_output_token, joules_per_total_token + if not power_valid: + print( + f"[aggregate_power] Power validation failed: {', '.join(reasons)} " + f"(details: {validation_result})", + file=sys.stderr, ) - except (OSError, json.JSONDecodeError) as exc: - print(f"[aggregate_power] Failed to patch {agg_result}: {exc}", file=sys.stderr) - return 0 + return 1 if require_power else 0 print( - f"[aggregate_power] avg_power_w={avg_power_w:.2f} (per GPU, n={num_gpus}) " - f"joules_per_output_token={joules_per_output_token:.4f} " - f"joules_per_total_token={joules_per_total_token:.4f} " - f"duration={duration:.1f}s output_tokens={total_output} input_tokens={total_input} " - f"-> {agg_result}" + f"[aggregate_power] avg_power_w={metrics['avg_power_w']:.2f} " + f"avg_total_gpu_power_w={metrics['avg_total_gpu_power_w']:.2f} " + f"total_gpu_energy_j={metrics['total_gpu_energy_j']:.2f} " + f"(n={integration.observed_num_gpus}) -> {agg_result}" ) return 0 @@ -309,8 +811,32 @@ def main() -> int: required=True, help="Path to the agg_.json output of process_result.py (will be patched in place)", ) + parser.add_argument( + "--expected-num-gpus", + type=int, + required=True, + help="Expected single-node GPU count from TP * PP * PCP", + ) + parser.add_argument( + "--validation-result", + type=Path, + help="Path for the power validation sidecar", + ) + parser.add_argument( + "--require-power", + action="store_true", + default=os.environ.get("REQUIRE_POWER", "").lower() in {"1", "true", "yes"}, + help="Fail when power telemetry is invalid (also enabled by REQUIRE_POWER=1)", + ) args = parser.parse_args() - return run(args.csv, args.bench_result, args.agg_result) + return run( + args.csv, + args.bench_result, + args.agg_result, + expected_num_gpus=args.expected_num_gpus, + validation_result=args.validation_result, + require_power=args.require_power, + ) if __name__ == "__main__": diff --git a/utils/process_result.py b/utils/process_result.py index bf3bcdaba1..f7759dd004 100644 --- a/utils/process_result.py +++ b/utils/process_result.py @@ -1,6 +1,6 @@ -import sys import json import os +import sys from pathlib import Path @@ -40,6 +40,73 @@ def get_optional_component_metadata(env_var): return metadata +def record_power_internal_error( + *, + csv_path, + bench_result, + agg_result, + validation_result, + expected_num_gpus, + error, +): + """Preserve an auditable invalid result when aggregation fails unexpectedly.""" + reason = "aggregation_internal_error" + try: + agg_data = json.loads(agg_result.read_text(encoding="utf-8")) + for key in ( + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", + ): + agg_data.pop(key, None) + agg_data["power_valid"] = 0 + agg_data.pop("power_invalid_reasons", None) + agg_tmp = agg_result.with_suffix(agg_result.suffix + ".tmp") + agg_tmp.write_text(json.dumps(agg_data, indent=2), encoding="utf-8") + agg_tmp.replace(agg_result) + + validation_data = { + "schema_version": 1, + "power_valid": False, + "reasons": [reason], + "telemetry_source": str(csv_path), + "benchmark_result": str(bench_result), + "benchmark_window": None, + "integration_method": ( + "per_device_trapezoidal_with_linear_boundary_interpolation" + ), + "expected_gpu_count": expected_num_gpus, + "observed_gpu_count": 0, + "observed_gpu_ids": [], + "per_gpu_sample_counts": {}, + "per_gpu_max_sample_gap_s": {}, + "per_gpu_energy_j": {}, + "device_issues": {}, + "metrics": {}, + "internal_error": { + "type": type(error).__name__, + "message": str(error)[:500], + }, + } + validation_tmp = validation_result.with_suffix( + validation_result.suffix + ".tmp" + ) + validation_tmp.write_text( + json.dumps(validation_data, indent=2), encoding="utf-8" + ) + validation_tmp.replace(validation_result) + except (OSError, json.JSONDecodeError) as fallback_error: + print( + f"[process_result] failed to preserve power validation fallback: " + f"{fallback_error}", + file=sys.stderr, + ) + + # Base required env vars base_env = get_required_env_vars([ 'RUNNER_TYPE', 'FRAMEWORK', 'PRECISION', 'SPEC_DECODING', @@ -204,16 +271,23 @@ def get_optional_component_metadata(env_var): data[key.replace('_ms', '').replace( 'tpot', 'intvty')] = 1000.0 / float(value) -print(json.dumps(data, indent=2)) - agg_path = Path(f'agg_{result_filename}.json') with open(agg_path, 'w') as f: json.dump(data, f, indent=2) -# Best-effort: patch measured power into the agg JSON. Never fails the run. -try: - from aggregate_power import run as _aggregate_power_run - +# Single-node measured power is best-effort by default. Power studies can set +# REQUIRE_POWER=1 to fail closed after the validation sidecar has been written. +_require_power = os.environ.get('REQUIRE_POWER', '').lower() in {'1', 'true', 'yes'} +_power_status = 0 +if is_multinode: + if _require_power: + print( + '[process_result] Power validation failed: PR1 supports only ' + 'single-node non-disaggregated telemetry', + file=sys.stderr, + ) + _power_status = 1 +else: _csv_candidates = [ os.environ.get('GPU_METRICS_CSV'), 'gpu_metrics.csv', @@ -221,13 +295,36 @@ def get_optional_component_metadata(env_var): ] _csv_path = next( (Path(p) for p in _csv_candidates if p and Path(p).is_file()), - None, + Path(next(p for p in _csv_candidates if p)), ) - if _csv_path is not None: - _aggregate_power_run( + _bench_path = Path(f'{result_filename}.json') + _validation_path = Path(f'power_validation_{result_filename}.json') + try: + from aggregate_power import run as _aggregate_power_run + + _power_status = _aggregate_power_run( csv_path=_csv_path, - bench_result=Path(f'{result_filename}.json'), + bench_result=_bench_path, agg_result=agg_path, + expected_num_gpus=num_gpus, + validation_result=_validation_path, + require_power=_require_power, ) -except Exception as exc: # noqa: BLE001 — never block on telemetry - print(f'[process_result] power aggregation skipped: {exc}', file=sys.stderr) + except Exception as exc: # noqa: BLE001 — preserve ordinary benchmark behavior + print(f'[process_result] power aggregation failed: {exc}', file=sys.stderr) + record_power_internal_error( + csv_path=_csv_path, + bench_result=_bench_path, + agg_result=agg_path, + validation_result=_validation_path, + expected_num_gpus=num_gpus, + error=exc, + ) + if _require_power: + _power_status = 1 + +with open(agg_path) as f: + print(json.dumps(json.load(f), indent=2)) + +if _power_status: + raise SystemExit(_power_status) diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index bf81ee7b16..8a6b160281 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -3,13 +3,13 @@ Covers: - NVIDIA CSV (nvidia-smi --query-gpu format with "X W" power cells) - AMD CSV (amd-smi --csv with ISO/epoch timestamps and bare numeric power) - - Window filtering (samples outside [start, end] are excluded) - - Multi-GPU per-sample aggregation (sum across GPUs at each timestamp, - then mean over samples — yields per-GPU mean) + - Backward-compatible arithmetic aggregation + - Per-device trapezoidal integration over the formal benchmark window + - Expected/observed GPU validation, window coverage, and sampling gaps - Missing / empty / malformed CSV: returns None, no exception - - End-to-end run(): patches agg JSON with avg_power_w + joules_per_output_token - + joules_per_total_token - - Missing bench window keys: skips gracefully without patching + - Whole-deployment power/energy/J-query/J-token metric semantics + - Best-effort invalid results and REQUIRE_POWER strict failures + - Atomic aggregate and validation artifacts """ from __future__ import annotations @@ -27,6 +27,8 @@ _parse_power, _parse_timestamp, aggregate_power, + integrate_power, + main, patch_agg_result, run, ) @@ -292,6 +294,7 @@ def _write_bench_result( duration: float, total_output: int, total_input: int = 0, + completed: int = 1, ) -> None: path.write_text( json.dumps( @@ -299,6 +302,7 @@ def _write_bench_result( "benchmark_start_time_unix": start, "benchmark_end_time_unix": end, "duration": duration, + "completed": completed, "total_output_tokens": total_output, "total_input_tokens": total_input, } @@ -307,20 +311,205 @@ def _write_bench_result( ) -def test_run_patches_agg_with_power_and_joules(tmp_path: Path): +def _write_constant_window_samples( + path: Path, + *, + start: float, + end: float, + watts_per_gpu: float, + num_gpus: int, +) -> None: + """Write samples that bracket the formal benchmark window.""" + duration = int(end - start) + timestamps = [start - 1.0] + timestamps.extend(start + offset for offset in range(duration + 1)) + timestamps.append(end + 1.0) + _write_nvidia_csv( + path, + [ + (timestamp, gpu, watts_per_gpu) + for timestamp in timestamps + for gpu in range(num_gpus) + ], + ) + + +# --------------------------------------------------------------------------- # +# Validated per-device integration contract +# --------------------------------------------------------------------------- # + + +def test_integrate_power_uses_per_device_trapezoidal_integration(tmp_path: Path): + """Irregular samples are integrated in time per GPU, not averaged by row.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + _write_nvidia_csv( + csv, + [ + # Both devices are linear in time. Boundary values at t=0 and + # t=10 must be interpolated from samples outside the window. + (base - 1, 0, 100.0), + (base + 1, 0, 140.0), + (base + 3.5, 0, 190.0), + (base + 6, 0, 240.0), + (base + 8.5, 0, 290.0), + (base + 11, 0, 340.0), + (base - 1, 1, 200.0), + (base + 1, 1, 240.0), + (base + 3.5, 1, 290.0), + (base + 6, 1, 340.0), + (base + 8.5, 1, 390.0), + (base + 11, 1, 440.0), + ], + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + # GPU 0: average 220 W over 10s = 2200 J. + # GPU 1: average 320 W over 10s = 3200 J. + assert result.per_gpu_energy_j == pytest.approx({"0": 2200.0, "1": 3200.0}) + assert result.total_gpu_energy_j == pytest.approx(5400.0) + assert result.avg_total_gpu_power_w == pytest.approx(540.0) + assert result.avg_power_w == pytest.approx(270.0) + + +def test_integrate_power_rejects_expected_gpu_count_mismatch(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" base = 1_700_000_000.0 + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=2, + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=4, + ) + + assert result.power_valid is False + assert "expected_gpu_count_mismatch" in result.invalid_reasons + assert result.expected_num_gpus == 4 + assert result.observed_num_gpus == 2 + + +def test_integrate_power_requires_every_gpu_to_bracket_window(tmp_path: Path): csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 _write_nvidia_csv( csv, [ - (base + 1 + sample_idx, gpu, 500.0) - for sample_idx in range(2) - for gpu in range(8) + (base - 1, 0, 500.0), + (base + 1, 0, 500.0), + (base + 3, 0, 500.0), + (base + 5, 0, 500.0), + (base + 7, 0, 500.0), + (base + 9, 0, 500.0), + (base + 11, 0, 500.0), + # GPU 1 starts after the formal benchmark start. + (base + 1, 1, 500.0), + (base + 3, 1, 500.0), + (base + 5, 1, 500.0), + (base + 7, 1, 500.0), + (base + 9, 1, 500.0), + (base + 11, 1, 500.0), ], ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is False + assert "benchmark_window_not_bracketed" in result.invalid_reasons + assert result.device_issues == {"1": ["benchmark_window_not_bracketed"]} + + +def test_integrate_power_rejects_sampling_gap(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + _write_nvidia_csv( + csv, + [ + (base - 1, 0, 500.0), + (base + 1, 0, 500.0), + # The 8-second gap makes a 1 Hz telemetry stream unusable. + (base + 9, 0, 500.0), + (base + 11, 0, 500.0), + ], + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=1, + ) + + assert result.power_valid is False + assert "sampling_gap_exceeded" in result.invalid_reasons + assert result.device_issues == {"0": ["sampling_gap_exceeded"]} + + +def test_integrate_power_ignores_invalid_samples_far_outside_window(tmp_path: Path): + """Warmup/eval corruption must not invalidate the formal benchmark slice.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + samples = [ + # Far-outside negative and conflicting duplicate samples are irrelevant. + (base - 100, 0, -1.0), + (base + 100, 0, 100.0), + (base + 100, 0, 200.0), + ] + samples.extend((base + offset, 0, 500.0) for offset in range(-1, 12)) + _write_nvidia_csv(csv, samples) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=1, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + assert result.total_gpu_energy_j == pytest.approx(5_000.0) + + +def test_run_patches_agg_with_power_and_joules(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, + ) bench = tmp_path / "bench.json" agg = tmp_path / "agg.json" - _write_bench_result(bench, start=base, end=base + 10, duration=10.0, total_output=20_000) + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + total_output=20_000, + total_input=20_000, + ) agg.write_text(json.dumps({"hw": "h200", "conc": 64}), encoding="utf-8") exit_code = run(csv, bench, agg) @@ -334,8 +523,8 @@ def test_run_patches_agg_with_power_and_joules(tmp_path: Path): assert patched["avg_power_w"] == pytest.approx(500.0) # J/output_token = 500W × 8 GPUs × 10s / 20_000 tokens = 2.0 assert patched["joules_per_output_token"] == pytest.approx(2.0) - # No input tokens were supplied -> J/total_token falls back to J/output_token. - assert patched["joules_per_total_token"] == pytest.approx(2.0) + # J/total_token = 40_000 J / (20_000 input + 20_000 output). + assert patched["joules_per_total_token"] == pytest.approx(1.0) def test_run_computes_j_per_total_token_with_input_tokens(tmp_path: Path): @@ -346,13 +535,12 @@ def test_run_computes_j_per_total_token_with_input_tokens(tmp_path: Path): """ base = 1_700_000_000.0 csv = tmp_path / "gpu_metrics.csv" - _write_nvidia_csv( + _write_constant_window_samples( csv, - [ - (base + 1 + sample_idx, gpu, 500.0) - for sample_idx in range(2) - for gpu in range(8) - ], + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, ) bench = tmp_path / "bench.json" agg = tmp_path / "agg.json" @@ -397,7 +585,10 @@ def test_run_skips_when_bench_window_missing(tmp_path: Path): patched = json.loads(agg.read_text()) assert "avg_power_w" not in patched - assert patched == {"hw": "h200"} + assert patched == { + "hw": "h200", + "power_valid": 0, + } def test_run_skips_when_csv_missing(tmp_path: Path): @@ -445,3 +636,195 @@ def test_patch_agg_result_is_atomic_via_tempfile(tmp_path: Path): assert data["joules_per_total_token"] == 0.5 # No .tmp leftover. assert not (tmp_path / "agg.json.tmp").exists() + + +def test_run_emits_complete_whole_deployment_metric_contract(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert type(patched["power_valid"]) is int + assert patched["power_valid"] == 1 + assert "power_invalid_reasons" not in patched + assert patched["avg_power_w"] == pytest.approx(500.0) + assert patched["avg_total_gpu_power_w"] == pytest.approx(4_000.0) + assert patched["total_gpu_energy_j"] == pytest.approx(40_000.0) + assert patched["joules_per_successful_query"] == pytest.approx(4_000.0) + assert patched["joules_per_input_token"] == pytest.approx(4.0) + assert patched["joules_per_output_token"] == pytest.approx(20.0) + assert patched["joules_per_total_token"] == pytest.approx(40_000 / 12_000) + + audit = json.loads(validation.read_text()) + assert audit["schema_version"] == 1 + assert audit["power_valid"] is True + assert audit["reasons"] == [] + assert audit["expected_gpu_count"] == 8 + assert audit["observed_gpu_count"] == 8 + assert audit["benchmark_window"]["start_time_unix"] == base + assert audit["benchmark_window"]["end_time_unix"] == base + 10 + assert audit["benchmark_window"]["integration_duration_s"] == 10.0 + assert audit["integration_method"] == ( + "per_device_trapezoidal_with_linear_boundary_interpolation" + ) + + +def test_run_best_effort_marks_invalid_power_and_preserves_benchmark(tmp_path: Path): + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=1_700_000_000.0, + end=1_700_000_010.0, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200", "conc": 4}), encoding="utf-8") + + exit_code = run( + tmp_path / "missing.csv", + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["hw"] == "h200" + assert patched["conc"] == 4 + assert type(patched["power_valid"]) is int + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + for metric in ( + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", + ): + assert metric not in patched + + audit = json.loads(validation.read_text()) + assert audit["power_valid"] is False + assert audit["reasons"] == ["telemetry_file_missing"] + + +def test_run_strict_mode_fails_after_writing_validation(tmp_path: Path): + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=1_700_000_000.0, + end=1_700_000_010.0, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + tmp_path / "missing.csv", + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + require_power=True, + ) + + assert exit_code == 1 + assert json.loads(agg.read_text())["power_valid"] == 0 + assert json.loads(validation.read_text())["reasons"] == ["telemetry_file_missing"] + + +def test_run_invalid_benchmark_denominator_is_auditable(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=1, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=0, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=1, + validation_result=validation, + ) + + assert exit_code == 0 + assert json.loads(agg.read_text())["power_valid"] == 0 + assert json.loads(validation.read_text())["reasons"] == [ + "invalid_successful_query_count" + ] + + +def test_cli_requires_expected_gpu_count(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + sys, + "argv", + [ + "aggregate_power.py", + "--bench-result", + "bench.json", + "--agg-result", + "agg.json", + ], + ) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 2 diff --git a/utils/test_process_result.py b/utils/test_process_result.py index ec6d245cb8..7a4deb1d75 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -4,12 +4,13 @@ 1. Testing the get_required_env_vars function directly 2. Running the script as a subprocess with mocked environment and files """ -import pytest import json import subprocess import sys from pathlib import Path +import pytest + SCRIPT_PATH = Path(__file__).parent / "process_result.py" @@ -104,6 +105,34 @@ def run_script(tmp_path, env, benchmark_result, result_filename="benchmark_resul ) +def run_script_with_broken_aggregator( + tmp_path, env, benchmark_result, result_filename="benchmark_result" +): + """Run process_result with an injected aggregator that raises unexpectedly.""" + result_file = tmp_path / f"{result_filename}.json" + result_file.write_text(json.dumps(benchmark_result)) + env = {**env, "RESULT_FILENAME": result_filename} + wrapper = f""" +import runpy +import sys +import types + +aggregate_power = types.ModuleType("aggregate_power") +def broken_run(**kwargs): + raise RuntimeError("forced aggregation failure") +aggregate_power.run = broken_run +sys.modules["aggregate_power"] = aggregate_power +runpy.run_path({str(SCRIPT_PATH)!r}, run_name="__main__") +""" + return subprocess.run( + [sys.executable, "-c", wrapper], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + ) + + # ============================================================================= # Test get_required_env_vars function # ============================================================================= @@ -622,10 +651,10 @@ def test_conc_from_benchmark_result(self, tmp_path, single_node_env_vars): class TestPowerAggregationIntegration: """End-to-end wiring: process_result.py invokes aggregate_power.py and - patches avg_power_w + joules_per_output_token into the agg JSON. + patches the validated whole-deployment power contract into the agg JSON. Exercises the env-var path resolution (GPU_METRICS_CSV), the subprocess - boundary, and the best-effort try/except that wraps the aggregator call. + boundary, topology validation, and best-effort/strict modes. """ @staticmethod @@ -668,6 +697,8 @@ def test_agg_json_gets_patched_with_power_and_joules(self, tmp_path, single_node "benchmark_start_time_unix": start, "benchmark_end_time_unix": end, "duration": 60.0, + "completed": 30, + "total_input_tokens": 240_000, "total_output_tokens": 30_000, } env = {**single_node_env_vars, "GPU_METRICS_CSV": str(csv_path)} @@ -684,9 +715,15 @@ def test_agg_json_gets_patched_with_power_and_joules(self, tmp_path, single_node assert patched["tp"] == 8 assert patched["conc"] == 64 # New power fields. + assert patched["power_valid"] == 1 assert patched["avg_power_w"] == pytest.approx(600.0, abs=0.5) + assert patched["avg_total_gpu_power_w"] == pytest.approx(4_800.0, abs=0.5) + assert patched["total_gpu_energy_j"] == pytest.approx(288_000.0, abs=0.5) + assert patched["joules_per_successful_query"] == pytest.approx(9_600.0, abs=0.05) + assert patched["joules_per_input_token"] == pytest.approx(1.2, abs=0.01) # 600W × 8 GPUs × 60s / 30_000 tokens = 9.6 J/tok assert patched["joules_per_output_token"] == pytest.approx(9.6, abs=0.05) + assert (tmp_path / "power_validation_benchmark_result.json").is_file() def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_env_vars): """Without GPU_METRICS_CSV (or with a missing file), process_result.py @@ -697,6 +734,12 @@ def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_e "max_concurrency": 64, "total_token_throughput": 1000.0, "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, } result = run_script(tmp_path, single_node_env_vars, benchmark_result) @@ -706,6 +749,13 @@ def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_e patched = json.loads(agg_path.read_text()) assert "avg_power_w" not in patched assert "joules_per_output_token" not in patched + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["reasons"] == ["telemetry_file_missing"] def test_missing_bench_timestamps_does_not_patch(self, tmp_path, single_node_env_vars): """A CSV is present but the bench JSON predates the timestamp fields @@ -730,3 +780,256 @@ def test_missing_bench_timestamps_does_not_patch(self, tmp_path, single_node_env patched = json.loads(agg_path.read_text()) assert "avg_power_w" not in patched assert "joules_per_output_token" not in patched + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + + def test_expected_gpu_count_mismatch_is_invalid(self, tmp_path, single_node_env_vars): + """TP/PP/PCP topology is checked against the observed device IDs.""" + start, end = 1_700_000_100.0, 1_700_000_110.0 + csv_path = tmp_path / "gpu_metrics.csv" + self._write_nvidia_csv(csv_path, start, end, watts_per_gpu=600.0, num_gpus=4) + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = {**single_node_env_vars, "GPU_METRICS_CSV": str(csv_path)} + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode == 0, f"Script failed: {result.stderr}" + patched = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + assert "total_gpu_energy_j" not in patched + + def test_require_power_propagates_validation_failure( + self, tmp_path, single_node_env_vars + ): + """Study/CI mode must fail after preserving validation artifacts.""" + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = {**single_node_env_vars, "REQUIRE_POWER": "1"} + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode != 0 + assert "Power validation failed" in result.stderr + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["reasons"] == ["telemetry_file_missing"] + + def test_require_power_accepts_valid_single_node_measurement( + self, tmp_path, single_node_env_vars + ): + """The strict H100/H200 canary path also has a protected success case.""" + start, end = 1_700_000_100.0, 1_700_000_110.0 + csv_path = tmp_path / "gpu_metrics.csv" + self._write_nvidia_csv( + csv_path, + start, + end, + watts_per_gpu=500.0, + num_gpus=8, + ) + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = { + **single_node_env_vars, + "GPU_METRICS_CSV": str(csv_path), + "REQUIRE_POWER": "1", + } + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode == 0, result.stderr + agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_valid"] == 1 + assert agg["total_gpu_energy_j"] == pytest.approx(40_000.0) + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["power_valid"] is True + assert validation["reasons"] == [] + + @pytest.mark.parametrize( + ("require_power", "expected_returncode"), + [(False, 0), (True, 1)], + ) + def test_internal_aggregation_error_is_always_auditable( + self, + tmp_path, + single_node_env_vars, + require_power, + expected_returncode, + ): + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = single_node_env_vars.copy() + if require_power: + env["REQUIRE_POWER"] = "1" + + result = run_script_with_broken_aggregator(tmp_path, env, benchmark_result) + + assert result.returncode == expected_returncode + agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_valid"] == 0 + assert "power_invalid_reasons" not in agg + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["power_valid"] is False + assert validation["reasons"] == ["aggregation_internal_error"] + assert validation["internal_error"]["type"] == "RuntimeError" + + def test_stop_gpu_monitor_appends_final_nvidia_sample(self, tmp_path): + """Stopping between 1 Hz ticks still records one post-benchmark sample.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + args_log = tmp_path / "nvidia_args.txt" + fake_nvidia_smi = fake_bin / "nvidia-smi" + fake_nvidia_smi.write_text( + "#!/usr/bin/env bash\n" + f"printf '%s\\n' \"$*\" > {str(args_log)!r}\n" + "printf '%s\\n' " + "'2026/07/23 12:00:11.000, 0, 500.00 W, 65, 1000, 1000, 90 %, 10 %'\n" + ) + fake_nvidia_smi.chmod(0o755) + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text( + "timestamp, index, power.draw [W], temperature.gpu, " + "clocks.current.sm [MHz], clocks.current.memory [MHz], " + "utilization.gpu [%], utilization.memory [%]\n" + ) + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=nvidia +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "--format=csv,noheader" in args_log.read_text() + assert "2026/07/23 12:00:11.000, 0, 500.00 W" in metrics.read_text() + + +# ============================================================================= +# Static CI/artifact contract +# ============================================================================= + + +class TestPowerWorkflowContract: + """Keep power audit inputs separate from the canonical agg artifact.""" + + @staticmethod + def _repo_root(): + return Path(__file__).parents[1] + + def test_benchmark_workflow_uploads_independent_power_audit_bundle(self): + workflow = ( + self._repo_root() / ".github/workflows/benchmark-tmpl.yml" + ).read_text() + + assert "name: power_audit_${{ env.RESULT_FILENAME }}" in workflow + assert "${{ env.RESULT_FILENAME }}.json" in workflow + assert "agg_${{ env.RESULT_FILENAME }}.json" in workflow + assert "gpu_metrics.csv" in workflow + assert "power_validation_${{ env.RESULT_FILENAME }}.json" in workflow + assert ( + "always() && !inputs.eval-only && " + "inputs.scenario-type != 'agentic-coding'" + ) in workflow + assert "require-power:" in workflow + assert "REQUIRE_POWER: ${{ inputs.require-power && '1' || '0' }}" in workflow + + # The canonical bmk artifact remains aggregate-only so downstream + # result collection cannot mistake the raw benchmark JSON for an agg. + upload_result = workflow.split("- name: Upload result", 1)[1].split( + "- name: Upload agentic aggregated result", 1 + )[0] + assert "path: agg_${{ env.RESULT_FILENAME }}.json" in upload_result + assert "power_validation_" not in upload_result + + def test_cpu_workflow_covers_power_code_and_both_test_suites(self): + workflow = ( + self._repo_root() / ".github/workflows/test-process-result.yml" + ).read_text() + + for watched_path in ( + "utils/aggregate_power.py", + "utils/test_aggregate_power.py", + "utils/process_result.py", + "utils/test_process_result.py", + ".github/workflows/benchmark-tmpl.yml", + ".github/workflows/e2e-tests.yml", + ".github/workflows/test-process-result.yml", + ): + assert watched_path in workflow + assert "test_aggregate_power.py" in workflow + assert "test_process_result.py" in workflow + assert "github.event.pull_request.draft" not in workflow + + def test_e2e_dispatch_can_enable_strict_power_validation(self): + workflow = ( + self._repo_root() / ".github/workflows/e2e-tests.yml" + ).read_text() + + assert workflow.count("require-power:") >= 3 + single_node_job = workflow.split("test-sweep-single-node:", 1)[1].split( + "test-sweep-evals:", 1 + )[0] + assert "require-power: ${{ inputs.require-power }}" in single_node_job From f074b7a9b97b8cd4aeb1579f9a44a8a516466090 Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 23 Jul 2026 16:31:21 -0700 Subject: [PATCH 2/3] fix(power): reject malformed telemetry samples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Treat unparseable timestamps and in-window power values as validation failures so best-effort runs emit power_valid=0 and strict runs fail without publishing Joule metrics. Add regression coverage for both cases. 中文：拒绝格式异常的 GPU 遥测样本。将无法解析的时间戳和窗口内功耗值视为校验失败，使尽力模式输出 power_valid=0，严格模式失败，并避免发布 Joule 指标；同时补充两个回归测试。 --- utils/aggregate_power.py | 9 +++-- utils/test_aggregate_power.py | 64 +++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 3 deletions(-) diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index 71aa9daca4..ef432a8129 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -311,9 +311,10 @@ def integrate_power( ) -> PowerIntegration: """Validate and integrate per-device GPU power over the formal window. - A valid stream must expose stable GPU identities, match the expected - topology when supplied, bracket both window boundaries for every device, - and have no in-window sampling gap larger than ``max_sample_gap_s``. + A valid stream must contain parseable timestamps and power values, expose + stable GPU identities, match the expected topology when supplied, bracket + both window boundaries for every device, and have no in-window sampling gap + larger than ``max_sample_gap_s``. """ reasons: list[str] = [] if ( @@ -383,6 +384,7 @@ def integrate_power( for row in reader: timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip()) if timestamp is None or not math.isfinite(timestamp): + _append_reason(reasons, "invalid_timestamp_sample") continue # Only the formal window and its possible boundary neighbors # can affect integration. Bad warmup/eval rows farther away @@ -396,6 +398,7 @@ def integrate_power( power = _parse_power((row.get(power_col) or "").strip()) gpu_id = (row.get(gpu_col) or "").strip() if power is None: + _append_reason(reasons, "invalid_power_sample") continue if not math.isfinite(power) or power < 0: _append_reason(reasons, "invalid_power_sample") diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index 8a6b160281..7f4affc2cd 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -490,6 +490,70 @@ def test_integrate_power_ignores_invalid_samples_far_outside_window(tmp_path: Pa assert result.total_gpu_energy_j == pytest.approx(5_000.0) +@pytest.mark.parametrize( + ("malformed_row", "expected_reason"), + [ + ("not-a-timestamp, 0, 500 W", "invalid_timestamp_sample"), + ("{timestamp}, 0, [N/A]", "invalid_power_sample"), + ], + ids=["timestamp", "power"], +) +def test_run_rejects_malformed_telemetry_inside_window( + tmp_path: Path, + malformed_row: str, + expected_reason: str, +): + """A rejected 1 Hz row must not be interpolated into valid energy metrics.""" + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + rows = ["timestamp, index, power.draw [W]"] + for offset in range(-1, 12): + timestamp = _nvidia_ts(base + offset) + if offset == 5: + rows.append(malformed_row.format(timestamp=timestamp)) + else: + rows.append(f"{timestamp}, 0, 500 W") + csv.write_text("\n".join(rows) + "\n", encoding="utf-8") + + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=1, + total_input=1_000, + total_output=1_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=1, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["power_valid"] == 0 + for metric in ( + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", + ): + assert metric not in patched + audit = json.loads(validation.read_text()) + assert audit["reasons"] == [expected_reason] + + def test_run_patches_agg_with_power_and_joules(tmp_path: Path): base = 1_700_000_000.0 csv = tmp_path / "gpu_metrics.csv" From c2f27fe453ef613f4eb0f8d356a6dbd80ffed7dd Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 23 Jul 2026 16:57:43 -0700 Subject: [PATCH 3/3] fix(power): discard truncated final telemetry row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove an incomplete trailing nvidia-smi CSV row before appending the deterministic post-benchmark sample, preventing row concatenation from silently corrupting energy integration. Add a shell-lifecycle regression test covering the interrupted-write case. 中文：在追加确定性的基准测试后采样前，删除未写完整的 nvidia-smi CSV 末行，避免行拼接静默污染能耗积分；新增覆盖中断写入场景的 shell 生命周期回归测试。 --- benchmarks/benchmark_lib.sh | 21 +++++++++++-- utils/test_process_result.py | 60 ++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index efb55cb967..6c4606b536 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -165,9 +165,24 @@ stop_gpu_monitor() { # version, so strict AMD lifecycle validation remains follow-up work. case "$GPU_MONITOR_VENDOR" in nvidia) - nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ - --format=csv,noheader >> "$GPU_METRICS_CSV" 2>/dev/null || - echo "[GPU Monitor] Warning: final NVIDIA sample failed" >&2 + local append_final_nvidia_sample=true + local repaired_metrics="${GPU_METRICS_CSV}.repair.$$" + if [[ -s "$GPU_METRICS_CSV" ]] && + ! tail -c 1 "$GPU_METRICS_CSV" | grep -q '^$'; then + if sed '$d' "$GPU_METRICS_CSV" > "$repaired_metrics" && + mv "$repaired_metrics" "$GPU_METRICS_CSV"; then + echo "[GPU Monitor] Dropped truncated trailing sample" + else + rm -f "$repaired_metrics" + append_final_nvidia_sample=false + echo "[GPU Monitor] Warning: could not repair truncated trailing sample" >&2 + fi + fi + if [[ "$append_final_nvidia_sample" == true ]]; then + nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ + --format=csv,noheader >> "$GPU_METRICS_CSV" 2>/dev/null || + echo "[GPU Monitor] Warning: final NVIDIA sample failed" >&2 + fi ;; esac echo "[GPU Monitor] Stopped (PID=$GPU_MONITOR_PID)" diff --git a/utils/test_process_result.py b/utils/test_process_result.py index 7a4deb1d75..28e06a62eb 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -966,6 +966,66 @@ def test_stop_gpu_monitor_appends_final_nvidia_sample(self, tmp_path): assert "--format=csv,noheader" in args_log.read_text() assert "2026/07/23 12:00:11.000, 0, 500.00 W" in metrics.read_text() + def test_stop_gpu_monitor_drops_truncated_row_before_final_sample(self, tmp_path): + """A killed monitor cannot concatenate its partial row with the final sample.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_nvidia_smi = fake_bin / "nvidia-smi" + final_sample = ( + "2026/07/23 12:00:11.000, 0, 500.00 W, " + "65, 1000, 1000, 90 %, 10 %" + ) + fake_nvidia_smi.write_text( + "#!/usr/bin/env bash\n" + f"printf '%s\\n' {final_sample!r}\n" + ) + fake_nvidia_smi.chmod(0o755) + + header = ( + "timestamp, index, power.draw [W], temperature.gpu, " + "clocks.current.sm [MHz], clocks.current.memory [MHz], " + "utilization.gpu [%], utilization.memory [%]" + ) + complete_sample = ( + "2026/07/23 12:00:09.000, 0, 490.00 W, " + "64, 990, 990, 89 %, 9 %" + ) + truncated_sample = "2026/07/23 12:00:10.000, 0, 52" + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text( + f"{header}\n{complete_sample}\n{truncated_sample}" + ) + + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=nvidia +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert metrics.read_text().splitlines() == [ + header, + complete_sample, + final_sample, + ] + # ============================================================================= # Static CI/artifact contract