diff --git a/.github/AGENT_OPERATIONS.md b/.github/AGENT_OPERATIONS.md index 86fadeb6c3..8710529af6 100644 --- a/.github/AGENT_OPERATIONS.md +++ b/.github/AGENT_OPERATIONS.md @@ -68,10 +68,16 @@ Eval selection, the `--no-evals` / `--evals-only` / `--all-evals` flags, and cha ## Power telemetry -Multinode srt-slurm results, and single-node fixed-sequence and AgentX results with a retained native telemetry package, may include `power_valid`, `avg_power_w`, `avg_total_gpu_power_w`, `total_gpu_energy_j`, and joules per query/input/output/total token. Invalid telemetry records `power_valid: 0` without energy metrics and fails only with `REQUIRE_POWER=1`. +Multinode srt-slurm results, and single-node fixed-sequence and AgentX results with a retained native telemetry package, may include `power_valid`, `avg_power_w`, `avg_total_gpu_power_w`, `total_gpu_energy_j`, and joules per query/input/output/total token. Invalid telemetry records `power_valid: 0` without energy metrics and fails only with `REQUIRE_POWER=1`, or always on AgentX lanes whose launcher validates power. 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. +A multinode master row with `cpu-power: true` beside `power: true` adds srt-slurm's Grace CPU leg. The binder merges `telemetry.cpu_power_exporter` into the bound recipe from the cluster's `srt-slurm.cpu-power-exporter` record in `inferencex-e2e/configs/runners.yaml` (ACPI on `gb200-nv` and `gb300-nv`), the same way it merges the DCGM block, so recipe fragments and `additional-settings` never carry it; `make setup` pins the exporter release from `CPU_POWER_EXPORTER_RELEASE` in `infx/launch/drivers/srt/checkout.py` for those jobs only, and the workflow exports `CPU_POWER=1` once for both the launcher and the result processor. + +The package then carries `power/cpu/samples.csv` in srt-slurm's wide format (one row per scrape, host, and Grace socket; two sockets per worker host in the manifest topology), and the aggregate adds `cpu_power_valid`, `avg_cpu_socket_power_w`, `avg_total_cpu_power_w`, and `total_cpu_energy_j` from the `CPU:cpuSidePowerUsageW` series, plus `avg_total_cpu_rail_power_w` / `total_cpu_rail_energy_j` (ACPI `CPU Power Socket N`, cores only) and `avg_total_cpu_sysio_power_w` / `total_cpu_sysio_energy_j` (ACPI `SysIO Power Socket N`), each pair only when every socket reports that rail without gaps. The meters nest ([NVIDIA Grace power guide](https://docs.nvidia.com/dccpu/grace-perf-tuning-guide/power-thermals.html)): `Grace Power Socket N` is the whole socket (CPU rail, SysIO rail, LPDDR5X, and regulator loss) and the two rails are a breakdown inside it, so never add them to or subtract them from the total; a GB300 socket reads about 97 W while its rails sum to about 55 W. The DRAM rail is blank on the CI NVL72 firmware. srt-slurm v2.59.0+ writes NVML superchip module power as `module_w` (CPU samples schema v3) when its exporter can load NVML (the aarch64 glibc build, srt-slurm #609), and the aggregate then publishes `avg_total_module_power_w` / `total_module_energy_j` under the same every-socket, gap-free rule; the module includes both GPUs, so never add it to the CPU or GPU totals. Every series is integrated over the bound formal GPU window with the GPU leg's trapezoid, boundary interpolation, and 3.0 s gap limit; the sidecar's `cpu` block, mirrored as `power_audit.cpu`, records `sensor_kind: grace_socket`, `source`, socket and row counts, and reason codes. + +The CPU verdict is independent of `power_valid`: a failed leg records `cpu_power_valid: 0` with no CPU keys and leaves every GPU field unchanged. It fails the job only when power is required (`REQUIRE_POWER=1`, or always on AgentX lanes whose launcher validates power) and the row set `cpu-power`; missing CPU artifacts are then a verdict (`cpu_artifacts_missing`) rather than silence. Other reason codes: `cpu_samples_missing`, `cpu_samples_header_mismatch`, `cpu_samples_malformed`, `cpu_socket_count_mismatch` (also when a host reports only the DCGM CPU rail), `cpu_sample_gap_exceeded`, `cpu_window_not_bracketed`, `cpu_window_unavailable`, and the shared `non_finite_power_metric`. + Every power result, valid or invalid, 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. A multinode master row with `power: true` enables official energy collection through the shared telemetry fragment; native single-node throughput and AgentX jobs enable telemetry through their bound recipe. Native single-node jobs inherit the exporter image, labels and metrics from the cluster's `default_gpu_exporter` in `inferencex-e2e/configs/runners.yaml`: DCGM on NVIDIA, the `kind: custom` AMD device-metrics-exporter on AMD. Multi-node power rows keep their shared DCGM telemetry block. Eval-only jobs omit collection. The Git submodule pointer at `inferencex-e2e/utils/srt-slurm` is the source of truth for every srt-slurm job, including TileRT. CI derives `POWER_PRODUCER_SHA` from the launcher stamp. The aggregate-power and AgentX power tests validate telemetry and provenance. These local tests do not prove hardware power collection. Eligible `dynamo-sglang` dcgm-power lanes are validated. diff --git a/.github/workflows/benchmark-multinode-tmpl.yml b/.github/workflows/benchmark-multinode-tmpl.yml index ac4347b3fe..e0b4d79492 100644 --- a/.github/workflows/benchmark-multinode-tmpl.yml +++ b/.github/workflows/benchmark-multinode-tmpl.yml @@ -146,6 +146,7 @@ env: EVAL_CONC: ${{ inputs.eval-conc }} EVAL_LIMIT: ${{ inputs.eval-limit }} POWER: ${{ fromJSON(inputs.config).power && '1' || '0' }} + CPU_POWER: ${{ fromJSON(inputs.config)['cpu-power'] && '1' || '0' }} REQUIRE_POWER: ${{ inputs.require-power && '1' || '0' }} POWER_PRODUCER_SHA: ${{ inputs.power-producer-sha }} # Agentic-coding env. Fixed-seq-len jobs leave these empty. diff --git a/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml b/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml index 1511ea8661..d0526b16b3 100644 --- a/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml +++ b/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml @@ -10,11 +10,11 @@ services: - name: etcd type: etcd placement: - node: dedicated + node: head - name: nats type: nats placement: - node: dedicated + node: head dynamo: install: true @@ -22,6 +22,8 @@ dynamo: wheel: "1.5.0.dev20260917" frontend: type: dynamo + placement: + node: dedicated enable_multiple_frontends: true num_additional_frontends: 4 nginx_container: nginx diff --git a/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml b/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml index 259798c14b..3506fab0cf 100644 --- a/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml +++ b/inferencex-e2e/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-8p1d-dep4-dep16-stp.yaml @@ -9,11 +9,11 @@ services: - name: etcd type: etcd placement: - node: dedicated + node: head - name: nats type: nats placement: - node: dedicated + node: head dynamo: install: true @@ -21,6 +21,8 @@ dynamo: wheel: "1.5.0.dev20260917" frontend: type: dynamo + placement: + node: dedicated enable_multiple_frontends: true num_additional_frontends: 4 nginx_container: nginx diff --git a/inferencex-e2e/configs/nvidia-master.yaml b/inferencex-e2e/configs/nvidia-master.yaml index 2ff820f5b9..1d94560779 100644 --- a/inferencex-e2e/configs/nvidia-master.yaml +++ b/inferencex-e2e/configs/nvidia-master.yaml @@ -3726,12 +3726,14 @@ qwen3.5-fp8-gb200-dynamo-sglang: fixed-seq-len: - isl: 8192 osl: 1024 + require-power: true search-space: # 1P1D STP: TP4 prefill + TP4 decode (pure tensor parallel). 2 nodes (1+1). - spec-decoding: "none" conc-list: [1, 2, 4, 8, 16, 32, 64, 128] srt-recipe: disagg-1p1d-tp4-tp4-stp.yaml power: true + cpu-power: true prefill: num-worker: 1 tp: 4 @@ -3747,6 +3749,7 @@ qwen3.5-fp8-gb200-dynamo-sglang: conc-list: [1024] srt-recipe: disagg-4p1d-dep4-dep16-stp.yaml power: true + cpu-power: true prefill: num-worker: 4 tp: 4 @@ -3761,6 +3764,8 @@ qwen3.5-fp8-gb200-dynamo-sglang: - spec-decoding: "none" conc-list: [2048, 4096] srt-recipe: disagg-8p1d-dep4-dep16-stp.yaml + power: true + cpu-power: true prefill: num-worker: 8 tp: 4 @@ -6198,6 +6203,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-dcp16-agg: num-nodes: 4 srt-recipe: agg-dcp16-dspark4-maxseq2-mooncake.yaml power: true + cpu-power: true worker: num-worker: 1 tp: 16 @@ -6225,6 +6231,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-mooncake-dcp16-agg: num-nodes: 4 srt-recipe: agg-dcp16-nospec-mooncake.yaml power: true + cpu-power: true worker: num-worker: 1 tp: 16 @@ -6361,6 +6368,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2: num-nodes: 4 srt-recipe: agg-tp8pp2-variants.yaml:override_tp8pp2_c16 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -6375,6 +6383,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2: num-nodes: 4 srt-recipe: agg-tp8pp2-variants.yaml:override_tp8pp2_c32 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -6388,6 +6397,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2: num-nodes: 4 srt-recipe: agg-tp8pp2-variants.yaml:override_tp8pp2_c48 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -6401,6 +6411,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2: num-nodes: 4 srt-recipe: agg-tp8pp2-variants.yaml:override_tp8pp2_c72 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -6414,6 +6425,7 @@ kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2: num-nodes: 4 srt-recipe: agg-tp8pp2-variants.yaml:override_tp8pp2_c96 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -6679,12 +6691,14 @@ qwen3.5-fp8-gb300-dynamo-sglang: fixed-seq-len: - isl: 8192 osl: 1024 + require-power: true search-space: # 1P1D STP: TP4 prefill + TP4 decode (pure tensor parallel). 2 nodes (1+1). - spec-decoding: "none" conc-list: [1, 2, 4, 8, 16, 32, 64, 128] srt-recipe: disagg-1p1d-tp4-tp4-stp.yaml power: true + cpu-power: true prefill: num-worker: 1 tp: 4 @@ -6700,6 +6714,7 @@ qwen3.5-fp8-gb300-dynamo-sglang: conc-list: [1024] srt-recipe: disagg-4p1d-dep4-dep16-stp.yaml power: true + cpu-power: true prefill: num-worker: 4 tp: 4 @@ -6714,6 +6729,8 @@ qwen3.5-fp8-gb300-dynamo-sglang: - spec-decoding: "none" conc-list: [2048, 4096] srt-recipe: disagg-8p1d-dep4-dep16-stp.yaml + power: true + cpu-power: true prefill: num-worker: 8 tp: 4 @@ -7869,6 +7886,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-disagg: conc-list: [48, 56] srt-recipe: disagg-variants.yaml:override_1p1d_dcp8_dcp8_dspark4 power: true + cpu-power: true prefill: num-worker: 1 tp: 8 @@ -7882,6 +7900,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-disagg: conc-list: [48, 64] srt-recipe: disagg-variants.yaml:override_1p2d_dcp8_dcp8_dspark4 power: true + cpu-power: true prefill: num-worker: 1 tp: 8 @@ -7895,6 +7914,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-disagg: conc-list: [1] srt-recipe: disagg-variants.yaml:override_1p3d_dcp8_dcp8_dspark7 power: true + cpu-power: true prefill: num-worker: 1 tp: 8 @@ -7908,6 +7928,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-disagg: conc-list: [48] srt-recipe: disagg-variants.yaml:override_1p3d_dcp8_dcp8_dspark4 power: true + cpu-power: true prefill: num-worker: 1 tp: 8 @@ -7937,6 +7958,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-agg: num-nodes: 2 srt-recipe: agg-variants.yaml:override_dcp8_dspark7_maxseq2 power: true + cpu-power: true worker: num-worker: 1 tp: 8 @@ -7950,6 +7972,7 @@ kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-agg: num-nodes: 2 srt-recipe: agg-variants.yaml:override_dcp8_dspark4 power: true + cpu-power: true worker: num-worker: 1 tp: 8 diff --git a/inferencex-e2e/configs/runners.yaml b/inferencex-e2e/configs/runners.yaml index 4d25914d7e..52d2b63057 100644 --- a/inferencex-e2e/configs/runners.yaml +++ b/inferencex-e2e/configs/runners.yaml @@ -622,6 +622,7 @@ clusters: aiperf-cache: /aiperf_mmap_cache hf-hub-cache: /hf_hub_cache power-exporter-port: 9401 + cpu-power-exporter: {source: acpi} gb300-nv: gpus-per-node: 4 available-cpu-dram-mib: 920_000 @@ -680,6 +681,7 @@ clusters: hf-hub-cache: /hf_hub_cache dynamo-wheels: /configs/dynamo-wheels power-exporter-port: 19401 + cpu-power-exporter: {source: acpi} mi300x-amd: gpus-per-node: 8 available-cpu-dram-mib: 1_547_820 diff --git a/inferencex-e2e/infx/clusters/slurm.py b/inferencex-e2e/infx/clusters/slurm.py index c7c670f00c..4731be5171 100644 --- a/inferencex-e2e/infx/clusters/slurm.py +++ b/inferencex-e2e/infx/clusters/slurm.py @@ -173,6 +173,16 @@ def rendered(self) -> dict[str, str | None]: return self.model_dump(by_alias=True) +class CpuPowerExporter(Record): + """The Grace CPU power exporter srt-slurm scrapes on this cluster's compute trays. + + A point that selects ``cpu-power`` binds ``telemetry.cpu_power_exporter`` with this source. + Only ACPI serves the Grace socket meter the CPU leg integrates. + """ + + source: Literal["acpi"] + + class SrtSlurmSettings(Record): """Cluster-owned srtslurm.yaml facts; job-specific values are added by the driver.""" @@ -203,6 +213,7 @@ class SrtSlurmSettings(Record): mounts: dict[str, str] = Field(default_factory=dict) workspace_mounts: dict[str, str] = Field(default_factory=dict, alias="workspace-mounts") power_exporter_port: int | None = Field(default=None, alias="power-exporter-port", gt=0) + cpu_power_exporter: CpuPowerExporter | None = Field(default=None, alias="cpu-power-exporter") extra: dict[str, Any] = Field(default_factory=dict) @field_validator("workspace_mounts") diff --git a/inferencex-e2e/infx/launch/artifacts.py b/inferencex-e2e/infx/launch/artifacts.py index 57505e4562..08f2bd3729 100644 --- a/inferencex-e2e/infx/launch/artifacts.py +++ b/inferencex-e2e/infx/launch/artifacts.py @@ -153,6 +153,7 @@ def collect_agentic_power_results( concurrencies: Sequence[int], *, results_python: str | None, + expected_cpu: bool = False, ) -> int: """Stage the AgentX power audit inputs and validate each concurrency's power window. @@ -175,7 +176,7 @@ def collect_agentic_power_results( if step := validate_agentic_power( logs_dir, workspace, result_filename, producer_sha, concurrencies, - results_python=results_python, require_power=True, + results_python=results_python, require_power=True, expected_cpu=expected_cpu, ): # fmt: skip rc = step return rc @@ -190,10 +191,12 @@ def validate_agentic_power( *, results_python: str | None, require_power: bool, + expected_cpu: bool = False, ) -> int: """Run the AgentX power adapter for every concurrency; return the last failing code. - With ``require_power`` a concurrency without a valid power window fails. + With ``require_power`` a concurrency without a valid power window fails, and with + ``expected_cpu`` one without a valid Grace CPU leg fails too. """ logs_dir = logs_dir.resolve() workspace = workspace.resolve() @@ -207,7 +210,10 @@ def validate_agentic_power( results_python, workspace, logs_dir, logs_dir / "agentic" / f"conc_{concurrency}", workspace / f"{result_filename}_conc{concurrency}.json", - producer_sha, require_power=require_power, + producer_sha, + "--audit-source", f"LOGS/agentic/conc_{concurrency}/power_validation.json", + *(["--expected-cpu-power"] if expected_cpu else []), + require_power=require_power, ): # fmt: skip rc = step return rc diff --git a/inferencex-e2e/infx/launch/drivers/srt/__init__.py b/inferencex-e2e/infx/launch/drivers/srt/__init__.py index cc89b4b4c0..9ece0ff26c 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/__init__.py +++ b/inferencex-e2e/infx/launch/drivers/srt/__init__.py @@ -163,14 +163,16 @@ def run_multinode(launch: Launch) -> int: ) install_srtctl(run, checkout, python=system_python) config.write_lane_config(run, lane, checkout, decision, model_paths) - if rc := run_setup(run, checkout): + if rc := run_setup(run, checkout, cpu_power_exporter=decision.cpu): return rc infmax = compute_workspace(run, checkout, shared=shared) run.env["INFMAX_WORKSPACE"] = str(infmax) recipe = str(checkout.root / BOUND_RECIPE) - power_port, client_env = config.binder_inputs(run.cluster, run.srt, lane, request, decision) - if rc := submit.bind_recipe(run, checkout, staged, recipe, power_port, client_env): + power_port, cpu_source, client_env = config.binder_inputs( + run.cluster, run.srt, lane, request, decision + ) + if rc := submit.bind_recipe(run, checkout, staged, recipe, power_port, cpu_source, client_env): return rc prepare_recipe(Path(recipe), srtctl_job_name(request.runner_name), run.srt.dist_timeout_s) arguments = submit.multinode_arguments(run, lane, recipe, overrides, preflight=preflight) diff --git a/inferencex-e2e/infx/launch/drivers/srt/checkout.py b/inferencex-e2e/infx/launch/drivers/srt/checkout.py index 822ef43e3b..f2c81c3f46 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/checkout.py +++ b/inferencex-e2e/infx/launch/drivers/srt/checkout.py @@ -18,6 +18,9 @@ from infx.launch.drivers.srt.run import SrtRun, require SUBMODULE = Path("utils/srt-slurm") +# The cpu-power-exporter release asset ``make setup`` installs for a job that scrapes Grace CPU +# power; it ships with the srt-slurm release the submodule pins, so bump both together. +CPU_POWER_EXPORTER_RELEASE = "v2.51.1" PATCHES = Path("runners/srt-slurm/patches") SETUP_LOG = "srt-setup.log" SETUP_ATTEMPTS = 5 @@ -173,12 +176,17 @@ def _discard_corrupt_archives(configs: Path) -> bool: return discarded -def run_setup(run: SrtRun, checkout: Checkout) -> int: - """Run ``make setup`` in the checkout, logging to SETUP_LOG; return its exit code.""" +def run_setup(run: SrtRun, checkout: Checkout, *, cpu_power_exporter: bool = False) -> int: + """Run ``make setup`` in the checkout, logging to SETUP_LOG; return its exit code. + + ``cpu_power_exporter`` pins the cpu-power-exporter release, which makes its download fatal. + """ log = run.workspace / SETUP_LOG for attempt in range(1, SETUP_ATTEMPTS + 1): print(f"Setting up srt-slurm, attempt {attempt} (details: {SETUP_LOG})", flush=True) argv = ["make", "setup", f"ARCH={run.cluster.arch}"] + if cpu_power_exporter: + argv.append(f"CPU_POWER_EXPORTER_RELEASE={CPU_POWER_EXPORTER_RELEASE}") rc = _run_logged(argv, log, env=run.env, cwd=checkout.root) if rc == 0: print("srt-slurm setup complete", flush=True) diff --git a/inferencex-e2e/infx/launch/drivers/srt/collect.py b/inferencex-e2e/infx/launch/drivers/srt/collect.py index 7868b2e65d..cc2b96993e 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/collect.py +++ b/inferencex-e2e/infx/launch/drivers/srt/collect.py @@ -181,11 +181,21 @@ def stage_logs_on_exit() -> None: python = request.inferencex_results_python if power.agentx: power_rc = collect_agentic_power_results( - status, job.id, logs, infmax, *audit, results_python=python + status, + job.id, + logs, + infmax, + *audit, + results_python=python, + expected_cpu=power.cpu, ) else: power_rc = validate_agentic_power( - logs, *audit, results_python=python, require_power=request.require_power + logs, + *audit, + results_python=python, + require_power=request.require_power, + expected_cpu=power.cpu, ) if power_rc: print( diff --git a/inferencex-e2e/infx/launch/drivers/srt/config.py b/inferencex-e2e/infx/launch/drivers/srt/config.py index 849acb4131..837fd25845 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/config.py +++ b/inferencex-e2e/infx/launch/drivers/srt/config.py @@ -248,21 +248,33 @@ def binder_inputs( lane: SrtLane, request: SrtRequest, decision: PowerDecision, -) -> tuple[int | None, dict[str, str]]: - """The DCGM exporter port and AgentX client env a multi-node recipe is bound with. +) -> tuple[int | None, str | None, dict[str, str]]: + """The DCGM exporter port, CPU exporter source and AgentX client env a recipe is bound with. - The client marks power windows exactly when the launcher collects the point's AgentX power, + The CPU source is the cluster's Grace exporter when the point selected cpu-power. The + client marks power windows exactly when the launcher collects the point's AgentX power, and fails invalid power when the workflow or the lane's power rule requires it. """ port = srt.power_exporter_port if decision.dcgm else None + cpu_source = None + if decision.cpu: + if srt.cpu_power_exporter is None: + raise LaunchError( + f"cluster {cluster.id!r} declares no srt-slurm.cpu-power-exporter for cpu-power" + ) + cpu_source = srt.cpu_power_exporter.source if not request.is_agentic: - return port, {} - return port, { - **agentic_client_env(cluster, srt, lane, request), - "IS_MULTINODE": str(request.is_multinode).lower(), - "ENABLE_AGENTX_POWER": "1" if decision.agentx or decision.adapter else "0", - "REQUIRE_POWER": "1" if decision.require or request.require_power else "0", - } + return port, cpu_source, {} + return ( + port, + cpu_source, + { + **agentic_client_env(cluster, srt, lane, request), + "IS_MULTINODE": str(request.is_multinode).lower(), + "ENABLE_AGENTX_POWER": "1" if decision.agentx or decision.adapter else "0", + "REQUIRE_POWER": "1" if decision.require or request.require_power else "0", + }, + ) def write_lane_config( diff --git a/inferencex-e2e/infx/launch/drivers/srt/power.py b/inferencex-e2e/infx/launch/drivers/srt/power.py index 61d2270970..54079d0767 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/power.py +++ b/inferencex-e2e/infx/launch/drivers/srt/power.py @@ -138,6 +138,8 @@ class PowerDecision: agentx: bool adapter: bool = False require: bool = False + # The point selected cpu-power: the cluster's Grace CPU exporter is bound and expected. + cpu: bool = False NO_POWER = PowerDecision(dcgm=False, agentx=False) @@ -156,7 +158,11 @@ def decide_power(cluster_id: str, path: LaunchPath, request: MultiNodeRequest) - for rule in lane.rules: if rule.when(request): return PowerDecision( - dcgm=True, agentx=rule.agentx, adapter=rule.adapter, require=rule.require + dcgm=True, + agentx=rule.agentx, + adapter=rule.adapter, + require=rule.require, + cpu=request.cpu_power, ) message = lane.agentic_error if request.is_agentic and lane.agentic_error else lane.error raise PowerPolicyError(message) diff --git a/inferencex-e2e/infx/launch/drivers/srt/submit.py b/inferencex-e2e/infx/launch/drivers/srt/submit.py index 0b20fe9fb9..26e574eb52 100644 --- a/inferencex-e2e/infx/launch/drivers/srt/submit.py +++ b/inferencex-e2e/infx/launch/drivers/srt/submit.py @@ -84,12 +84,16 @@ def bind_recipe( staged: str, output: str, power_port: int | None, + cpu_power_source: str | None, client_env: Mapping[str, str], ) -> int: - """Compose and bind the staged fragment into ``output``, with telemetry if ``power_port``.""" + """Compose and bind the staged fragment into ``output``, with telemetry if ``power_port`` + and the Grace CPU exporter if ``cpu_power_source``.""" bind = [str(checkout.venv / "bin/python"), "-m", "infx.srt_slurm.workload", staged, output] if power_port is not None: bind += ["--power-port", str(power_port)] + if cpu_power_source is not None: + bind += ["--cpu-power-source", cpu_power_source] for name, value in client_env.items(): bind += ["--client-env", f"{name}={value}"] bind += _fabric(run) diff --git a/inferencex-e2e/infx/launch/request.py b/inferencex-e2e/infx/launch/request.py index 88150eaa61..ebb1b46c70 100644 --- a/inferencex-e2e/infx/launch/request.py +++ b/inferencex-e2e/infx/launch/request.py @@ -150,6 +150,7 @@ class MultiNodeRequest(SrtRequest): srt_recipe: str = Field(alias="SRT_RECIPE") eval_srt_recipe: str | None = Field(None, alias="EVAL_SRT_RECIPE") power: OneFlag = Field(False, alias="POWER") + cpu_power: OneFlag = Field(False, alias="CPU_POWER") class SingleNodeRequest(SrtRequest): diff --git a/inferencex-e2e/infx/matrix/fingerprint.py b/inferencex-e2e/infx/matrix/fingerprint.py index f29cfc6424..982916e883 100644 --- a/inferencex-e2e/infx/matrix/fingerprint.py +++ b/inferencex-e2e/infx/matrix/fingerprint.py @@ -25,6 +25,7 @@ ("benchmark", "concurrencies"), ("benchmark", "env", "CONC"), ("telemetry", "dcgm_exporter", "port"), + ("telemetry", "cpu_power_exporter"), ) EXPORTER_PORT = 0 diff --git a/inferencex-e2e/infx/matrix/generate.py b/inferencex-e2e/infx/matrix/generate.py index 5865039af3..4ef22e359a 100644 --- a/inferencex-e2e/infx/matrix/generate.py +++ b/inferencex-e2e/infx/matrix/generate.py @@ -1074,6 +1074,8 @@ def _fixed_sequence_entries( entry.update(srt_recipe_fields(config, benchmark)) if benchmark.get(Fields.POWER.value): entry[Fields.POWER.value] = True + if benchmark.get(Fields.CPU_POWER.value): + entry[Fields.CPU_POWER.value] = True entry.update( { Fields.EXP_NAME.value: f"{model_code}_{seq_len_to_str(isl, osl)}", @@ -1197,6 +1199,8 @@ def _agentic_entries( entry.update(srt_recipe_fields(config, benchmark)) if benchmark.get(Fields.POWER.value): entry[Fields.POWER.value] = True + if benchmark.get(Fields.CPU_POWER.value): + entry[Fields.CPU_POWER.value] = True entry.update( { Fields.KV_OFFLOADING.value: kv_offloading, diff --git a/inferencex-e2e/infx/matrix/validation.py b/inferencex-e2e/infx/matrix/validation.py index 1ae17ab77f..01c118d0bb 100644 --- a/inferencex-e2e/infx/matrix/validation.py +++ b/inferencex-e2e/infx/matrix/validation.py @@ -83,6 +83,7 @@ class Fields(Enum): DURATION = "duration" REQUIRE_POWER = "require-power" POWER = "power" + CPU_POWER = "cpu-power" CONC = "conc" MAX_MODEL_LEN = "max-model-len" @@ -288,6 +289,7 @@ class MultiNodeMatrixEntry(BaseModel): osl: int require_power: bool = Field(default=False, alias=Fields.REQUIRE_POWER.value, strict=True) power: bool = Field(default=False, strict=True) + cpu_power: bool = Field(default=False, alias=Fields.CPU_POWER.value, strict=True) prefill: WorkerConfig decode: WorkerConfig conc: list[int] @@ -389,6 +391,7 @@ class MultiNodeAgenticMatrixEntry(BaseModel): ) node_count: int = Field(alias=Fields.NODE_COUNT.value, gt=0, strict=True) power: bool = Field(default=False, strict=True) + cpu_power: bool = Field(default=False, alias=Fields.CPU_POWER.value, strict=True) prefill: WorkerConfig decode: WorkerConfig conc: list[int] @@ -604,6 +607,13 @@ def validate_single_node_topology(self) -> Self: return _validate_tp_context_topology(self) +def _cpu_power_rides_on_power(self: Any) -> Any: + """``cpu-power`` adds the cluster's Grace CPU exporter to a point that measures power.""" + if self.cpu_power and not self.power: + raise ValueError(f"{Fields.CPU_POWER.value} requires {Fields.POWER.value} on the entry") + return self + + class MultiNodeSearchSpaceEntry(BaseModel): """Multinode search space configuration.""" @@ -620,6 +630,7 @@ class MultiNodeSearchSpaceEntry(BaseModel): default=None, alias=Fields.EVAL_SRT_RECIPE.value ) power: bool = Field(default=False, strict=True) + cpu_power: bool = Field(default=False, alias=Fields.CPU_POWER.value, strict=True) num_nodes: int | None = Field(default=None, alias=Fields.NUM_NODES.value, gt=0, strict=True) router: ComponentMetadata | None = None kv_p2p_transfer: str | None = Field( @@ -629,6 +640,10 @@ class MultiNodeSearchSpaceEntry(BaseModel): conc_end: int | None = Field(default=None, alias=Fields.CONC_END.value) conc_list: list[int] | None = Field(default=None, alias=Fields.CONC_LIST.value) + @model_validator(mode="after") + def validate_cpu_power(self) -> Self: + return _cpu_power_rides_on_power(self) + @model_validator(mode="after") def validate_conc_fields(self) -> Self: return _validate_conc_fields(self) @@ -682,6 +697,7 @@ class AgenticCodingSearchSpaceEntry(BaseModel): default=None, alias=Fields.EVAL_SRT_RECIPE.value ) power: bool = Field(default=False, strict=True) + cpu_power: bool = Field(default=False, alias=Fields.CPU_POWER.value, strict=True) pp: int = Field(default=1, gt=0, strict=True) dcp_size: int = Field(default=1, alias=Fields.DCP_SIZE.value, gt=0, strict=True) pcp_size: int = Field(default=1, alias=Fields.PCP_SIZE.value, gt=0, strict=True) @@ -745,6 +761,7 @@ def validate_topology_fields(self) -> Self: f"Only multinode entries measure DCGM power; remove {Fields.POWER.value}" ) _validate_tp_context_topology(self) + _cpu_power_rides_on_power(self) if has_aggregate_worker or has_complete_multinode: explicitly_single_node_fields = { "pp", diff --git a/inferencex-e2e/infx/results/agentic/power_adapter.py b/inferencex-e2e/infx/results/agentic/power_adapter.py index 9973c0ed7a..0cb4cca17f 100644 --- a/inferencex-e2e/infx/results/agentic/power_adapter.py +++ b/inferencex-e2e/infx/results/agentic/power_adapter.py @@ -355,6 +355,7 @@ def run_native_agentic_power( logs_root: Path, expected_producer_sha: str, require_power: bool = False, + expected_cpu: bool = False, audit_source: str | None = None, validation_result: Path | None = None, ) -> int: @@ -415,6 +416,7 @@ def run_native_agentic_power( logs_root=logs_root, validation_result=validation_result, require_power=require_power, + expected_cpu=expected_cpu, ) if audit_source is not None: try: @@ -447,6 +449,7 @@ def main() -> int: action="store_true", default=os.environ.get("REQUIRE_POWER", "").lower() in {"1", "true", "yes"}, ) + parser.add_argument("--expected-cpu-power", action="store_true") args = parser.parse_args() if args.multinode_contract_missing: if args.agg_result is None: @@ -499,6 +502,7 @@ def main() -> int: logs_root=args.logs_root, expected_producer_sha=args.expected_producer_sha, require_power=args.require_power, + expected_cpu=args.expected_cpu_power, audit_source=args.audit_source, validation_result=args.validation_result, ) diff --git a/inferencex-e2e/infx/results/fixed_sequence.py b/inferencex-e2e/infx/results/fixed_sequence.py index 68fa62b6f7..53ac1ce670 100644 --- a/inferencex-e2e/infx/results/fixed_sequence.py +++ b/inferencex-e2e/infx/results/fixed_sequence.py @@ -279,6 +279,7 @@ def aggregate_power_result( aggregate_gpus=aggregate_gpus, expected_producer_sha=env.get("POWER_PRODUCER_SHA") or None, logs_root=Path(env.get("POWER_RESULT_ROOT", "LOGS")), + expected_cpu=env.get("CPU_POWER") == "1", validation_result=validation_path, require_power=require_power, ) diff --git a/inferencex-e2e/infx/results/power/__init__.py b/inferencex-e2e/infx/results/power/__init__.py index 6a74fc8dd4..a765c894e4 100644 --- a/inferencex-e2e/infx/results/power/__init__.py +++ b/inferencex-e2e/infx/results/power/__init__.py @@ -30,6 +30,26 @@ "decode_joules_per_output_token", ) ALL_POWER_METRIC_KEYS = WHOLE_METRIC_KEYS + ROLE_METRIC_KEYS +# Additive metrics from the multinode cpu/ sub-package: the Grace socket, its +# ACPI rails and the NVML module rail. They share power metric schema version 2 +# and are governed by cpu_power_valid, not power_valid. +CPU_METRIC_KEYS = ( + "avg_cpu_socket_power_w", + "avg_total_cpu_power_w", + "total_cpu_energy_j", + "avg_total_cpu_rail_power_w", + "total_cpu_rail_energy_j", + "avg_total_cpu_sysio_power_w", + "total_cpu_sysio_energy_j", + "avg_total_module_power_w", + "total_module_energy_j", +) + + +def _rounded_metric(key: str, value: float | None) -> float: + if value is None or not math.isfinite(value): + raise ValueError(f"non-finite power metric: {key}") + return round(value, 3 if key.endswith(("_w", "_j")) else 6) def with_power_metrics( @@ -39,15 +59,21 @@ def with_power_metrics( schema_version: int, power_valid: bool, metrics: Mapping[str, float], + cpu_power_valid: bool | None = None, + cpu_metrics: Mapping[str, float] | None = None, ) -> dict[str, Any]: """Return a result with stale metrics replaced by one validated metric set. The caller supplies the metric family and schema version. Detailed invalid reasons belong in its validation sidecar, not the numeric metric payload. Neither the input result nor the supplied metrics is mutated. + + The CPU-side leg is tri-state: ``cpu_power_valid=None`` means the package + carried no ``cpu/`` sub-package, so only stale CPU keys are stripped and the + output stays byte-identical to a run without the leg. """ data = dict(result) if isinstance(result, Mapping) else result - for key in metric_keys: + for key in (*metric_keys, *CPU_METRIC_KEYS, "cpu_power_valid"): data.pop(key, None) data["power_metric_schema_version"] = schema_version data["power_valid"] = int(power_valid) @@ -55,8 +81,10 @@ def with_power_metrics( data.pop("power_audit", 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) + data[key] = _rounded_metric(key, value) + if cpu_power_valid is not None: + data["cpu_power_valid"] = int(cpu_power_valid) + if cpu_power_valid: + for key, value in (cpu_metrics or {}).items(): + data[key] = _rounded_metric(key, value) return data diff --git a/inferencex-e2e/infx/results/power/audit.py b/inferencex-e2e/infx/results/power/audit.py index 443b18f3e5..79303ac2ba 100644 --- a/inferencex-e2e/infx/results/power/audit.py +++ b/inferencex-e2e/infx/results/power/audit.py @@ -7,6 +7,35 @@ from collections.abc import Mapping from typing import Any +from .cpu_side import SENSOR_GRACE + +_REASON_CODE = re.compile(r"[a-z][a-z0-9_]{0,63}") +_CPU_COUNT_FIELDS = ("expected_sockets", "observed_sockets", "sample_row_count") + + +def _bounded_reasons(values: Any) -> list[str]: + if not isinstance(values, list): + return [] + return [ + reason for reason in values if isinstance(reason, str) and _REASON_CODE.fullmatch(reason) + ][:32] + + +def _cpu_audit(cpu: Mapping[str, Any]) -> dict[str, Any]: + """Project the CPU-side provenance: sensor kind, source, counts, reasons.""" + summary: dict[str, Any] = {} + if cpu.get("sensor_kind") == SENSOR_GRACE: + summary["sensor_kind"] = SENSOR_GRACE + source = cpu.get("source") + if isinstance(source, str) and 0 < len(source) <= 32: + summary["source"] = source + for key in _CPU_COUNT_FIELDS: + value = cpu.get(key) + if type(value) is int and value >= 0: + summary[key] = value + summary["reason_codes"] = _bounded_reasons(cpu.get("reason_codes")) + return summary + def audit_summary(validation: Mapping[str, Any], source: str) -> dict[str, Any]: """Project the shared app audit contract without publishing raw telemetry.""" @@ -49,11 +78,10 @@ def audit_summary(validation: Mapping[str, Any], source: str) -> dict[str, Any]: audit["observed_gpu_ids"] = list( dict.fromkeys(str(value) for value in ids if 0 < len(str(value)) <= 128) )[:1024] + cpu = validation.get("cpu") + if isinstance(cpu, Mapping): + audit["cpu"] = _cpu_audit(cpu) return { - "power_invalid_reasons": [ - reason - for reason in validation.get("reasons", []) - if isinstance(reason, str) and re.fullmatch(r"[a-z][a-z0-9_]{0,63}", reason) - ][:32], + "power_invalid_reasons": _bounded_reasons(validation.get("reasons", [])), "power_audit": audit, } diff --git a/inferencex-e2e/infx/results/power/common.py b/inferencex-e2e/infx/results/power/common.py index a9b33c2eeb..aaf191fa27 100644 --- a/inferencex-e2e/infx/results/power/common.py +++ b/inferencex-e2e/infx/results/power/common.py @@ -7,12 +7,10 @@ import json import math import sys -from collections.abc import Iterable, Mapping +from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path -from . import POWER_METRIC_SCHEMA_VERSION, with_power_metrics - @dataclass(frozen=True) class BenchmarkData: @@ -30,6 +28,11 @@ def integration_duration_s(self) -> float: return self.end_unix - self.start_unix +# Widest gap between consecutive samples inside the integration window that +# still counts as continuous coverage (srt-slurm contract.MAX_SAMPLE_GAP_SECONDS). +MAX_SAMPLE_GAP_SECONDS = 3.0 + + def _append_reason(reasons: list[str], reason: str) -> None: """Append a validation reason once while preserving discovery order.""" if reason not in reasons: @@ -53,6 +56,16 @@ def _interpolate_power(samples: list[tuple[float, float]], timestamp: float) -> return left_power + fraction * (right_power - left_power) +def _bracketing_sequence(times: tuple[float, ...], start: float, end: float) -> list[float] | None: + """Samples covering ``[start, end]``: the last at/before, the inside, the first at/after.""" + before = [value for value in times if value <= start] + after = [value for value in times if value >= end] + if not before or not after: + return None + inside = [value for value in times if start < value < end] + return [before[-1], *inside, after[0]] + + def _integrate_device( samples: list[tuple[float, float]], *, @@ -253,22 +266,3 @@ def audit_metrics(metrics: Mapping[str, float | None]) -> dict[str, float]: for key, value in metrics.items() if value is not None and math.isfinite(value) } - - -def patch_power_metrics( - path: Path, - *, - metric_keys: Iterable[str], - power_valid: bool, - metrics: Mapping[str, float], -) -> None: - """Validate replacement metrics before atomically updating an aggregate.""" - data = json.loads(path.read_text(encoding="utf-8")) - data = with_power_metrics( - data, - metric_keys=metric_keys, - schema_version=POWER_METRIC_SCHEMA_VERSION, - power_valid=power_valid, - metrics=metrics, - ) - _write_json_atomic(path, data) diff --git a/inferencex-e2e/infx/results/power/cpu_side.py b/inferencex-e2e/infx/results/power/cpu_side.py new file mode 100644 index 0000000000..e2fb4f5484 --- /dev/null +++ b/inferencex-e2e/infx/results/power/cpu_side.py @@ -0,0 +1,321 @@ +"""Grace CPU-side power leg of the multinode srt-slurm package. + +srt-slurm's ``cpu_power_exporter`` leg writes ``power/cpu/samples.csv`` beside +the GPU DCGM package (srt-slurm ``src/srtctl/core/power/cpu_session.py``); its +``cpu_manifest.json`` is debugging metadata and is not read. Both accepted +schemas are the wide format: one row per (scrape, host, socket) whose ``sensor`` names the +Grace socket meter and whose ``cpu_rail_w`` / ``soc_w`` columns carry the ACPI +component rails of that socket. Schema v3 (srt-slurm v2.59.0+) adds ``module_w``, +NVML superchip module power; the module contains both GPUs, so it is published +as its own rail and never added to the CPU or GPU totals. + +Each socket's Grace socket series is integrated over the GPU leg's bound formal +window with the shared trapezoid and boundary interpolation and published as +additive metrics next to the GPU-board numbers. The CPU-only, SysIO and module +rails are integrated the same way when every socket reports them; a rail with +gaps drops only its own metrics. Any failure records ``cpu_power_valid=0`` with +reason codes and leaves every GPU field untouched. This module only judges the +leg; the caller decides whether an invalid leg fails the job. +""" + +from __future__ import annotations + +import csv +import itertools +import math +from collections.abc import Iterable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .common import ( + MAX_SAMPLE_GAP_SECONDS, + _append_reason, + _bracketing_sequence, + _integrate_device, + audit_metrics, +) + +CPU_DIRNAME = "cpu" +CPU_SAMPLES_FILENAME = "samples.csv" + +_CPU_SAMPLES_HEADER_V2 = ( + "schema_version", + "timestamp_unix", + "hostname", + "source", + "sensor", + "socket_id", + "power_w", + "cpu_rail_w", + "soc_w", + "dram_w", + "total_power_w", +) +# srt-slurm cpu_samples._SCHEMA_VERSION_BY_HEADER: every row carries its header's +# version. v2.51.1 writes v2; v2.59.0+ writes v3, module_w before total_power_w. +_SCHEMA_VERSION_BY_HEADER = { + _CPU_SAMPLES_HEADER_V2: 2, + (*_CPU_SAMPLES_HEADER_V2[:-1], "module_w", "total_power_w"): 3, +} + +# One GB200/GB300 compute tray per worker host, two Grace sockets per tray. +SOCKETS_PER_HOST = 2 + +# The headline meter: "Grace Power Socket N", the whole socket (CPU rail, SysIO, +# LPDDR5X and regulator loss). srt-slurm names its wide-format rows +# ``CPU:cpuSidePowerUsageW`` (cpu_rails.SENSOR_SUFFIXES). +SENSOR_GRACE = "grace_socket" +_GRACE_SENSOR_SUFFIX = "cpuSidePowerUsageW" +# Wide-format rail column -> published rail family: "CPU Power Socket N" (cores, +# no SysIO), "SysIO Power Socket N" and the v3 NVML superchip module. dram_w +# stays blank on NVL72 firmware; module_w is blank when NVML is unavailable. +RAIL_COLUMNS = {"cpu_rail_w": "cpu_rail", "soc_w": "cpu_sysio", "module_w": "module"} + +_INTEGRATION_METHOD = "per_socket_trapezoidal_with_linear_boundary_interpolation" + + +@dataclass(frozen=True) +class CpuSampleRow: + timestamp_unix: float + hostname: str + source: str + sensor: str + socket_id: int + power_w: float + rails: tuple[tuple[str, float], ...] = () + + +@dataclass +class CpuPowerAudit: + """Everything the sidecar needs to explain the CPU-side verdict.""" + + valid: bool = False + sensor_kind: str | None = None + source: str | None = None + expected_sockets: int = 0 + observed_sockets: int = 0 + sample_row_count: int = 0 + reason_codes: list[str] = field(default_factory=list) + per_series_energy_j: dict[str, float] = field(default_factory=dict) + per_series_max_sample_gap_s: dict[str, float] = field(default_factory=dict) + metrics: dict[str, float] = field(default_factory=dict) + + def invalidate(self, *reasons: str) -> None: + for reason in reasons: + _append_reason(self.reason_codes, reason) + self.valid = False + self.metrics = {} + + def to_payload(self) -> dict[str, Any]: + return { + "cpu_power_valid": self.valid, + "reason_codes": list(self.reason_codes), + "sensor_kind": self.sensor_kind, + "source": self.source, + "sockets_per_host": SOCKETS_PER_HOST, + "expected_sockets": self.expected_sockets, + "observed_sockets": self.observed_sockets, + "sample_row_count": self.sample_row_count, + "integration_method": _INTEGRATION_METHOD, + "max_sample_gap_seconds": MAX_SAMPLE_GAP_SECONDS, + "per_series_max_sample_gap_s": dict(self.per_series_max_sample_gap_s), + # Overflowed integrations reach here as inf; keep the sidecar strict JSON. + "per_series_energy_j": { + key: value if math.isfinite(value) else None + for key, value in self.per_series_energy_j.items() + }, + "metrics": audit_metrics(self.metrics), + } + + +def is_grace_sensor(sensor: str) -> bool: + """True for the Grace socket meter's ``CPU:cpuSidePowerUsageW`` sensor name.""" + return sensor.partition(":")[2] == _GRACE_SENSOR_SUFFIX + + +def _parse_row(raw: list[str], header: tuple[str, ...], version: int) -> CpuSampleRow | None: + if len(raw) != len(header): + return None + cell = dict(zip(header, raw, strict=True)) + try: + schema_version = int(cell["schema_version"]) + timestamp_unix = float(cell["timestamp_unix"]) + socket_id = int(cell["socket_id"]) + power_w = float(cell["power_w"]) + except ValueError: + return None + hostname, source, sensor = cell["hostname"], cell["source"], cell["sensor"] + if schema_version != version or not hostname or not source or not sensor: + return None + if not math.isfinite(timestamp_unix) or not math.isfinite(power_w) or power_w < 0: + return None + if socket_id < 0: + return None + rails = [] + for column, family in RAIL_COLUMNS.items(): + value = cell.get(column, "") + if not value: + continue + try: + watts = float(value) + except ValueError: + return None + if not math.isfinite(watts) or watts < 0: + return None + rails.append((family, watts)) + return CpuSampleRow( + timestamp_unix=timestamp_unix, + hostname=hostname, + source=source, + sensor=sensor, + socket_id=socket_id, + power_w=power_w, + rails=tuple(rails), + ) + + +def read_cpu_samples(path: Path) -> tuple[tuple[CpuSampleRow, ...], tuple[str, ...]]: + """Strictly parse a v2 or v3 wide sample file; malformed rows invalidate the leg.""" + if not path.is_file(): + return (), ("cpu_samples_missing",) + rows: list[CpuSampleRow] = [] + reasons: list[str] = [] + try: + with open(path, newline="", encoding="utf-8") as handle: + reader = csv.reader(handle) + header = tuple(next(reader, ())) + version = _SCHEMA_VERSION_BY_HEADER.get(header) + if version is None: + return (), ("cpu_samples_header_mismatch",) + for raw in reader: + row = _parse_row(raw, header, version) + if row is None: + _append_reason(reasons, "cpu_samples_malformed") + continue + rows.append(row) + except (OSError, UnicodeDecodeError, csv.Error): + _append_reason(reasons, "cpu_samples_malformed") + return tuple(rows), tuple(reasons) + + +SocketKey = tuple[str, int] +# (host, socket) -> family -> {timestamp: watts}; the Grace meter and its rails. +SocketSeries = dict[SocketKey, dict[str, dict[float, float]]] + + +def _grace_series(rows: Iterable[CpuSampleRow]) -> tuple[SocketSeries, set[str], bool]: + """Group the Grace socket rows per socket, with the sources seen and a duplicate flag.""" + series: SocketSeries = {} + sources: set[str] = set() + duplicate = False + for row in rows: + if not is_grace_sensor(row.sensor): + continue + by_family = series.setdefault((row.hostname, row.socket_id), {}) + points = by_family.setdefault(SENSOR_GRACE, {}) + if row.timestamp_unix in points: + duplicate = True + points[row.timestamp_unix] = row.power_w + sources.add(row.source) + for family, watts in row.rails: + by_family.setdefault(family, {})[row.timestamp_unix] = watts + return series, sources, duplicate + + +def _max_sample_gap(samples: list[tuple[float, float]], start: float, end: float) -> float | None: + """The widest gap between the samples bracketing ``[start, end]``; None when not bracketed.""" + sequence = _bracketing_sequence(tuple(t for t, _ in samples), start, end) + if sequence is None: + return None + return max((later - earlier for earlier, later in itertools.pairwise(sequence)), default=0.0) + + +def validate_cpu_leg( + cpu_dir: Path, + *, + window: tuple[float, float] | None, + expected_hosts: Iterable[str], + expected: bool = False, +) -> CpuPowerAudit | None: + """Validate and integrate the CPU sub-package; None when the run carried none. + + ``expected`` is the point's ``cpu-power`` switch: a missing sub-package is + then a verdict rather than silence. + """ + if not cpu_dir.is_dir(): + if not expected: + return None + audit = CpuPowerAudit() + audit.invalidate("cpu_artifacts_missing") + return audit + audit = CpuPowerAudit() + rows, sample_reasons = read_cpu_samples(cpu_dir / CPU_SAMPLES_FILENAME) + audit.sample_row_count = len(rows) + audit.invalidate(*sample_reasons) + series, sources, duplicate = _grace_series(rows) + if duplicate or len(sources) > 1: + audit.invalidate("cpu_samples_malformed") + expected_keys = { + (host, socket) for host in expected_hosts for socket in range(SOCKETS_PER_HOST) + } + audit.expected_sockets = len(expected_keys) + audit.observed_sockets = len(series) + if window is None: + audit.invalidate("cpu_window_unavailable") + if audit.reason_codes: + return audit + if not expected_keys or set(series) != expected_keys: + audit.invalidate("cpu_socket_count_mismatch") + return audit + audit.sensor_kind = SENSOR_GRACE + audit.source = next(iter(sources)) + + # A rail is published only when every socket reports it, gap-free; the Grace + # meter must be gap-free on every socket or the whole leg is invalid. + families = [SENSOR_GRACE] + [ + rail + for rail in RAIL_COLUMNS.values() + if all(rail in by_family for by_family in series.values()) + ] + start, end = window + energy: dict[str, float] = {} + dropped_rails: set[str] = set() + for (host, socket), by_family in sorted(series.items()): + for family in families: + samples = sorted(by_family[family].items()) + label = f"{host}/socket{socket}/{family}" + gap = _max_sample_gap(samples, start, end) + if gap is not None: + audit.per_series_max_sample_gap_s[label] = gap + if gap is None or gap > MAX_SAMPLE_GAP_SECONDS: + if family != SENSOR_GRACE: + dropped_rails.add(family) + elif gap is None: + audit.invalidate("cpu_window_not_bracketed") + else: + audit.invalidate("cpu_sample_gap_exceeded") + continue + joules = _integrate_device(samples, start_unix=start, end_unix=end) + audit.per_series_energy_j[label] = joules + energy[family] = energy.get(family, 0.0) + joules + if audit.reason_codes: + audit.per_series_energy_j = {} + return audit + + duration = end - start + metrics = { + "avg_cpu_socket_power_w": energy[SENSOR_GRACE] / duration / len(expected_keys), + "avg_total_cpu_power_w": energy[SENSOR_GRACE] / duration, + "total_cpu_energy_j": energy[SENSOR_GRACE], + } + for rail in RAIL_COLUMNS.values(): + if rail in energy and rail not in dropped_rails: + metrics[f"avg_total_{rail}_power_w"] = energy[rail] / duration + metrics[f"total_{rail}_energy_j"] = energy[rail] + if any(not math.isfinite(value) for value in metrics.values()): + audit.invalidate("non_finite_power_metric") + return audit + audit.metrics = metrics + audit.valid = True + return audit diff --git a/inferencex-e2e/infx/results/power/multinode.py b/inferencex-e2e/infx/results/power/multinode.py index b118c1245e..29877b1283 100644 --- a/inferencex-e2e/infx/results/power/multinode.py +++ b/inferencex-e2e/infx/results/power/multinode.py @@ -27,6 +27,13 @@ failed sibling window, the sidecar retains the healthy measurement in ``selected_window`` and its per-GPU diagnostics at the top level. Publication and aggregate power metrics remain blocked. + +Grace CPU-side leg: when the package also carries ``power/cpu/`` (see +:mod:`.cpu_side`), each socket's Grace socket power and its reference rails are +integrated over the same bound formal window and published under +``cpu_power_valid``. That verdict is independent of ``power_valid``. A missing +or invalid CPU leg fails the run only with ``require_power`` and +``expected_cpu``; otherwise it never changes the exit code. """ from __future__ import annotations @@ -45,20 +52,25 @@ from . import ( ALL_POWER_METRIC_KEYS as _ALL_POWER_METRIC_KEYS, + CPU_METRIC_KEYS as CPU_METRIC_KEYS, + POWER_METRIC_SCHEMA_VERSION, ROLE_METRIC_KEYS as ROLE_METRIC_KEYS, WHOLE_METRIC_KEYS as WHOLE_METRIC_KEYS, + with_power_metrics, ) from .common import ( + MAX_SAMPLE_GAP_SECONDS, BenchmarkData, _append_reason, + _bracketing_sequence, _integrate_device, _load_benchmark_data, _percentile_total_power, _write_json_atomic, audit_metrics, benchmark_window_payload, - patch_power_metrics, ) +from .cpu_side import CPU_DIRNAME, CpuPowerAudit, validate_cpu_leg SCHEMA_VERSION = 1 PRODUCER = "srt-slurm.dcgm-power" @@ -83,7 +95,6 @@ SAMPLES_HEADER_V3 = (*SAMPLES_HEADER_V2, "temperature_c") SAMPLES_HEADERS = {1: SAMPLES_HEADER, 2: SAMPLES_HEADER_V2, 3: SAMPLES_HEADER_V3} -MAX_SAMPLE_GAP_SECONDS = 3.0 MAX_TEMPERATURE_C = 200.0 WORKER_ROLES = ("prefill", "decode", "agg") @@ -696,15 +707,6 @@ def _check_window_result(window: ParsedWindow, result_root: Path) -> list[str]: return [] -def _bracketing_sequence(times: tuple[float, ...], start: float, end: float) -> list[float] | None: - before = [value for value in times if value <= start] - after = [value for value in times if value >= end] - if not before or not after: - return None - inside = [value for value in times if start < value < end] - return [before[-1], *inside, after[0]] - - def _check_coverage( start: float, end: float, @@ -891,6 +893,10 @@ class MultinodePowerAudit: observed_gpu_count: int = 0 expected_gpu_count: int = 0 metrics: dict[str, float] = field(default_factory=dict) + # Input the CPU-side leg borrows from the GPU package; not serialized. + expected_worker_hosts: tuple[str, ...] = () + # None when the package carried no cpu/ sub-package. + cpu: CpuPowerAudit | None = None def _canonical_sha256(payload: dict) -> str: @@ -915,8 +921,48 @@ def validate_and_integrate( decode_gpus: int, aggregate_gpus: int, expected_producer_sha: str | None, + expected_cpu: bool = False, +) -> MultinodePowerAudit: + """Recompute package validity, cross-check verdicts, and integrate energy. + + The GPU leg runs first and binds the formal window; the CPU-side leg then + reuses that window and the manifest topology as inputs. No GPU verdict + reaches it: an unpinned producer or failed GPU coverage withholds GPU + energy only, and ``cpu_power_valid`` judges the CPU samples on their own. + """ + audit = _validate_gpu_leg( + power_dir=power_dir, + logs_root=logs_root, + bench_result_path=bench_result_path, + benchmark=benchmark, + prefill_gpus=prefill_gpus, + decode_gpus=decode_gpus, + aggregate_gpus=aggregate_gpus, + expected_producer_sha=expected_producer_sha, + ) + window = None + if audit.window is not None and _window_contract_holds(audit.window, audit.window_validations): + window = (audit.window["start_time_unix"], audit.window["end_time_unix"]) + audit.cpu = validate_cpu_leg( + power_dir / CPU_DIRNAME, + window=window, + expected_hosts=audit.expected_worker_hosts, + expected=expected_cpu, + ) + return audit + + +def _validate_gpu_leg( + *, + power_dir: Path, + logs_root: Path, + bench_result_path: Path, + benchmark: BenchmarkData | None, + prefill_gpus: int, + decode_gpus: int, + aggregate_gpus: int, + expected_producer_sha: str | None, ) -> MultinodePowerAudit: - """Recompute package validity, cross-check verdicts, and integrate energy.""" audit = MultinodePowerAudit( expected_producer_git_commit=expected_producer_sha, ) @@ -965,6 +1011,7 @@ def validate_and_integrate( _add_reason(audit, "manifest_malformed", f"manifest malformed: {exc!r}") audit.recomputed_publication_valid = False return audit + audit.expected_worker_hosts = tuple(sorted({device.hostname for device in expected_devices})) rows, sample_reasons = read_samples(power_dir / SAMPLES_FILENAME) recompute_failures += [f"{reason} in {SAMPLES_FILENAME}" for reason in sample_reasons] @@ -1070,7 +1117,6 @@ def validate_and_integrate( window = _select_window_for_result( audit, parsed_windows, logs_root, bench_result_path, benchmark ) - if ( not audit.package_integrity_valid or not topology_valid @@ -1238,13 +1284,83 @@ def _select_window_for_result( return window +# Per-window audit codes that describe GPU sample coverage rather than the +# window's own contract (srt-slurm power/windows.py _check_coverage); the +# CPU-side leg re-checks its own coverage. +_GPU_COVERAGE_REASONS = frozenset( + { + "measurement_window_not_bracketed", + "sample_gap_exceeded", + "gpu_uuid_changed", + "sample_loss_exceeded", + } +) + + +def _window_contract_holds(window: dict, validations: list[dict]) -> bool: + """True when the bound window passed every check that is not GPU coverage.""" + for validation in validations: + if (validation["benchmark_type"], validation["concurrency"]) == ( + window["benchmark_type"], + window["concurrency"], + ): + return all(reason in _GPU_COVERAGE_REASONS for reason in validation["reason_codes"]) + return False + + +# --- aggregate patch + sidecar + entry point --------------------------------- + + +def _host_workers(audit: MultinodePowerAudit) -> list[dict]: + """Project validated device energy onto physical host/role power entries.""" + if not audit.power_valid or audit.window is None: + return [] + duration_s = audit.window["end_time_unix"] - audit.window["start_time_unix"] + energies: dict[tuple[str, str], list[float]] = {} + for device, joules in audit.per_gpu_energy_j.items(): + host = device.rsplit("/", 1)[0] + role = audit.per_gpu_role[device] + energies.setdefault((role, host), []).append(joules) + workers = [] + indices: dict[str, int] = {} + for (role, host), values in sorted(energies.items()): + index = indices.get(role, 0) + workers.append( + { + "role": role, + "worker_idx": index, + "hosts": [host], + "num_gpus": len(values), + "avg_power_w": round(sum(values) / duration_s / len(values), 3), + } + ) + indices[role] = index + 1 + return workers + + def _patch_agg(agg_path: Path, audit: MultinodePowerAudit) -> None: - patch_power_metrics( - agg_path, + cpu = audit.cpu + data = with_power_metrics( + json.loads(agg_path.read_text(encoding="utf-8")), + schema_version=POWER_METRIC_SCHEMA_VERSION, metric_keys=_ALL_POWER_METRIC_KEYS, power_valid=audit.power_valid, metrics=audit.metrics, + cpu_power_valid=None if cpu is None else cpu.valid, + cpu_metrics={} if cpu is None else cpu.metrics, ) + data.pop("workers", None) + if workers := _host_workers(audit): + data["workers"] = workers + _write_json_atomic(agg_path, data) + + +def _withhold_metrics(audit: MultinodePowerAudit, cpu_reason: str | None = None) -> None: + """Drop every publishable number; the CPU leg only when the failure is shared.""" + audit.power_valid = False + audit.metrics = {} + if audit.cpu is not None and cpu_reason is not None: + audit.cpu.invalidate(cpu_reason) def _sidecar_payload( @@ -1254,7 +1370,7 @@ def _sidecar_payload( bench_result: Path, benchmark: BenchmarkData | None, ) -> dict: - return { + payload = { "schema_version": 1, "power_valid": audit.power_valid, "reasons": list(audit.reasons), @@ -1292,6 +1408,10 @@ def _sidecar_payload( }, "metrics": audit_metrics(audit.metrics), } + # Only when the package carried cpu/: sidecars without it stay byte-identical. + if audit.cpu is not None: + payload["cpu"] = audit.cpu.to_payload() + return payload def run( @@ -1306,6 +1426,7 @@ def run( aggregate_gpus: int = 0, validation_result: Path | None = None, require_power: bool = False, + expected_cpu: bool = False, ) -> int: """Validate multinode power artifacts, patch metrics, and audit the verdict.""" validation_result = validation_result or bench_result.with_name( @@ -1323,24 +1444,22 @@ def run( decode_gpus=decode_gpus, aggregate_gpus=aggregate_gpus, expected_producer_sha=expected_producer_sha, + expected_cpu=expected_cpu, ) for reason in benchmark_reasons: _add_reason(audit, reason) if benchmark is None or audit.reasons: - audit.power_valid = False - audit.metrics = {} + _withhold_metrics(audit) if not agg_result.is_file(): _add_reason(audit, "aggregate_result_missing", f"aggregate missing: {agg_result}") - audit.power_valid = False - audit.metrics = {} + _withhold_metrics(audit, "aggregate_result_missing") else: try: _patch_agg(agg_result, audit) except (OSError, json.JSONDecodeError, ValueError) as exc: _add_reason(audit, "aggregate_result_unwritable", str(exc)) - audit.power_valid = False - audit.metrics = {} + _withhold_metrics(audit, "aggregate_result_unwritable") print( f"[aggregate_power_multinode] Failed to patch {agg_result}: {exc}", file=sys.stderr, @@ -1364,6 +1483,20 @@ def run( ) return 1 if require_power else 0 + if audit.cpu is not None: + cpu = audit.cpu + print( + f"[aggregate_power_multinode] cpu_power_valid={int(cpu.valid)} " + f"sensor_kind={cpu.sensor_kind} " + f"avg_total_cpu_power_w={cpu.metrics.get('avg_total_cpu_power_w', 0.0):.2f} " + f"total_cpu_energy_j={cpu.metrics.get('total_cpu_energy_j', 0.0):.2f} " + f"reasons={','.join(cpu.reason_codes) or '-'}" + ) + + if require_power and expected_cpu and (audit.cpu is None or not audit.cpu.valid): + print("[aggregate_power_multinode] Required CPU power is invalid", file=sys.stderr) + return 1 + if not audit.power_valid: print( f"[aggregate_power_multinode] Power validation failed: " diff --git a/inferencex-e2e/infx/srt_slurm/generate.py b/inferencex-e2e/infx/srt_slurm/generate.py index 9d3c30fd79..6d9b7f400d 100644 --- a/inferencex-e2e/infx/srt_slurm/generate.py +++ b/inferencex-e2e/infx/srt_slurm/generate.py @@ -67,6 +67,7 @@ def point_environment(point: Mapping[str, Any]) -> dict[str, str]: IS_MULTINODE="true", SRT_RECIPE=point["srt-recipe"], POWER="1" if point.get("power") else "0", + CPU_POWER="1" if point.get("cpu-power") else "0", CONC_LIST=" ".join(map(str, point["conc"])), ) if agentic: @@ -98,15 +99,16 @@ def _binder_inputs( environment: Mapping[str, str], placement: tuple[str, Cluster] | None, root: Path, -) -> tuple[int | None, dict[str, str]]: - """The DCGM exporter port and AgentX client env the launcher binds a multi-node point with. +) -> tuple[int | None, str | None, dict[str, str]]: + """The DCGM exporter port, CPU exporter source and AgentX client env the launcher binds a + multi-node point with. Raises ``ValueError`` when they depend on a cluster the runner label does not single out, or the cluster's lanes refuse the point. """ agentic = environment["IS_AGENTIC"] == "1" if "prefill" not in point or not (agentic or point.get("power")): - return None, {} + return None, None, {} if placement is None: raise ValueError( f"runner {point['runner']!r} must schedule on one cluster to bind the cluster's " @@ -139,13 +141,15 @@ def bound_variant( *, expand: Callable[..., list[tuple[str | None, dict[str, Any]]]] = selected_recipes, power_port: int | None = None, + cpu_power_source: str | None = None, client_env: Mapping[str, str] | None = None, ) -> tuple[str | None, dict[str, Any]]: """The variant the launcher submits for ``point``, composed and bound. - Multi-node variants get the DCGM telemetry block on ``power_port`` and the client paths - ``client_env``, the binder inputs the launcher computes. ``'@fabric.'`` values stay - unresolved: recipe fingerprints hash them as written. + Multi-node variants get the DCGM telemetry block on ``power_port``, the Grace CPU exporter + on ``cpu_power_source``, and the client paths ``client_env``: the binder inputs the launcher + computes. ``'@fabric.'`` values stay unresolved: recipe fingerprints hash them as + written. """ if "prefill" not in point: selected, recipe = select_recipe( @@ -158,6 +162,7 @@ def bound_variant( root=root, expand=expand, power_port=power_port, + cpu_power_source=cpu_power_source, client_env=client_env, ) @@ -233,11 +238,12 @@ def generate_recipes( for point in points: environment = point_environment(point) placement = _placement(inventory, point["runner"]) - power_port, client_env = _binder_inputs(point, environment, placement, root) + power_port, cpu_source, client_env = _binder_inputs(point, environment, placement, root) fabric = _cluster_fabric(placement[1]) if placement else None variant, recipe = bound_variant( - point, environment, root, power_port=power_port, client_env=client_env - ) + point, environment, root, + power_port=power_port, cpu_power_source=cpu_source, client_env=client_env, + ) # fmt: skip if fabric is not None: recipe = resolve_fabric(recipe, fabric) check_setup_script(recipe, root / point["srt-recipe"].partition(":")[0], root) diff --git a/inferencex-e2e/infx/srt_slurm/workload.py b/inferencex-e2e/infx/srt_slurm/workload.py index 50a8372552..95e4ac57c9 100644 --- a/inferencex-e2e/infx/srt_slurm/workload.py +++ b/inferencex-e2e/infx/srt_slurm/workload.py @@ -41,6 +41,7 @@ ("benchmark", "concurrencies"), ("telemetry", "enabled"), ("telemetry", "dcgm_exporter", "port"), + ("telemetry", "cpu_power_exporter"), ) _FIXED_ENV = ("MODEL", "ISL", "OSL", "CONC", "CONC_LIST", "RANDOM_RANGE_RATIO", "USE_CHAT_TEMPLATE") _AGENTIC_ENV = ("CONC", "CONC_LIST", "RESULT_DIR", "AGENTIC_OUTPUT_DIR", "HF_HUB_CACHE", @@ -187,11 +188,13 @@ def compose_recipe( multinode: bool, root: Path, power_port: int | None = None, + cpu_power_source: str | None = None, native_power: bool = False, ) -> dict[str, Any]: """The fragment at ``path`` over its lane's shared block, under ``base`` for bundles. - ``power_port`` adds the DCGM telemetry block with its exporter on that port. + ``power_port`` adds the DCGM telemetry block with its exporter on that port, and + ``cpu_power_source`` the cluster's Grace CPU power exporter beside it. """ raw = yaml.safe_load(path.read_text()) if not isinstance(raw, dict): @@ -201,7 +204,9 @@ def compose_recipe( if native_power: shared = merge_blocks(shared, _block(root / NATIVE_TELEMETRY_BLOCK)) if power_port is not None: - exporter = {"telemetry": {"dcgm_exporter": {"port": power_port}}} + exporter: dict[str, Any] = {"telemetry": {"dcgm_exporter": {"port": power_port}}} + if cpu_power_source is not None: + exporter["telemetry"]["cpu_power_exporter"] = {"source": cpu_power_source} shared = merge_blocks(shared, merge_blocks(_block(root / TELEMETRY_BLOCK), exporter)) if "base" in raw: return {**raw, "base": merge_blocks(shared, raw["base"])} @@ -351,13 +356,19 @@ def bind_multinode( root: Path, expand: Callable[..., list[tuple[str | None, dict[str, Any]]]] = selected_recipes, power_port: int | None = None, + cpu_power_source: str | None = None, client_env: Mapping[str, str] | None = None, ) -> tuple[str | None, dict[str, Any]]: """Bind the one variant ``recipe`` (``fragment[:selector]``) selects; return its name too.""" agentic = environment["IS_AGENTIC"] == "1" path, _, selector = recipe.partition(":") composed = compose_recipe( - Path(path), agentic=agentic, multinode=True, root=root, power_port=power_port + Path(path), + agentic=agentic, + multinode=True, + root=root, + power_port=power_port, + cpu_power_source=cpu_power_source, ) variants = expand(composed, selector or None) if len(variants) != 1: @@ -420,6 +431,11 @@ def main(argv: list[str] | None = None) -> None: parser.add_argument( "--power-port", type=int, help="merge the DCGM telemetry block, its exporter on this port" ) + parser.add_argument( + "--cpu-power-source", + choices=("acpi",), + help="bind the cluster's Grace CPU power exporter with this source (needs --power-port)", + ) parser.add_argument( "--client-env", type=_assignment, @@ -437,6 +453,7 @@ def main(argv: list[str] | None = None) -> None: os.environ, root=root, power_port=args.power_port, + cpu_power_source=args.cpu_power_source, client_env=dict(args.client_env), ) bound = resolve_fabric(bound, args.fabric) diff --git a/inferencex-e2e/infx/tests/launch/test_launch_artifacts.py b/inferencex-e2e/infx/tests/launch/test_launch_artifacts.py index e82f1b37dc..b0e34fbef6 100644 --- a/inferencex-e2e/infx/tests/launch/test_launch_artifacts.py +++ b/inferencex-e2e/infx/tests/launch/test_launch_artifacts.py @@ -1,7 +1,9 @@ """Workspace staging behavior of the launcher artifact helpers.""" +import json import os import stat +import sys import tarfile import pytest @@ -112,3 +114,53 @@ def test_power_collection_records_the_job_status_and_stages_results_either_way(t assert (logs / "power" / "native-job-status.txt").read_text() == f"42|{status.raw}\n" assert (workspace / "point_conc4.json").exists() + + +def test_agentic_collection_requires_declared_cpu_and_keeps_gpu_valid(tmp_path): + from infx.tests.results.power.test_aggregate_power_multinode import ( + PRODUCER_SHA, + build_package, + ) + + pkg = build_package(tmp_path / "captured") + result_dir = pkg.logs_root / "agentic/conc_4" + result_dir.mkdir(parents=True) + formal_result = result_dir / "agentic_power_concurrency_4.json" + formal_result.write_bytes(pkg.original_result.read_bytes()) + [window_path] = pkg.windows_dir.glob("*.json") + window = json.loads(window_path.read_text()) + window["result_path"] = str(formal_result.relative_to(pkg.logs_root)) + window_path.unlink() + window_path = pkg.windows_dir / formal_result.name + window_path.write_text(json.dumps(window)) + manifest_path = pkg.power_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["window_validations"][0]["window_file"] = f"windows/{formal_result.name}" + for device in manifest["expected_devices"]: + device["assignments"][0].update(worker_role="agg", het_group=0) + manifest_path.write_text(json.dumps(manifest)) + source, workspace = tmp_path / "source", tmp_path / "output" + source.mkdir() + workspace.mkdir() + (source / "point_conc4.json").write_text( + json.dumps({"disagg": False, "num_prefill_gpu": 4, "num_decode_gpu": 0}) + ) + status = JobStatus(JobState.SUCCEEDED, "COMPLETED|0:0", 0) + assert collect_agentic_power_results( + status, + "42", + pkg.logs_root, + source, + workspace, + "point", + PRODUCER_SHA, + [4], + results_python=sys.executable, + expected_cpu=True, + ) == 1 + result = json.loads((workspace / "point_conc4.json").read_text()) + assert result["power_valid"] == 1 + assert result["cpu_power_valid"] == 0 + assert result["total_gpu_energy_j"] == 84000.0 + assert result["power_audit"]["source"] == "LOGS/agentic/conc_4/power_validation.json" + assert result["power_audit"]["cpu"]["reason_codes"] == ["cpu_artifacts_missing"] diff --git a/inferencex-e2e/infx/tests/launch/test_srt_config.py b/inferencex-e2e/infx/tests/launch/test_srt_config.py index 44cb3cfa41..75a5466ca7 100644 --- a/inferencex-e2e/infx/tests/launch/test_srt_config.py +++ b/inferencex-e2e/infx/tests/launch/test_srt_config.py @@ -12,7 +12,7 @@ from infx.clusters import Cluster, load_inventory from infx.clusters.slurm import slurm_settings from infx.launch.backends.slurm import SlurmBackend -from infx.launch.context import Launch +from infx.launch.context import Launch, LaunchError from infx.launch.drivers.srt.checkout import Checkout, compute_workspace from infx.launch.drivers.srt.config import SrtJob, binder_inputs, pyxis_spelling, render, write from infx.launch.drivers.srt.lanes import SrtLane @@ -292,9 +292,10 @@ def test_the_agentx_client_measures_power_exactly_when_the_launcher_collects_it( "CONC_LIST": "4", }) # fmt: skip - port, client_env = binder_inputs( + port, cpu_source, client_env = binder_inputs( record, slurm_settings(record).srt_slurm, SrtLane(), request, decision ) + assert cpu_source is None switches = ( client_env.get("IS_MULTINODE"), @@ -302,3 +303,26 @@ def test_the_agentx_client_measures_power_exactly_when_the_launcher_collects_it( client_env.get("REQUIRE_POWER"), ) assert (port, *switches) == expected + + +@pytest.mark.parametrize("declared", [True, False]) +def test_cpu_power_binds_the_clusters_grace_exporter_or_fails_before_binding(tmp_path, declared): + srt = {"power-exporter-port": 9401} + if declared: + srt["cpu-power-exporter"] = {"source": "acpi"} + record = cluster(srt=srt) + request = MultiNodeRequest.from_env({ + "RUNNER_NAME": "c_0", "GITHUB_WORKSPACE": str(tmp_path), "IMAGE": "i", + "FRAMEWORK": "dynamo-sglang", "MODEL_PREFIX": "m", "PRECISION": "fp8", + "SPEC_DECODING": "none", "RESULT_FILENAME": "r", "IS_AGENTIC": "0", "RUN_EVAL": "false", + "EVAL_ONLY": "false", "SRT_RECIPE": "r.yaml", + }) # fmt: skip + settings = slurm_settings(record).srt_slurm + cpu = PowerDecision(dcgm=True, agentx=False, cpu=True) + if not declared: + with pytest.raises(LaunchError, match=r"no srt-slurm\.cpu-power-exporter"): + binder_inputs(record, settings, SrtLane(), request, cpu) + return + assert binder_inputs(record, settings, SrtLane(), request, cpu)[:2] == (9401, "acpi") + gpu_only = PowerDecision(dcgm=True, agentx=False) + assert binder_inputs(record, settings, SrtLane(), request, gpu_only)[:2] == (9401, None) diff --git a/inferencex-e2e/infx/tests/launch/test_srt_driver.py b/inferencex-e2e/infx/tests/launch/test_srt_driver.py index af1e1b5a30..865c279bdb 100644 --- a/inferencex-e2e/infx/tests/launch/test_srt_driver.py +++ b/inferencex-e2e/infx/tests/launch/test_srt_driver.py @@ -19,6 +19,7 @@ from infx.launch.__main__ import main from infx.launch.drivers.srt import lanes, models +from infx.launch.drivers.srt.checkout import CPU_POWER_EXPORTER_RELEASE from infx.launch.drivers.srt.lanes import LaneMount, SrtLane from infx.launch.drivers.srt.models import Override from infx.launch.policy import LaunchPath, Match @@ -376,6 +377,8 @@ def test_power_lane_stages_provenance_and_validates_each_concurrency( assert bound["benchmark"]["env"]["REQUIRE_POWER"] == require_power assert bound["telemetry"]["collector_join_timeout_seconds"] == 12 assert (bound["telemetry"]["enabled"], bound["telemetry"]["dcgm_exporter"]["port"]) == (True, 9401) + assert "cpu_power_exporter" not in bound["telemetry"] + assert all("CPU_POWER_EXPORTER_RELEASE" not in call for call in lines(harness.logs, "make")) runs = lines(harness.logs, "adapter") assert [run.split("--result-dir ")[1].split()[0].rsplit("/", 1)[1] for run in runs] == ["conc_4"] @@ -666,3 +669,49 @@ def test_multinode_eval_overrides_image_offline_mode_and_host_model_path(harness "http://worker:8000", str(harness.workspace), ] + + +def test_launch_cpu_expectation_reaches_fixed_sequence_processor(harness, monkeypatch): + from infx.results.fixed_sequence import aggregate_power_result + from infx.tests.results.power.test_aggregate_power_multinode import ( + PRODUCER_SHA, + build_package, + ) + + env = lane_env( + harness, + "gb300-nv", + LANE_FRAGMENT + POWER_TELEMETRY, + MODEL_PREFIX="qwen3.5", + PRECISION="fp8", + FRAMEWORK="dynamo-sglang", + MODEL="Qwen/Qwen3.5-397B-A17B-FP8", + REQUIRE_POWER="1", + POWER="1", + CPU_POWER="1", + ) + assert_ok(launch(env, harness.config, harness.workspace)) + [checkout] = harness.workspace.glob("srt-slurm-9001-*") + bound = yaml.safe_load((checkout / "recipe.yaml").read_text()) + assert bound["telemetry"]["cpu_power_exporter"] == {"source": "acpi"} + assert bound["telemetry"]["dcgm_exporter"]["port"] == 19401 + [setup] = [call for call in lines(harness.logs, "make") if call.startswith("setup ")] + assert f"CPU_POWER_EXPORTER_RELEASE={CPU_POWER_EXPORTER_RELEASE}" in setup.split() + pkg = build_package(harness.tmp / "power-input") + monkeypatch.chdir(harness.tmp) + # The workflow exports CPU_POWER once; the launcher and the processor both read it. + result_env = { + "CPU_POWER": "1", + "REQUIRE_POWER": "1", + "IS_MULTINODE": "true", + "RESULT_FILENAME": "fixture", + "PREFILL_GPUS": "2", + "DECODE_GPUS": "2", + "POWER_PRODUCER_SHA": PRODUCER_SHA, + "POWER_ARTIFACT_DIR": str(pkg.power_dir), + "POWER_RESULT_ROOT": str(pkg.logs_root), + } + assert aggregate_power_result(result_env, pkg.bench_result, pkg.agg_result) == 1 + aggregate = pkg.agg() + assert aggregate["power_valid"] == 1 + assert aggregate["cpu_power_valid"] == 0 diff --git a/inferencex-e2e/infx/tests/matrix/test_generate_sweep_configs.py b/inferencex-e2e/infx/tests/matrix/test_generate_sweep_configs.py index 66f5871307..f32bce0f9e 100644 --- a/inferencex-e2e/infx/tests/matrix/test_generate_sweep_configs.py +++ b/inferencex-e2e/infx/tests/matrix/test_generate_sweep_configs.py @@ -3268,6 +3268,22 @@ def test_require_power_is_scoped_to_one_multinode_fixed_sequence( expand_full_sweep(config, sample_runner_config) +def test_cpu_power_reaches_the_planner_row_beside_power( + sample_multinode_config, sample_runner_config +): + from infx.matrix.generate import expand_full_sweep + + config = sample_multinode_config + point = next(iter(config.values()))["scenarios"]["fixed-seq-len"][0]["search-space"][0] + point["power"] = True + rows = expand_full_sweep(config, sample_runner_config) + assert rows + assert all(row["power"] is True and "cpu-power" not in row for row in rows) + point["cpu-power"] = True + rows = expand_full_sweep(config, sample_runner_config) + assert all(row["cpu-power"] is True for row in rows) + + def test_single_node_fixed_sequence_rejects_require_power(sample_single_node_config): from infx.matrix.validation import validate_master_config diff --git a/inferencex-e2e/infx/tests/matrix/test_validation.py b/inferencex-e2e/infx/tests/matrix/test_validation.py index ac78746d81..eb63d6524a 100644 --- a/inferencex-e2e/infx/tests/matrix/test_validation.py +++ b/inferencex-e2e/infx/tests/matrix/test_validation.py @@ -639,6 +639,13 @@ def test_valid_aggregate_worker(self): assert entry.prefill is None assert entry.decode is None + def test_cpu_power_rides_on_power(self): + row = {"worker": {"tp": 8, "ep": 1, "dp-attn": False}, "num-nodes": 2, + "conc-list": [1], "cpu-power": True} # fmt: skip + with pytest.raises(ValidationError, match="cpu-power requires power"): + MultiNodeSearchSpaceEntry(**row) + assert MultiNodeSearchSpaceEntry(**{**row, "power": True}).cpu_power is True + def test_missing_conc_specification(self): with pytest.raises(ValidationError): MultiNodeSearchSpaceEntry(**{ @@ -1489,3 +1496,11 @@ def test_only_multinode_agentic_rows_measure_power(topology, accepted): else: with pytest.raises(ValidationError, match="Only multinode entries measure DCGM power"): AgenticCodingSearchSpaceEntry(**row) + + +def test_agentic_cpu_power_rides_on_power(): + worker = {"worker": {"num-worker": 1, "tp": 8, "ep": 1, "dp-attn": False}} + row = {"srt-recipe": "r.yaml", "conc-list": [4], "cpu-power": True, **worker} + with pytest.raises(ValidationError, match="cpu-power requires power"): + AgenticCodingSearchSpaceEntry(**row) + assert AgenticCodingSearchSpaceEntry(**{**row, "power": True}).cpu_power is True diff --git a/inferencex-e2e/infx/tests/results/power/test_aggregate_power.py b/inferencex-e2e/infx/tests/results/power/test_aggregate_power.py index 7cdf87142b..e501cec5bc 100644 --- a/inferencex-e2e/infx/tests/results/power/test_aggregate_power.py +++ b/inferencex-e2e/infx/tests/results/power/test_aggregate_power.py @@ -33,6 +33,7 @@ def test_power_replacement_removes_stale_metrics(tmp_path, patch_validated_power "hw": "fixture", "avg_power_w": 99, "total_gpu_energy_j": 50, "power_invalid_reasons": ["stale"], "power_audit": {"source": "previous-run.json"}, + "cpu_power_valid": 1, "total_cpu_energy_j": 1.0, "workers": [{"role": "agg"}], })) patch_validated_power(path, power_valid=valid, metrics={ "avg_power_w": 12.34567, "joules_per_output_token": 0.12345678, diff --git a/inferencex-e2e/infx/tests/results/power/test_aggregate_power_multinode.py b/inferencex-e2e/infx/tests/results/power/test_aggregate_power_multinode.py index f3d959881c..c81e554847 100644 --- a/inferencex-e2e/infx/tests/results/power/test_aggregate_power_multinode.py +++ b/inferencex-e2e/infx/tests/results/power/test_aggregate_power_multinode.py @@ -61,6 +61,7 @@ def run( aggregate_gpus=0, sha=PRODUCER_SHA, require_power=False, + expected_cpu=False, ): return apm.run( self.power_dir, @@ -73,6 +74,7 @@ def run( logs_root=self.logs_root, validation_result=self.validation_result, require_power=require_power, + expected_cpu=expected_cpu, ) def agg(self): @@ -358,6 +360,12 @@ def test_emits_all_metrics_exactly(self, tmp_path): assert agg["joules_per_input_token"] == 2.563477 assert agg["joules_per_output_token"] == 20.507812 assert agg["joules_per_total_token"] == 2.278646 + assert agg["workers"] == [ + {"role": "decode", "worker_idx": 0, "hosts": ["node-d"], + "num_gpus": 2, "avg_power_w": 300.0}, + {"role": "prefill", "worker_idx": 0, "hosts": ["node-p"], + "num_gpus": 2, "avg_power_w": 400.0}, + ] assert agg["prefill_gpu_energy_j"] == 48000.0 assert agg["decode_gpu_energy_j"] == 36000.0 assert agg["prefill_avg_power_w"] == 400.0 @@ -754,3 +762,267 @@ def test_v3_samples_reject_invalid_temperature(tmp_path, temperature): rows, reasons = apm.read_samples(path) assert rows == () assert reasons == ("samples_csv_malformed",) + + +# --- Grace CPU-side power leg (power/cpu/) ----------------------------------- + +CPU_HOSTS = tuple(sorted({host for host, *_ in DEVICES})) +CPU_HEADER = [ + "schema_version", "timestamp_unix", "hostname", "source", "sensor", "socket_id", + "power_w", "cpu_rail_w", "soc_w", "dram_w", "total_power_w", +] +CPU_HEADER_V3 = [*CPU_HEADER[:-1], "module_w", "total_power_w"] +GRACE_W = {"node-d": 500.0, "node-p": 700.0} +DCGM_W = 300.0 +# Component rails never feed a headline metric; the values are chosen so any +# leak into a headline key is visible against the Grace constants. +RAILS_W = {"cpu_rail": 200.0, "soc": 50.0} +# NVML superchip module power per socket, distinct from the Grace and rail +# constants so a leak into a CPU key is visible. +MODULE_W = 1500.0 +# srt-slurm cpu_rails.SENSOR_SUFFIXES: the Grace socket meter and the DCGM CPU rail. +GRACE_SENSOR = "CPU{s}:cpuSidePowerUsageW" +DCGM_SENSOR = "CPU{s}:cpuPowerUsageW" + + +def _cpu_row(ts, host, socket, kind, watts): + if kind == "dcgm": + return [2, repr(ts), host, "dcgm", DCGM_SENSOR.format(s=socket), socket, repr(watts), "", "", "", ""] + rails = [repr(w) for w in RAILS_W.values()] + return [2, repr(ts), host, "acpi", GRACE_SENSOR.format(s=socket), socket, repr(watts), *rails, "", ""] + + +def _cpu_rows(kinds_by_host=None, *, hosts=CPU_HOSTS): + """One row per (scrape, host, socket, kind); ``None`` watts means the Grace constant.""" + kinds_by_host = kinds_by_host or {host: {"grace": None} for host in hosts} + rows = [] + ts = FIRST_TS + while ts <= LAST_TS: + for host in hosts: + for socket in (0, 1): + for kind, watts in kinds_by_host.get(host, {}).items(): + watts = GRACE_W[host] if watts is None else watts + rows.append(_cpu_row(ts, host, socket, kind, watts)) + ts += 1.0 + return rows + + +def _v3(rows, module_w=MODULE_W): + """The same scrapes in srt-slurm v2.59.0+'s schema 3; ``None`` leaves module_w blank.""" + module = "" if module_w is None else repr(module_w) + return [[3, *row[1:-1], module, row[-1]] for row in rows] + + +def add_cpu_package(pkg, rows, header=CPU_HEADER): + cpu_dir = pkg.power_dir / "cpu" + cpu_dir.mkdir() + with open(cpu_dir / "samples.csv", "w", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(header) + writer.writerows(rows) + + +def _gpu_fields(agg): + return {k: v for k, v in agg.items() if k != "cpu_power_valid" and k not in apm.CPU_METRIC_KEYS} + + +def _reference_agg(tmp_path): + """The GPU-only aggregate every CPU-leg outcome must reproduce field for field.""" + reference = build_package(tmp_path / "reference") + assert reference.run() == 0 + return reference.agg() + + +def assert_grace_keys(agg): + assert agg["cpu_power_valid"] == 1 + assert agg["avg_cpu_socket_power_w"] == 600.0 + assert agg["avg_total_cpu_power_w"] == 2400.0 + assert agg["total_cpu_energy_j"] == 144000.0 + + +class TestCpuSidePower: + def test_grace_total_emits_cpu_keys_and_provenance(self, tmp_path): + from infx.results.power.audit import audit_summary + + pkg = build_package(tmp_path) + rows = _cpu_rows() + add_cpu_package(pkg, rows) + assert pkg.run(require_power=True) == 0 + + agg = pkg.agg() + assert_grace_keys(agg) + assert agg["avg_total_cpu_rail_power_w"] == 800.0 + assert agg["total_cpu_rail_energy_j"] == 48000.0 + assert agg["avg_total_cpu_sysio_power_w"] == 200.0 + assert agg["total_cpu_sysio_energy_j"] == 12000.0 + assert _gpu_fields(agg) == _reference_agg(tmp_path) + + cpu = pkg.sidecar()["cpu"] + assert cpu["cpu_power_valid"] is True + assert cpu["reason_codes"] == [] + assert cpu["sensor_kind"] == "grace_socket" + assert cpu["source"] == "acpi" + assert cpu["expected_sockets"] == 4 + assert cpu["observed_sockets"] == 4 + assert cpu["sample_row_count"] == len(rows) + assert set(cpu["per_series_energy_j"]) == { + f"{host}/socket{socket}/{kind}" + for host in CPU_HOSTS + for socket in (0, 1) + for kind in ("grace_socket", "cpu_rail", "cpu_sysio") + } + assert audit_summary(pkg.sidecar(), "power_validation.json")["power_audit"]["cpu"] == { + "sensor_kind": "grace_socket", + "source": "acpi", + "expected_sockets": 4, + "observed_sockets": 4, + "sample_row_count": len(rows), + "reason_codes": [], + } + + def test_rail_gap_drops_only_that_rail(self, tmp_path): + pkg = build_package(tmp_path) + rows = _cpu_rows() + for row in rows: + if row[2] == "node-d" and row[5] == 0 and 1020.0 <= float(row[1]) <= 1030.0: + row[7] = "" + add_cpu_package(pkg, rows) + assert pkg.run(require_power=True) == 0 + agg = pkg.agg() + assert_grace_keys(agg) + assert "avg_total_cpu_rail_power_w" not in agg + assert agg["avg_total_cpu_sysio_power_w"] == 200.0 + + def test_v3_module_rail_is_published_apart_from_cpu_keys(self, tmp_path): + pkg = build_package(tmp_path) + add_cpu_package(pkg, _v3(_cpu_rows()), CPU_HEADER_V3) + assert pkg.run(require_power=True) == 0 + agg = pkg.agg() + assert_grace_keys(agg) + assert agg["avg_total_module_power_w"] == 6000.0 + assert agg["total_module_energy_j"] == 360000.0 + assert _gpu_fields(agg) == _reference_agg(tmp_path) + module_series = {f"{host}/socket{socket}/module" for host in CPU_HOSTS for socket in (0, 1)} + assert module_series <= set(pkg.sidecar()["cpu"]["per_series_energy_j"]) + + @pytest.mark.parametrize("module", ["blank", "gap"]) + def test_v3_incomplete_module_drops_only_module_keys(self, tmp_path, module): + pkg = build_package(tmp_path) + if module == "blank": + # The musl exporter build cannot load NVML and leaves module_w empty. + rows = _v3(_cpu_rows(), module_w=None) + else: + rows = _v3(_cpu_rows()) + for row in rows: + if row[2] == "node-d" and row[5] == 0 and 1020.0 <= float(row[1]) <= 1030.0: + row[10] = "" + add_cpu_package(pkg, rows, CPU_HEADER_V3) + assert pkg.run(require_power=True) == 0 + agg = pkg.agg() + assert_grace_keys(agg) + assert agg["avg_total_cpu_rail_power_w"] == 800.0 + assert agg["avg_total_cpu_sysio_power_w"] == 200.0 + assert {"avg_total_module_power_w", "total_module_energy_j"}.isdisjoint(agg) + + @pytest.mark.parametrize( + "cpu, require_power", + [("missing", True), ("complete", True), ("missing", False), ("dcgm_only", True)], + ) + def test_expected_acpi_cpu_controls_strict_exit_without_changing_gpu( + self, tmp_path, cpu, require_power + ): + pkg = build_package(tmp_path) + if cpu == "complete": + add_cpu_package(pkg, _cpu_rows()) + elif cpu == "dcgm_only": + add_cpu_package(pkg, _cpu_rows({host: {"dcgm": DCGM_W} for host in CPU_HOSTS})) + assert pkg.run(require_power=require_power, expected_cpu=True) == int( + require_power and cpu != "complete" + ) + assert _gpu_fields(pkg.agg()) == _reference_agg(tmp_path) + assert pkg.sidecar()["power_valid"] is True + assert pkg.agg()["cpu_power_valid"] == int(cpu == "complete") + if cpu == "complete": + assert_grace_keys(pkg.agg()) + else: + assert set(apm.CPU_METRIC_KEYS).isdisjoint(pkg.agg()) + assert pkg.sidecar()["cpu"]["reason_codes"] + if cpu == "dcgm_only": + # The DCGM CPU rail is not the Grace socket meter: no socket series. + assert pkg.sidecar()["cpu"]["reason_codes"] == ["cpu_socket_count_mismatch"] + + @pytest.mark.parametrize( + "reason, window_trusted", + [ + # srt-slurm power/windows.py _check_coverage: GPU coverage only. + ("measurement_window_not_bracketed", True), + ("sample_gap_exceeded", True), + ("gpu_uuid_changed", True), + ("sample_loss_exceeded", True), + # The window's own contract: the CPU leg has no window either. + ("measurement_window_clock_mismatch", False), + ], + ) + def test_only_gpu_coverage_failures_leave_the_cpu_window_bound(self, reason, window_trusted): + window = {"benchmark_type": "sa-bench", "concurrency": 4} + validations = [{**window, "reason_codes": [reason], "power_coverage_valid": False}] + assert apm._window_contract_holds(window, validations) is window_trusted + + def test_package_without_cpu_dir_emits_no_cpu_fields(self, tmp_path): + pkg = build_package(tmp_path) + assert pkg.run() == 0 + agg = pkg.agg() + assert "cpu_power_valid" not in agg + assert set(apm.CPU_METRIC_KEYS).isdisjoint(agg) + assert "cpu" not in pkg.sidecar() + + def _drop(self, rows, host, socket, predicate): + return [ + row + for row in rows + if not (row[2] == host and row[5] == socket and predicate(float(row[1]))) + ] + + @pytest.fixture + def tampered(self, request, tmp_path): + kind = request.param + pkg = build_package(tmp_path) + rows = _cpu_rows() + header = CPU_HEADER + if kind == "socket": + rows = self._drop(rows, "node-p", 1, lambda ts: True) + elif kind == "gap": + rows = self._drop(rows, "node-d", 0, lambda ts: 1020.0 <= ts <= 1024.0) + elif kind == "mixed": + rows = _cpu_rows({"node-d": {"grace": None}, "node-p": {"dcgm": DCGM_W}}) + elif kind == "malformed": + rows.append(_cpu_row(FIRST_TS, "node-d", 0, "grace", "n/a")) + elif kind == "version": + rows, header = _v3(rows), CPU_HEADER_V3 + rows[0][0] = 2 + add_cpu_package(pkg, rows, header) + return pkg + + @pytest.mark.parametrize( + "tampered, reason", + [ + ("socket", "cpu_socket_count_mismatch"), + ("gap", "cpu_sample_gap_exceeded"), + ("mixed", "cpu_socket_count_mismatch"), + ("malformed", "cpu_samples_malformed"), + ("version", "cpu_samples_malformed"), + ], + indirect=["tampered"], + ) + def test_cpu_leg_failures_leave_gpu_fields_untouched(self, tmp_path, tampered, reason): + pkg = tampered + # Without an expected CPU source, REQUIRE_POWER gates only the GPU leg. + assert pkg.run(require_power=True) == 0 + agg = pkg.agg() + assert agg["cpu_power_valid"] == 0 + assert set(apm.CPU_METRIC_KEYS).isdisjoint(agg) + assert _gpu_fields(agg) == _reference_agg(tmp_path) + cpu = pkg.sidecar()["cpu"] + assert cpu["cpu_power_valid"] is False + assert reason in cpu["reason_codes"] + assert pkg.sidecar()["power_valid"] is True diff --git a/inferencex-e2e/infx/tests/srt_slurm/test_generate.py b/inferencex-e2e/infx/tests/srt_slurm/test_generate.py index 200015fb41..0159220589 100644 --- a/inferencex-e2e/infx/tests/srt_slurm/test_generate.py +++ b/inferencex-e2e/infx/tests/srt_slurm/test_generate.py @@ -198,6 +198,34 @@ def test_a_multinode_agentx_power_point_gets_its_clusters_exporter_port_and_clie assert recipe["benchmark"]["concurrencies"] == [8] +@pytest.mark.parametrize("declared", [True, False]) +def test_a_cpu_power_point_binds_the_clusters_grace_exporter_or_fails( + project, power_lane, tmp_path, capsys, declared +): + master = project / "configs/master.yaml" + config = yaml.safe_load(master.read_text()) + config["agentx-multi"]["scenarios"]["agentic-coding"][0]["search-space"][0]["cpu-power"] = True + master.write_text(yaml.safe_dump(config)) + if declared: + runners = project / "configs/runners.yaml" + inventory = yaml.safe_load(runners.read_text()) + srt = inventory["clusters"]["fixture"]["slurm"]["srt-slurm"] + srt["cpu-power-exporter"] = {"source": "acpi"} + runners.write_text(yaml.safe_dump(inventory)) + output = tmp_path / "out" + + if not declared: + with pytest.raises(SystemExit): + generate(project, output, "agentx-multi") + assert "declares no srt-slurm.cpu-power-exporter" in capsys.readouterr().err + assert not output.exists() + return + assert generate(project, output, "agentx-multi") == 0 + [(_, recipe)] = written(output) + assert recipe["telemetry"]["cpu_power_exporter"] == {"source": "acpi"} + assert recipe["telemetry"]["dcgm_exporter"]["port"] == 9555 + + def test_a_multinode_agentx_point_without_power_turns_its_clients_power_off( project, power_lane, tmp_path ): diff --git a/inferencex-e2e/infx/tests/srt_slurm/test_workload.py b/inferencex-e2e/infx/tests/srt_slurm/test_workload.py index b2aca470eb..b01668a3b0 100644 --- a/inferencex-e2e/infx/tests/srt_slurm/test_workload.py +++ b/inferencex-e2e/infx/tests/srt_slurm/test_workload.py @@ -94,6 +94,8 @@ def test_bundles_take_the_shared_block_under_base_and_keep_their_variants(projec ({"base": {}, "override_c": {"benchmark": {"concurrencies": [4]}}}, False, False, "override_c.benchmark.concurrencies (= [4])"), ({"telemetry": {"enabled": True}}, False, True, "telemetry.enabled (= True)"), + ({"base": {"telemetry": {"cpu_power_exporter": {"source": "acpi"}}}}, True, True, + "base.telemetry.cpu_power_exporter (= {'source': 'acpi'})"), ({"benchmark": {"env": {"HF_HUB_CACHE": "/elsewhere"}}}, True, True, "benchmark.env.HF_HUB_CACHE (= '/elsewhere')"), ({"base": {"benchmark": {"env": {"RESULT_DIR": "/logs"}}}}, True, False, @@ -168,6 +170,10 @@ def test_a_power_point_gets_telemetry_and_its_concurrencies_on_a_bundle_variant( "dcgm_exporter": {"container_image": "dcgm-exporter", "port": 19401}, } # fmt: skip assert bound["benchmark"]["concurrencies"] == [8] + with_cpu = compose_recipe( + path, agentic=True, multinode=True, root=project, power_port=19401, cpu_power_source="acpi" + ) + assert with_cpu["base"]["telemetry"]["cpu_power_exporter"] == {"source": "acpi"} unpowered = compose_recipe(path, agentic=True, multinode=True, root=project)["base"] assert "concurrencies" not in bind_workload(unpowered, AGENTX_ENV, agentic=True, multinode=True)[ "benchmark" diff --git a/inferencex-e2e/perf-changelog.yaml b/inferencex-e2e/perf-changelog.yaml index f66d1177c7..85376ccbbe 100644 --- a/inferencex-e2e/perf-changelog.yaml +++ b/inferencex-e2e/perf-changelog.yaml @@ -9892,3 +9892,20 @@ - "The new image brings Kimi-K3 ROCm changes vllm-project/vllm#59591 (latent MoE up-proj stores only the local rank's shards), #59069 (AttnRes output fused with per-token FP8 input quantization), #60717 (vLLM's FP8 max in that fused quantization), #60446 (KDA autotune keys), #59523 (cuMem CUDA-graph pool offload on ROCm) and #59880 (speculative block reservations across KV cache groups with a KV connector)." - "No change to the Inferact/Kimi-K3-DSpark draft's precision: K3DSparkModel and get_draft_quant_config() are unchanged between the two images, so the draft keeps BF16 weights and an fp8 KV cache." pr-link: https://github.com/SemiAnalysisAI/InferenceX/pull/3951 + +- config-keys: + - qwen3.5-fp8-gb200-dynamo-sglang + - qwen3.5-fp8-gb300-dynamo-sglang + - kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-dcp16-agg + - kimik3-fp4-gb200-dynamo-vllm-agentic-mooncake-dcp16-agg + - kimik3-fp4-gb200-dynamo-vllm-agentic-dspark-mooncake-tp8pp2 + - kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-disagg + - kimik3-fp4-gb300-dynamo-vllm-agentic-dspark-mooncake-dcp8-agg + scenario-type: + - fixed-seq-len + - agentic-coding + description: + - "Collect ACPI Grace CPU power on the Qwen3.5 GB200/GB300 1P1D, 4P1D, and 8P1D NVL72 recipes and the Kimi K3 GB200/GB300 AgentX recipes. The master rows set cpu-power beside power; the binder merges the cluster's cpu-power-exporter record (ACPI on gb200-nv and gb300-nv) into the bound recipe, and make setup pins cpu-power-exporter v2.51.1 for those jobs, matching the srt-slurm v2.51.1 source pin." + - "Integrate each Grace socket total over the formal GPU measurement window and emit cpu_power_valid, avg_cpu_socket_power_w, avg_total_cpu_power_w, and total_cpu_energy_j independently of the GPU power verdict, with the ACPI CPU-only and SysIO rails as separate keys. When the producer writes module_w (CPU samples schema v3), also emit avg_total_module_power_w and total_module_energy_j as a separate NVML superchip module reference rail; the module includes both GPUs and is never added to the CPU totals. Publish validated per-host GPU power so cross-host workers can be modeled as physical trays. Under required power a cpu-power row fails on a missing or invalid CPU leg; other rows keep the best-effort CPU verdict." + - "Enable required DCGM telemetry with collect_interval_ms: 1000 on the Qwen3.5 GB200/GB300 8P1D recipes, and set require-power on both Qwen3.5 GB200/GB300 8k1k sequences. Keep the Qwen3.5 8P1D frontend, benchmark client, and infra on the reserved head so power and benchmark timestamps share its clock, while retaining 13 allocated nodes and 12 GPU workers." + pr-link: https://github.com/SemiAnalysisAI/InferenceX/pull/3296