[Power] feat: measure GPU power for single-node AgentX runs / 为单机 AgentX 运行接入 GPU 功耗测量 #2601
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2170,10 +2170,43 @@ validate_required_agentic_server_metrics() { | |
| echo "Validated required AIPerf server metrics prefix '$required_prefix'" | ||
| } | ||
|
|
||
| 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 | ||
|
Comment on lines
+2181
to
+2185
There was a problem hiding this comment.
Enabling 1 Hz GPU monitoring by default changes every single-node AgentX benchmark's execution environment and can affect the reported performance, but this commit leaves AGENTS.md reference: AGENTS.md:L21-L21 Useful? React with 👍 / 👎. |
||
| ;; | ||
| 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 | ||
|
Comment on lines
+2206
to
+2207
There was a problem hiding this comment.
When the replay subprocess group receives SIGTERM, Bash can run this TERM handler and subsequently the installed EXIT handler without retaining the guard update between trap contexts, so Useful? React with 👍 / 👎. |
||
| 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" | ||
| ) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
|
Check warning on line 191 in utils/agentic/aggregation/power_adapter.py
|
||
|
Comment on lines
+165
to
+191
Contributor
There was a problem hiding this comment. 🟡 run_agentic_power() has two near-identical ~12-line blocks (lines 175-191 and 198-214) that each call _record_adapter_failure() inside the same try/except, print the same failure/adaptation messages, and return the same Extended reasoning...What the duplication looks like In
Both blocks do exactly the same four things in the same order:
The only thing that differs between the two call sites is the source of the Why this matters This isn't a correctness bug — both code paths behave identically today, and the tests ( Suggested fix Extract a small helper: def _fail_adapter(
*,
result_dir: Path,
agg_result: Path,
expected_num_gpus: int | None,
reasons: list[str],
require_power: bool,
) -> int:
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 0and call it from both the Proof of duplication (step-by-step)
|
||
|
|
||
| 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()) | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 The pull_request path filter for this new CI job lists power_adapter.py, process_agentic_result.py, and their test files, but omits the modules those files import transitively: utils/agentic/aggregation/request_metrics.py (extract_per_record_ints, load_aggregate, load_records), plus aggregation_common.py, server_metrics.py, and server_log_metrics.py. A change to any of these transitive dependencies (e.g. tweaking extract_per_record_ints filtering in request_metrics.py) would silently skip the power/agentic test suite this PR just wired in. Consider adding those paths, or switching to a glob like 'utils/agentic/aggregation/**'.
Extended reasoning...
What the bug is:
.github/workflows/test-process-result.ymlgained six new path-filter entries in this PR to trigger the newly-wired agentic/power test suite onpull_request. The filter listspower_adapter.py,test_power_adapter.py,test_power_lifecycle.py,process_agentic_result.py,test_process_agentic_result.py, andpytest.ini— but it does not list the modules those production files import.The code path:
utils/agentic/aggregation/power_adapter.py(added by this PR) containsfrom .request_metrics import extract_per_record_ints, load_aggregate, load_records.utils/agentic/aggregation/process_agentic_result.py(an existing file whose tests are newly wired in here) imports from.aggregation_common,.request_metrics,.server_log_metrics, and.server_metrics;request_metrics.pyitself pulls intrace_metadata.pytransitively. None ofrequest_metrics.py,aggregation_common.py,server_metrics.py, orserver_log_metrics.pyappear anywhere in the path filter (lines 14-25), and a repo-wide grep across.github/workflowsconfirms no other workflow covers them either.Why nothing else catches this: the path filter is an allowlist gate on
pull_requesttriggers — GitHub Actions simply won't start the job unless a changed file matches one of the listed globs. There's no fallback trigger (like a periodic or push-to-main run) configured in this workflow that would catch a miss here on a PR that only touches, say,request_metrics.py.Impact: a PR that edits
request_metrics.py— for example changing the filtering logic inextract_per_record_ints(used directly by the newpower_adapter.build_power_window) or adjustingload_records— would not triggertest-process-result.ymlat all. The 195+ tests covering the agentic/power suite this PR just wired in would silently not run on that PR, even though the change could breakpower_adapter.py's window derivation orprocess_agentic_result.py's aggregation.Step-by-step proof:
utils/agentic/aggregation/request_metrics.py, e.g. tightening the token-count validation inextract_per_record_ints.pull_request.pathsfilter intest-process-result.ymlagainst the changed file list:['utils/agentic/aggregation/request_metrics.py'].power_adapter.py,test_power_adapter.py, etc. are file-specific entries, not directory globs.extract_per_record_intschange silently breakstest_power_adapter.py'stest_build_power_window_rejects_ambiguous_inputscase, that regression ships to main undetected until some later, unrelated PR happens to touch one of the explicitly-listed files and the job finally runs against stale-broken code.Fix: add
utils/agentic/aggregation/request_metrics.py,aggregation_common.py,server_metrics.py, andserver_log_metrics.pyto the path list, or replace the file-by-file enumeration with a directory glob such asutils/agentic/aggregation/**so future additions to that package don't require another manual path-filter update.