From 9e214274cdc177d585a3d5fc8c8e261e155c3143 Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 12 Aug 2026 22:16:48 -0700 Subject: [PATCH 1/3] feat(power): stamp whole-deployment semantics with a schema version --- .github/AGENT_OPERATIONS.md | 2 ++ utils/aggregate_power.py | 7 +++++++ utils/aggregate_power_multinode.py | 2 ++ utils/process_result.py | 2 ++ utils/test_aggregate_power.py | 3 +++ utils/test_aggregate_power_multinode.py | 2 ++ utils/test_process_result.py | 4 ++++ 7 files changed, 22 insertions(+) diff --git a/.github/AGENT_OPERATIONS.md b/.github/AGENT_OPERATIONS.md index 0ef9acd1ea..cb90d423b5 100644 --- a/.github/AGENT_OPERATIONS.md +++ b/.github/AGENT_OPERATIONS.md @@ -94,6 +94,8 @@ Single-node fixed-sequence results may include `power_valid`, `avg_power_w`, `av Multinode disaggregated results add `prefill_gpu_energy_j`, `decode_gpu_energy_j`, `prefill_avg_power_w`, `decode_avg_power_w`, `prefill_joules_per_input_token`, and `decode_joules_per_output_token`. Role energy covers the full formal benchmark window, not kernel-level phases, and the role watts are that energy divided by the same window and by the role's GPU count. +Every power result — valid or invalid, single-node or multinode — carries `power_metric_schema_version`. Version 2 defines each unprefixed `joules_per_*` field as whole-deployment GPU-board energy over the named denominator; role-scoped energy uses the explicit `prefill_*` / `decode_*` keys. Rows without the field predate the whole-deployment switch and their unprefixed joules are not comparable across topologies. + For srt-slurm recipes, `telemetry: {provider: dcgm-power}` enables official energy collection. `runners/launch_gb200-nv.sh` and `runners/launch_gb300-nv.sh` are the source of truth for `POWER_SRT_SLURM_PIN`. CI derives `POWER_PRODUCER_SHA` from the launcher stamp. `utils/test_gb200_power_official_contract.py` and `utils/test_gb300_power_official_contract.py` enforce the recipe/launcher contract. Only `PRECISION=fp8` dcgm-power lanes are validated. Power audit artifacts are named `power_audit_` and contain `power_validation_.json` for single-node runs or `power_validation__*.json` for multinode runs. They are uploaded even when validation fails. diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index 24db2cd829..b1368580f0 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -46,6 +46,12 @@ "joules_per_total_token", } +# The unprefixed joules_per_* fields silently switched from role-local to +# whole-deployment energy when multinode aggregation landed, and the values +# alone cannot distinguish the two. Stamp the semantics so consumers fail +# closed on unversioned rows instead of guessing. +POWER_METRIC_SCHEMA_VERSION = 2 + @dataclass(frozen=True) class PowerIntegration: @@ -756,6 +762,7 @@ def _patch_power_result( 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_metric_schema_version"] = POWER_METRIC_SCHEMA_VERSION data["power_valid"] = int(power_valid) data.pop("power_invalid_reasons", None) if power_valid: diff --git a/utils/aggregate_power_multinode.py b/utils/aggregate_power_multinode.py index cabed59d6d..685cd4e65f 100644 --- a/utils/aggregate_power_multinode.py +++ b/utils/aggregate_power_multinode.py @@ -39,6 +39,7 @@ from pathlib import Path, PurePosixPath from aggregate_power import ( + POWER_METRIC_SCHEMA_VERSION, BenchmarkData, _append_reason, _integrate_device, @@ -1165,6 +1166,7 @@ def _patch_agg(agg_path: Path, audit: MultinodePowerAudit) -> None: data = json.loads(agg_path.read_text(encoding="utf-8")) for key in _ALL_POWER_METRIC_KEYS: data.pop(key, None) + data["power_metric_schema_version"] = POWER_METRIC_SCHEMA_VERSION data["power_valid"] = int(audit.power_valid) data.pop("power_invalid_reasons", None) if audit.power_valid: diff --git a/utils/process_result.py b/utils/process_result.py index a8bdc8cca6..9137b55384 100644 --- a/utils/process_result.py +++ b/utils/process_result.py @@ -67,6 +67,7 @@ def record_power_internal_error( try: from aggregate_power import ( _POWER_METRIC_KEYS, + POWER_METRIC_SCHEMA_VERSION, _empty_integration, _validation_payload, _write_json_atomic, @@ -77,6 +78,7 @@ def record_power_internal_error( agg_data.pop(key, None) for key in _MULTINODE_ROLE_METRIC_KEYS: agg_data.pop(key, None) + agg_data["power_metric_schema_version"] = POWER_METRIC_SCHEMA_VERSION agg_data["power_valid"] = 0 agg_data.pop("power_invalid_reasons", None) _write_json_atomic(agg_result, agg_data) diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index 89e42034e8..1b7029a454 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -670,6 +670,7 @@ def test_run_skips_when_bench_window_missing(tmp_path: Path): assert "avg_power_w" not in patched assert patched == { "hw": "h200", + "power_metric_schema_version": 2, "power_valid": 0, } @@ -755,6 +756,7 @@ def test_run_emits_complete_whole_deployment_metric_contract(tmp_path: Path): assert exit_code == 0 patched = json.loads(agg.read_text()) + assert patched["power_metric_schema_version"] == 2 assert type(patched["power_valid"]) is int assert patched["power_valid"] == 1 assert "power_invalid_reasons" not in patched @@ -809,6 +811,7 @@ def test_run_best_effort_marks_invalid_power_and_preserves_benchmark(tmp_path: P patched = json.loads(agg.read_text()) assert patched["hw"] == "h200" assert patched["conc"] == 4 + assert patched["power_metric_schema_version"] == 2 assert type(patched["power_valid"]) is int assert patched["power_valid"] == 0 assert "power_invalid_reasons" not in patched diff --git a/utils/test_aggregate_power_multinode.py b/utils/test_aggregate_power_multinode.py index e4c78a1156..16e77db51b 100644 --- a/utils/test_aggregate_power_multinode.py +++ b/utils/test_aggregate_power_multinode.py @@ -213,6 +213,7 @@ def assert_invalid(pkg, expected_reason, **run_kwargs): """Both modes must reject: metrics withheld always, exit code differs.""" assert pkg.run(require_power=False, **run_kwargs) == 0 agg = pkg.agg() + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 0 for key in apm.WHOLE_METRIC_KEYS + apm.ROLE_METRIC_KEYS: assert key not in agg @@ -229,6 +230,7 @@ def test_emits_all_metrics_exactly(self, tmp_path): assert pkg.run() == 0 agg = pkg.agg() + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 1 assert agg["avg_power_w"] == 350.0 assert agg["avg_total_gpu_power_w"] == 1400.0 diff --git a/utils/test_process_result.py b/utils/test_process_result.py index 4d5219010f..c84f4dc5f5 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -875,6 +875,7 @@ def test_require_power_accepts_valid_single_node_measurement( assert result.returncode == 0, result.stderr agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 1 assert agg["total_gpu_energy_j"] == pytest.approx(40_000.0) validation = json.loads( @@ -914,6 +915,7 @@ def test_internal_aggregation_error_is_always_auditable( assert result.returncode == expected_returncode agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 0 assert "power_invalid_reasons" not in agg validation = json.loads( @@ -1221,6 +1223,7 @@ def test_valid_package_patches_role_energy(self, tmp_path, power_env): assert result.returncode == 0, f"Script failed: {result.stderr}" agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 1 assert agg["prefill_gpu_energy_j"] == 48000.0 assert agg["decode_gpu_energy_j"] == 36000.0 @@ -1233,6 +1236,7 @@ def test_missing_package_is_best_effort(self, tmp_path, power_env): assert result.returncode == 0, f"Script failed: {result.stderr}" agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_metric_schema_version"] == 2 assert agg["power_valid"] == 0 for key in WHOLE_METRIC_KEYS + ROLE_METRIC_KEYS: assert key not in agg From 5da99af905db50f3eca4f0d92b9f946d031267cc Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 12 Aug 2026 22:21:25 -0700 Subject: [PATCH 2/3] feat(agentx): adapt profiling windows for power validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：将 AIPerf profiling 生命周期与成功请求计数适配到严格功耗校验契约。 --- .github/workflows/test-process-result.yml | 6 + utils/agentic/aggregation/power_adapter.py | 245 +++++++++++++ .../agentic/aggregation/test_power_adapter.py | 321 ++++++++++++++++++ utils/pytest.ini | 4 + 4 files changed, 576 insertions(+) create mode 100644 utils/agentic/aggregation/power_adapter.py create mode 100644 utils/agentic/aggregation/test_power_adapter.py diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml index e6a380bd82..88506242c7 100644 --- a/.github/workflows/test-process-result.yml +++ b/.github/workflows/test-process-result.yml @@ -14,6 +14,12 @@ on: - 'runners/launch_gb300-nv.sh' - 'utils/aggregate_power.py' - 'utils/aggregate_power_multinode.py' + - 'utils/agentic/aggregation/power_adapter.py' + - 'utils/agentic/aggregation/test_power_adapter.py' + - 'utils/agentic/aggregation/test_power_lifecycle.py' + - 'utils/agentic/aggregation/process_agentic_result.py' + - 'utils/agentic/aggregation/test_process_agentic_result.py' + - 'utils/pytest.ini' - 'utils/bench_serving/benchmark_serving.py' - 'utils/process_result.py' - 'utils/test_aggregate_power.py' diff --git a/utils/agentic/aggregation/power_adapter.py b/utils/agentic/aggregation/power_adapter.py new file mode 100644 index 0000000000..0e98debbc9 --- /dev/null +++ b/utils/agentic/aggregation/power_adapter.py @@ -0,0 +1,245 @@ +"""Adapt AIPerf profiling artifacts to the strict power-window contract.""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import re +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from utils.aggregate_power import ( + _empty_integration, + _patch_power_result, + _validation_payload, + _write_json_atomic, +) +from utils.aggregate_power import run as run_power + +from .process_agentic_result import _resolve_artifact_dir +from .request_metrics import extract_per_record_ints, load_aggregate, load_records + +_UTC_OFFSET_RE = re.compile(r"^([+-])(\d{2}):?(\d{2})$") + + +def _captured_timezone(result_dir: Path) -> tuple[timezone | None, str | None]: + """Load the launch-time UTC offset captured beside AgentX telemetry.""" + offset_path = result_dir / "agentic_power_timezone_offset.txt" + if not offset_path.is_file(): + return None, "profile_timezone_offset_missing" + try: + raw_offset = offset_path.read_text(encoding="utf-8").strip() + except OSError: + return None, "profile_timezone_offset_invalid" + match = _UTC_OFFSET_RE.fullmatch(raw_offset) + if match is None: + return None, "profile_timezone_offset_invalid" + hours, minutes = int(match.group(2)), int(match.group(3)) + if hours > 23 or minutes > 59: + return None, "profile_timezone_offset_invalid" + direction = 1 if match.group(1) == "+" else -1 + return timezone(direction * timedelta(hours=hours, minutes=minutes)), None + + +def _parse_profile_timestamp(value: Any, *, fallback_tz: timezone | None) -> float | None: + """Parse a timezone-aware ISO timestamp or Unix epoch seconds.""" + if isinstance(value, bool): + return None + if isinstance(value, int | float): + timestamp = float(value) + return timestamp if math.isfinite(timestamp) else None + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + if fallback_tz is None: + return None + parsed = parsed.replace(tzinfo=fallback_tz) + return parsed.astimezone(timezone.utc).timestamp() + + +def build_power_window(result_dir: Path) -> tuple[dict[str, int | float] | None, list[str]]: + """Build a strict benchmark window from successful profiling requests.""" + artifact_dir = _resolve_artifact_dir(result_dir) + aggregate_path = artifact_dir / "profile_export_aiperf.json" + records_path = artifact_dir / "profile_export.jsonl" + if not aggregate_path.is_file() or not records_path.is_file(): + return None, ["profile_artifacts_missing"] + + try: + aggregate = load_aggregate(aggregate_path) + records = load_records(records_path) + except (OSError, json.JSONDecodeError, TypeError, ValueError): + return None, ["profile_artifacts_invalid"] + + raw_start = aggregate.get("start_time") + raw_end = aggregate.get("end_time") + if raw_start is None or raw_end is None: + return None, ["profile_window_missing"] + parsed_datetimes: list[datetime] = [] + for value in (raw_start, raw_end): + if isinstance(value, str): + try: + parsed_datetimes.append( + datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + ) + except ValueError: + pass + needs_captured_timezone = any(value.tzinfo is None for value in parsed_datetimes) + fallback_tz = None + if needs_captured_timezone: + fallback_tz, timezone_reason = _captured_timezone(result_dir) + if timezone_reason is not None: + return None, [timezone_reason] + + start = _parse_profile_timestamp(raw_start, fallback_tz=fallback_tz) + end = _parse_profile_timestamp(raw_end, fallback_tz=fallback_tz) + if start is None or end is None or end <= start: + return None, ["profile_window_invalid"] + + completed = len(records) + if completed <= 0: + return None, ["successful_request_count_invalid"] + + input_tokens = extract_per_record_ints(records, "input_sequence_length") + output_tokens = extract_per_record_ints(records, "output_sequence_length") + if ( + len(input_tokens) != completed + or len(output_tokens) != completed + or any(value < 0 for value in input_tokens + output_tokens) + or sum(input_tokens) <= 0 + or sum(output_tokens) <= 0 + ): + return None, ["incomplete_token_accounting"] + + return ( + { + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": end - start, + "completed": completed, + "total_input_tokens": sum(input_tokens), + "total_output_tokens": sum(output_tokens), + }, + [], + ) + + +def _record_adapter_failure( + *, + result_dir: Path, + agg_result: Path, + expected_num_gpus: int | None, + reasons: list[str], +) -> None: + """Write the same invalid aggregate and audit artifacts as aggregate_power.""" + csv_path = result_dir / "gpu_metrics.csv" + window_path = result_dir / "agentic_power_window.json" + validation_path = result_dir / "power_validation.json" + integration = _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + _patch_power_result(agg_result, power_valid=False, metrics={}) + payload = _validation_payload( + csv_path=csv_path, + bench_result=window_path, + benchmark=None, + integration=integration, + power_valid=False, + reasons=reasons, + metrics={}, + accumulator_check=None, + ) + payload["window_source"] = "aiperf_profile_lifecycle" + _write_json_atomic(validation_path, payload) + + +def run_agentic_power( + *, + result_dir: Path, + agg_result: Path, + expected_num_gpus: int | None, + require_power: bool = False, +) -> int: + """Validate AgentX power telemetry, failing only in strict mode.""" + window, reasons = build_power_window(result_dir) + if window is None: + try: + _record_adapter_failure( + result_dir=result_dir, + agg_result=agg_result, + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print( + f"[agentx_power] Failed to record adapter failure: {exc}", + file=sys.stderr, + ) + print( + f"[agentx_power] Power-window adaptation failed: {', '.join(reasons)}", + file=sys.stderr, + ) + return 1 if require_power else 0 + + window_path = result_dir / "agentic_power_window.json" + try: + _write_json_atomic(window_path, window) + except OSError: + reasons = ["power_window_unwritable"] + try: + _record_adapter_failure( + result_dir=result_dir, + agg_result=agg_result, + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print( + f"[agentx_power] Failed to record adapter failure: {exc}", + file=sys.stderr, + ) + print( + f"[agentx_power] Power-window adaptation failed: {', '.join(reasons)}", + file=sys.stderr, + ) + return 1 if require_power else 0 + return run_power( + result_dir / "gpu_metrics.csv", + window_path, + agg_result, + expected_num_gpus=expected_num_gpus, + validation_result=result_dir / "power_validation.json", + require_power=require_power, + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--result-dir", type=Path, required=True) + parser.add_argument("--agg-result", type=Path, required=True) + parser.add_argument("--expected-num-gpus", type=int) + parser.add_argument( + "--require-power", + action="store_true", + default=os.environ.get("REQUIRE_POWER", "").lower() in {"1", "true", "yes"}, + ) + args = parser.parse_args() + return run_agentic_power( + result_dir=args.result_dir, + agg_result=args.agg_result, + expected_num_gpus=args.expected_num_gpus, + require_power=args.require_power, + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/utils/agentic/aggregation/test_power_adapter.py b/utils/agentic/aggregation/test_power_adapter.py new file mode 100644 index 0000000000..e4a26c627a --- /dev/null +++ b/utils/agentic/aggregation/test_power_adapter.py @@ -0,0 +1,321 @@ +"""Strict AgentX-to-power window adaptation tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + + +def _record( + *, + start_ns: int, + end_ns: int, + input_tokens: int | None = 100, + output_tokens: int | None = 50, + phase: str = "profiling", + error: dict | None = None, +) -> dict: + metrics = {} + if input_tokens is not None: + metrics["input_sequence_length"] = {"value": input_tokens, "unit": "tokens"} + if output_tokens is not None: + metrics["output_sequence_length"] = {"value": output_tokens, "unit": "tokens"} + return { + "metadata": { + "benchmark_phase": phase, + "request_start_ns": start_ns, + "request_end_ns": end_ns, + }, + "metrics": metrics, + "error": error, + } + + +def _write_artifacts( + tmp_path: Path, + *, + aggregate: dict | None = None, + records: list[dict] | None = None, +) -> Path: + result_dir = tmp_path / "results" + artifacts = result_dir / "aiperf_artifacts" + artifacts.mkdir(parents=True) + aggregate = aggregate or { + "start_time": "2023-11-14T22:13:21+00:00", + "end_time": "2023-11-14T22:13:24+00:00", + } + records = records or [ + _record(start_ns=1_700_000_001_500_000_000, end_ns=1_700_000_002_000_000_000), + _record( + start_ns=1_700_000_002_500_000_000, + end_ns=1_700_000_003_500_000_000, + input_tokens=200, + output_tokens=100, + ), + _record( + start_ns=1_699_999_999_000_000_000, + end_ns=1_700_000_000_000_000_000, + phase="warmup", + ), + _record( + start_ns=1_700_000_003_000_000_000, + end_ns=1_700_000_004_000_000_000, + error={"type": "HTTPStatusError"}, + ), + ] + (artifacts / "profile_export_aiperf.json").write_text( + json.dumps(aggregate), encoding="utf-8" + ) + (artifacts / "profile_export.jsonl").write_text( + "".join(json.dumps(record) + "\n" for record in records), encoding="utf-8" + ) + return result_dir + + +def _write_power_csv(result_dir: Path) -> None: + rows = ["timestamp, index, power.draw [W]"] + for timestamp in (1_700_000_020.0, 1_700_000_021.0, 1_700_000_024.0, 1_700_000_025.0): + rows.append(f"{timestamp}, 0, 400 W") + rows.append(f"{timestamp}, 1, 600 W") + (result_dir / "gpu_metrics.csv").write_text("\n".join(rows) + "\n", encoding="utf-8") + + +def test_build_power_window_uses_profile_lifecycle_and_successful_records(tmp_path: Path): + from utils.agentic.aggregation.power_adapter import build_power_window + + result_dir = _write_artifacts(tmp_path) + + window, reasons = build_power_window(result_dir) + + assert reasons == [] + assert window == { + "benchmark_start_time_unix": 1_700_000_001.0, + "benchmark_end_time_unix": 1_700_000_004.0, + "duration": 3.0, + "completed": 2, + "total_input_tokens": 300, + "total_output_tokens": 150, + } + + +def test_build_power_window_applies_captured_offset_to_naive_aiperf_times(tmp_path: Path): + from utils.agentic.aggregation.power_adapter import build_power_window + + result_dir = _write_artifacts( + tmp_path, + aggregate={ + "start_time": "2023-11-14T14:13:21", + "end_time": "2023-11-14T14:13:24", + }, + ) + (result_dir / "agentic_power_timezone_offset.txt").write_text("-0800\n") + + window, reasons = build_power_window(result_dir) + + assert reasons == [] + assert window is not None + assert window["benchmark_start_time_unix"] == 1_700_000_001.0 + assert window["benchmark_end_time_unix"] == 1_700_000_004.0 + + +@pytest.mark.parametrize( + ("offset", "expected_reason"), + [(None, "profile_timezone_offset_missing"), ("PST", "profile_timezone_offset_invalid")], +) +def test_build_power_window_rejects_naive_times_without_valid_captured_offset( + tmp_path: Path, + offset: str | None, + expected_reason: str, +): + from utils.agentic.aggregation.power_adapter import build_power_window + + result_dir = _write_artifacts( + tmp_path, + aggregate={ + "start_time": "2023-11-14T14:13:21", + "end_time": "2023-11-14T14:13:24", + }, + ) + if offset is not None: + (result_dir / "agentic_power_timezone_offset.txt").write_text(offset) + + window, reasons = build_power_window(result_dir) + + assert window is None + assert expected_reason in reasons + + +@pytest.mark.parametrize( + ("aggregate", "records", "expected_reason"), + [ + ({"end_time": "2023-11-14T22:13:24+00:00"}, None, "profile_window_missing"), + ( + { + "start_time": "2023-11-14T22:13:24+00:00", + "end_time": "2023-11-14T22:13:21+00:00", + }, + None, + "profile_window_invalid", + ), + ( + None, + [_record(start_ns=1, end_ns=2, output_tokens=None)], + "incomplete_token_accounting", + ), + ( + None, + [_record(start_ns=1, end_ns=2, phase="warmup")], + "successful_request_count_invalid", + ), + ], +) +def test_build_power_window_rejects_ambiguous_inputs( + tmp_path: Path, + aggregate: dict | None, + records: list[dict] | None, + expected_reason: str, +): + from utils.agentic.aggregation.power_adapter import build_power_window + + result_dir = _write_artifacts(tmp_path, aggregate=aggregate, records=records) + + window, reasons = build_power_window(result_dir) + + assert window is None + assert expected_reason in reasons + + +def test_run_agentic_power_patches_strict_whole_deployment_metrics(tmp_path: Path): + from utils.agentic.aggregation.power_adapter import run_agentic_power + + result_dir = _write_artifacts(tmp_path) + # The ISO window above is 1700000001..1700000004. Offset the numeric + # telemetry by 20 seconds only when replacing the aggregate, keeping this + # fixture's human-readable timestamps obvious. + aggregate_path = result_dir / "aiperf_artifacts" / "profile_export_aiperf.json" + aggregate_path.write_text( + json.dumps( + { + "start_time": "2023-11-14T22:13:41+00:00", + "end_time": "2023-11-14T22:13:44+00:00", + } + ), + encoding="utf-8", + ) + _write_power_csv(result_dir) + agg_path = tmp_path / "agg_agentx.json" + agg_path.write_text(json.dumps({"hw": "h200", "scenario_type": "agentic-coding"})) + + exit_code = run_agentic_power( + result_dir=result_dir, + agg_result=agg_path, + expected_num_gpus=2, + require_power=True, + ) + + assert exit_code == 0 + agg = json.loads(agg_path.read_text()) + assert agg["power_metric_schema_version"] == 2 + assert agg["power_valid"] == 1 + assert agg["avg_power_w"] == 500.0 + assert agg["avg_total_gpu_power_w"] == 1_000.0 + assert agg["total_gpu_energy_j"] == 3_000.0 + assert agg["joules_per_successful_query"] == 1_500.0 + assert agg["joules_per_input_token"] == 10.0 + assert agg["joules_per_output_token"] == 20.0 + assert agg["joules_per_total_token"] == pytest.approx(6.666667) + assert (result_dir / "agentic_power_window.json").is_file() + validation = json.loads((result_dir / "power_validation.json").read_text()) + assert validation["power_valid"] is True + + +@pytest.mark.parametrize(("require_power", "expected_exit"), [(False, 0), (True, 1)]) +def test_run_agentic_power_records_window_write_failure_before_returning( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + require_power: bool, + expected_exit: int, +): + from utils.agentic.aggregation import power_adapter + + result_dir = _write_artifacts(tmp_path) + agg_path = tmp_path / "agg_agentx.json" + agg_path.write_text( + json.dumps( + { + "power_valid": 1, + "avg_power_w": 999, + "joules_per_output_token": 999, + } + ) + ) + window_path = result_dir / "agentic_power_window.json" + original_write_json_atomic = power_adapter._write_json_atomic + + def write_json_atomic(path: Path, payload: dict) -> None: + if path == window_path: + raise OSError("simulated full disk") + original_write_json_atomic(path, payload) + + monkeypatch.setattr(power_adapter, "_write_json_atomic", write_json_atomic) + + exit_code = power_adapter.run_agentic_power( + result_dir=result_dir, + agg_result=agg_path, + expected_num_gpus=2, + require_power=require_power, + ) + + assert exit_code == expected_exit + assert not window_path.exists() + agg = json.loads(agg_path.read_text()) + assert agg["power_metric_schema_version"] == 2 + assert agg["power_valid"] == 0 + assert "avg_power_w" not in agg + assert "joules_per_output_token" not in agg + validation = json.loads((result_dir / "power_validation.json").read_text()) + assert validation["power_valid"] is False + assert validation["reasons"] == ["power_window_unwritable"] + assert "Power-window adaptation failed: power_window_unwritable" in capsys.readouterr().err + + +@pytest.mark.parametrize(("require_power", "expected_exit"), [(False, 0), (True, 1)]) +def test_run_agentic_power_records_adapter_failure_before_returning( + tmp_path: Path, require_power: bool, expected_exit: int +): + from utils.agentic.aggregation.power_adapter import run_agentic_power + + result_dir = _write_artifacts( + tmp_path, + records=[_record(start_ns=1, end_ns=2, output_tokens=None)], + ) + agg_path = tmp_path / "agg_agentx.json" + agg_path.write_text( + json.dumps( + { + "power_valid": 1, + "avg_power_w": 999, + "joules_per_output_token": 999, + } + ) + ) + + exit_code = run_agentic_power( + result_dir=result_dir, + agg_result=agg_path, + expected_num_gpus=2, + require_power=require_power, + ) + + assert exit_code == expected_exit + agg = json.loads(agg_path.read_text()) + assert agg["power_metric_schema_version"] == 2 + assert agg["power_valid"] == 0 + assert "avg_power_w" not in agg + assert "joules_per_output_token" not in agg + validation = json.loads((result_dir / "power_validation.json").read_text()) + assert validation["power_valid"] is False + assert "incomplete_token_accounting" in validation["reasons"] diff --git a/utils/pytest.ini b/utils/pytest.ini index c3cd9aac7a..71e441e797 100644 --- a/utils/pytest.ini +++ b/utils/pytest.ini @@ -1,5 +1,9 @@ [pytest] testpaths = . +# The agentic aggregation tests import through the repo-root package path +# (utils.agentic....) because the production modules do; CI runs pytest from +# utils/, so the repo root must be on sys.path in both invocation modes. +pythonpath = .. python_files = test_*.py python_classes = Test* python_functions = test_* From 15a7b705d3a16c863cfbde57509e1ac1be90c71c Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 12 Aug 2026 22:26:21 -0700 Subject: [PATCH 3/3] feat(agentx): measure power across replay lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：在共享 AgentX 重放生命周期中采集单节点 GPU 功耗，并在异常退出时可靠清理监控进程。 --- .github/workflows/test-process-result.yml | 2 +- benchmarks/benchmark_lib.sh | 68 +++++- .../aggregation/test_power_lifecycle.py | 221 ++++++++++++++++++ 3 files changed, 288 insertions(+), 3 deletions(-) create mode 100644 utils/agentic/aggregation/test_power_lifecycle.py diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml index 88506242c7..97169ae65b 100644 --- a/.github/workflows/test-process-result.yml +++ b/.github/workflows/test-process-result.yml @@ -54,4 +54,4 @@ jobs: - name: Run pytest run: | cd utils - python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py test_gb200_power_official_contract.py test_gb300_power_official_contract.py test_process_result.py -v + python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/test_power_adapter.py agentic/aggregation/test_power_lifecycle.py agentic/aggregation/test_process_agentic_result.py test_gb200_power_official_contract.py test_gb300_power_official_contract.py test_process_result.py -v diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index 8cc894940f..0a7ad5c97b 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -2170,10 +2170,43 @@ validate_required_agentic_server_metrics() { echo "Validated required AIPerf server metrics prefix '$required_prefix'" } -run_agentic_replay_and_write_outputs() { +run_agentic_replay_and_write_outputs() ( local result_dir="$1" local replay_rc local validation_rc + local power_rc=0 + local agentx_power_enabled=0 + local agentx_monitor_stopped=1 + + case "${ENABLE_AGENTX_POWER:-1}" in + 1|true|TRUE|yes|YES) + if [ "${IS_MULTINODE:-false}" != "true" ]; then + agentx_power_enabled=1 + fi + ;; + esac + + _stop_agentx_power_monitor() { + if [ "$agentx_monitor_stopped" = "0" ]; then + agentx_monitor_stopped=1 + stop_gpu_monitor + fi + } + + if [ "$agentx_power_enabled" = "1" ]; then + # AIPerf currently exports naive local datetimes while SMI emits the + # same host wall clock. Capture the launch-time offset so the adapter + # can attach it explicitly before normalizing the profiling window. + date +%z > "$result_dir/agentic_power_timezone_offset.txt" + start_gpu_monitor --output "$result_dir/gpu_metrics.csv" + agentx_monitor_stopped=0 + # This function runs in a subshell, so these handlers cannot replace + # launcher-owned traps. The stopped flag keeps explicit and signal/EXIT + # cleanup idempotent. + trap '_stop_agentx_power_monitor' EXIT + trap '_stop_agentx_power_monitor; exit 130' INT + trap '_stop_agentx_power_monitor; exit 143' TERM + fi echo "$REPLAY_CMD" > "$result_dir/benchmark_command.txt" @@ -2184,8 +2217,34 @@ run_agentic_replay_and_write_outputs() { set +x set -e + if [ "$agentx_power_enabled" = "1" ]; then + _stop_agentx_power_monitor + trap - EXIT INT TERM + fi + write_agentic_result_json "$result_dir" + if [ "$agentx_power_enabled" = "1" ]; then + local expected_num_gpus + local -a power_args + expected_num_gpus=$((${TP:-1} * ${PP_SIZE:-1} * ${PCP_SIZE:-1})) + power_args=( + --result-dir "$result_dir" + --agg-result "${AGENTIC_OUTPUT_DIR:-$INFMAX_CONTAINER_WORKSPACE}/$RESULT_FILENAME.json" + --expected-num-gpus "$expected_num_gpus" + ) + case "${REQUIRE_POWER:-0}" in + 1|true|TRUE|yes|YES) power_args+=(--require-power) ;; + esac + set +e + ( + cd "$INFMAX_CONTAINER_WORKSPACE" + "$AIPERF_PYTHON" -m utils.agentic.aggregation.power_adapter "${power_args[@]}" + ) + power_rc=$? + set -e + fi + "$AIPERF_PYTHON" "$AGENTIC_DIR/scripts/analyze_benchmark_distributions.py" \ "$result_dir/aiperf_artifacts" -o "$result_dir" 2>&1 || true @@ -2209,5 +2268,10 @@ run_agentic_replay_and_write_outputs() { return "$validation_rc" fi + if [ "$power_rc" -ne 0 ]; then + echo "ERROR: AgentX power validation failed after writing audit artifacts" >&2 + return "$power_rc" + fi + validate_required_agentic_server_metrics "$result_dir" -} +) diff --git a/utils/agentic/aggregation/test_power_lifecycle.py b/utils/agentic/aggregation/test_power_lifecycle.py new file mode 100644 index 0000000000..4877ab59e9 --- /dev/null +++ b/utils/agentic/aggregation/test_power_lifecycle.py @@ -0,0 +1,221 @@ +"""Shell-contract tests for the shared single-node AgentX power lifecycle.""" + +from __future__ import annotations + +import os +import re +import signal +import subprocess +import time +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[3] +BENCHMARK_LIB = REPO_ROOT / "benchmarks" / "benchmark_lib.sh" + + +def _run_lifecycle( + tmp_path: Path, + *, + replay_rc: int = 0, + is_multinode: bool = False, + enable_power: bool = True, + require_power: bool = False, +) -> subprocess.CompletedProcess[str]: + result_dir = tmp_path / "results" + result_dir.mkdir() + event_log = tmp_path / "events.log" + script = f""" +source {str(BENCHMARK_LIB)!r} +start_gpu_monitor() {{ + printf 'monitor-start:%s\n' "$*" >> {str(event_log)!r} + printf 'timestamp,index,power.draw [W]\n' > "$2" +}} +stop_gpu_monitor() {{ printf 'monitor-stop\n' >> {str(event_log)!r}; }} +fake_replay() {{ + printf 'replay\n' >> {str(event_log)!r} + return {replay_rc} +}} +write_agentic_result_json() {{ + printf 'aggregate\n' >> {str(event_log)!r} + printf '{{}}\n' > "$AGENTIC_OUTPUT_DIR/$RESULT_FILENAME.json" +}} +fake_python() {{ + case "$*" in + *utils.agentic.aggregation.power_adapter*) + printf 'adapter:%s\n' "$*" >> {str(event_log)!r} + ;; + *validate_agentic_result*) + printf 'validate\n' >> {str(event_log)!r} + ;; + *) + printf 'analyze\n' >> {str(event_log)!r} + ;; + esac + return 0 +}} +validate_required_agentic_server_metrics() {{ + printf 'server-metrics\n' >> {str(event_log)!r} +}} +trap 'printf "parent-exit\\n" >> {str(event_log)!r}' EXIT +REPLAY_CMD=fake_replay +AIPERF_PYTHON=fake_python +AGENTIC_DIR={str(tmp_path)!r} +INFMAX_CONTAINER_WORKSPACE={str(tmp_path)!r} +AGENTIC_OUTPUT_DIR={str(tmp_path)!r} +RESULT_FILENAME=agg_agentx +AIPERF_FAILED_REQUEST_THRESHOLD=0 +TP=3 +PP_SIZE=2 +PCP_SIZE=2 +IS_MULTINODE={'true' if is_multinode else 'false'} +ENABLE_AGENTX_POWER={'1' if enable_power else '0'} +REQUIRE_POWER={'1' if require_power else '0'} +set +e +run_agentic_replay_and_write_outputs {str(result_dir)!r} +rc=$? +exit "$rc" +""" + return subprocess.run( + ["bash", "-c", script], + env={ + **os.environ, + "PATH": "/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + }, + capture_output=True, + text=True, + check=False, + ) + + +def _events(tmp_path: Path) -> list[str]: + return (tmp_path / "events.log").read_text().splitlines() + + +@pytest.mark.parametrize( + ("replay_rc", "expected_rc"), + [(0, 0), (7, 7)], +) +def test_single_node_monitor_wraps_replay_and_stops_once( + tmp_path: Path, replay_rc: int, expected_rc: int +): + result = _run_lifecycle(tmp_path, replay_rc=replay_rc) + + assert result.returncode == expected_rc, result.stderr + events = _events(tmp_path) + assert events.count("monitor-stop") == 1 + assert events.index("monitor-start:--output " + str(tmp_path / "results/gpu_metrics.csv")) < events.index( + "replay" + ) + assert events.index("replay") < events.index("monitor-stop") + assert events.index("monitor-stop") < events.index("aggregate") + assert (tmp_path / "results/gpu_metrics.csv").is_file() + captured_offset = (tmp_path / "results/agentic_power_timezone_offset.txt").read_text().strip() + assert re.fullmatch(r"[+-]\d{4}", captured_offset) + assert events[-1] == "parent-exit" + + +def test_single_node_invokes_adapter_with_gpu_shape_and_strict_mode(tmp_path: Path): + result = _run_lifecycle(tmp_path, require_power=True) + + assert result.returncode == 0, result.stderr + adapter_event = next(event for event in _events(tmp_path) if event.startswith("adapter:")) + assert "--result-dir " + str(tmp_path / "results") in adapter_event + assert "--agg-result " + str(tmp_path / "agg_agentx.json") in adapter_event + assert "--expected-num-gpus 12" in adapter_event + assert "--require-power" in adapter_event + + +@pytest.mark.parametrize( + ("is_multinode", "enable_power"), + [(True, True), (False, False)], +) +def test_multinode_and_explicit_opt_out_skip_local_power( + tmp_path: Path, is_multinode: bool, enable_power: bool +): + result = _run_lifecycle( + tmp_path, + is_multinode=is_multinode, + enable_power=enable_power, + ) + + assert result.returncode == 0, result.stderr + events = _events(tmp_path) + assert not any(event.startswith("monitor-") for event in events) + assert not any(event.startswith("adapter:") for event in events) + + +def test_shared_lifecycle_installs_idempotent_signal_cleanup(): + benchmark_lib = BENCHMARK_LIB.read_text() + + assert "trap '_stop_agentx_power_monitor; exit 130' INT" in benchmark_lib + assert "trap '_stop_agentx_power_monitor; exit 143' TERM" in benchmark_lib + assert 'if [ "$agentx_monitor_stopped" = "0" ]' in benchmark_lib + + +def test_single_node_workflow_uploads_agentx_power_audit_artifacts(): + workflow = (REPO_ROOT / ".github/workflows/benchmark-tmpl.yml").read_text() + agentic_upload = workflow.split( + "- name: Upload agentic raw results", 1 + )[1].split("- name:", 1)[0] + + assert "results/**" in agentic_upload + assert "!results/**/gpu_metrics" not in agentic_upload + assert "!results/**/power_validation.json" not in agentic_upload + assert "!results/**/agentic_power_window.json" not in agentic_upload + + +@pytest.mark.parametrize( + ("sent_signal", "expected_rc"), + [(signal.SIGINT, 130), (signal.SIGTERM, 143)], +) +def test_signal_stops_monitor_once_without_replacing_parent_trap( + tmp_path: Path, sent_signal: signal.Signals, expected_rc: int +): + result_dir = tmp_path / "results" + result_dir.mkdir() + event_log = tmp_path / "events.log" + script = f""" +source {str(BENCHMARK_LIB)!r} +start_gpu_monitor() {{ + printf 'monitor-pid:%s\n' "${{BASHPID:-$$}}" >> {str(event_log)!r} +}} +stop_gpu_monitor() {{ printf 'monitor-stop\n' >> {str(event_log)!r}; }} +fake_replay() {{ sleep 30; }} +trap 'printf "parent-exit\\n" >> {str(event_log)!r}' EXIT +trap 'printf "parent-int\\n" >> {str(event_log)!r}; exit 130' INT +trap 'printf "parent-term\\n" >> {str(event_log)!r}; exit 143' TERM +REPLAY_CMD=fake_replay +ENABLE_AGENTX_POWER=1 +IS_MULTINODE=false +run_agentic_replay_and_write_outputs {str(result_dir)!r} +""" + proc = subprocess.Popen( + ["bash", "-c", script], + env={**os.environ, "PATH": "/usr/bin:/bin"}, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, + ) + monitor_pid = None + for _ in range(100): + if event_log.exists(): + first_event = event_log.read_text().splitlines()[0] + if first_event.startswith("monitor-pid:"): + monitor_pid = int(first_event.split(":", 1)[1]) + break + time.sleep(0.01) + assert monitor_pid is not None + + os.killpg(proc.pid, sent_signal) + _, stderr = proc.communicate(timeout=5) + + assert proc.returncode == expected_rc, stderr + events = _events(tmp_path) + assert events.count("monitor-stop") == 1 + expected_parent_event = "parent-int" if sent_signal == signal.SIGINT else "parent-term" + assert expected_parent_event in events + assert events[-1] == "parent-exit"