diff --git a/.github/AGENT_OPERATIONS.md b/.github/AGENT_OPERATIONS.md index 92aa00e9e9..36a35773f7 100644 --- a/.github/AGENT_OPERATIONS.md +++ b/.github/AGENT_OPERATIONS.md @@ -94,6 +94,8 @@ Single-node fixed-sequence results may include `power_valid`, `avg_power_w`, `av Multinode disaggregated results add `prefill_gpu_energy_j`, `decode_gpu_energy_j`, `prefill_avg_power_w`, `decode_avg_power_w`, `prefill_joules_per_input_token`, and `decode_joules_per_output_token`. Role energy covers the full formal benchmark window, not kernel-level phases, and the role watts are that energy divided by the same window and by the role's GPU count. +NVL72 packages that also carry srt-slurm's `power/cpu/` sub-package add `cpu_power_valid`, `avg_cpu_socket_power_w`, `avg_total_cpu_power_w`, `total_cpu_energy_j`, and, when the module sensor is exposed on every socket, `avg_total_module_power_w` and `total_module_energy_j`. That verdict is independent of `power_valid` and of `REQUIRE_POWER`; see [`docs/results-and-ingestion.md`](../docs/results-and-ingestion.md#measured-grace-cpu-side-power-nvl72). + Every power result — valid or invalid, single-node or multinode — carries `power_metric_schema_version`. Version 2 defines each unprefixed `joules_per_*` field as whole-deployment GPU-board energy over the named denominator; role-scoped energy uses the explicit `prefill_*` / `decode_*` keys. Rows without the field predate the whole-deployment switch and their unprefixed joules are not comparable across topologies. For srt-slurm recipes, `telemetry.enabled: true` with `telemetry.dcgm_exporter` enables official energy collection. The Git submodule pointer at `utils/srt-slurm` is the source of truth for the shared srt-slurm commit, used by both power and non-power NVIDIA lanes. TileRT is the single documented fork exception. 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 recipe-gated `dynamo-sglang` dcgm-power lanes are validated. diff --git a/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml b/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml index e236b684a4..5cdbebea93 100644 --- a/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml +++ b/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb200-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml @@ -132,3 +132,6 @@ telemetry: dcgm_exporter: container_image: dcgm-exporter port: 9401 + cpu_power_exporter: + port: 9405 + source: acpi diff --git a/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml b/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml index b3a68a8e8e..d023950360 100644 --- a/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml +++ b/benchmarks/multi_node/srt-slurm-recipes/qwen3.5/sglang/gb300-fp8/8k1k/disagg-1p1d-tp4-tp4-stp.yaml @@ -134,3 +134,6 @@ telemetry: # 9401 is already bound by the cluster-level exporter on im-gb300 nodes; # use a port outside that range. port: 19401 + cpu_power_exporter: + port: 9405 + source: acpi diff --git a/docs/results-and-ingestion.md b/docs/results-and-ingestion.md index 655c085c37..156bb787b0 100644 --- a/docs/results-and-ingestion.md +++ b/docs/results-and-ingestion.md @@ -101,7 +101,7 @@ InferenceX-app treats routing fields as columns or config dimensions and stores The serving client records `benchmark_outcome` before saving its raw result. It retains the existing maximum request-failure rate of 5%, including the requested/completed/failed counts. The processor verifies this record, copies it to the aggregate, and returns failure even when telemetry is valid. Zero successful requests retain a diagnostic aggregate without fabricated reciprocal latency. Invalid request counts retain a failed diagnostic outcome with the raw `requested`/`completed` values and an `error`, without a fabricated failed count or rate; the client saves the raw JSON before exiting and the processor still rejects it. Legacy results without outcome metadata remain distinguishable; power validity alone never establishes benchmark success or answer quality. -`power_invalid_reasons` and `power_audit` carry a bounded summary alongside numeric metrics. The summary includes the available measurement window, expected/observed GPU counts, sampling diagnostics, observed device identifiers and producer pin. Its `source` names the retained `power_validation_*.json` sidecar. Device identifiers retain the collector's semantics; local SMI indices are not physical UUID proof. +`power_invalid_reasons` and `power_audit` carry a bounded summary alongside numeric metrics. The summary includes the available measurement window, expected/observed GPU counts, sampling diagnostics, observed device identifiers and producer pin. Its `source` names the retained `power_validation_*.json` sidecar. Device identifiers retain the collector's semantics; local SMI indices are not physical UUID proof. When the package carries the Grace CPU-side leg, `power_audit.cpu` adds its sensor kind, source, socket counts, row count, and reason codes; see [Measured Grace CPU-side power (NVL72)](#measured-grace-cpu-side-power-nvl72). For multinode fixed-sequence jobs, `utils/process_result.py --all` processes every available result before returning failure. It accepts `_c_gpus_...`, `_conc_gpus_...`, and AMD `_concurrency__req_rate__gpus_...` filenames, including `inf` request rates. It compares result concurrencies with `CONC_LIST`, rejects duplicate or contradictory point identities, and records omissions/errors in `result_processing_.json`. Aggregate workers pass `AGGREGATE_GPUS` with zero role GPU counts to telemetry validation; separate prefill/decode energy remains absent. For a `DISAGG=true` group with zero decode workers, the aggregate row intentionally sets `disagg: false` and reports `num_aggregate_gpu`; the filename, artifact name, and workflow inputs retain the group identity. Downstream consumers should use the row topology to interpret the measurement. @@ -437,6 +437,55 @@ All four values are withheld when telemetry validation fails. Older results rema missing until their original raw traces can be replayed; average watts cannot supply P75 or P90. The validation sidecar records `power_percentile_method`. +## Measured Grace CPU-side power (NVL72) + +GB200 and GB300 NVL72 recipes that enable srt-slurm's `telemetry.cpu_power_exporter` leg write +`LOGS/power/cpu/samples.csv` and a non-authoritative `cpu_manifest.json` beside the GPU DCGM +package. The multinode validator accepts the pinned v2.2.1 long format (one row per sensor +reading, header `schema_version,timestamp_unix,hostname,source,sensor,socket_id,power_w,total_power_w`) +and the later wide format (one row per scrape, host, and socket, with `cpu_rail_w`, `soc_w`, and +`dram_w` reference columns). It classifies each row by its `sensor` cell: `Module Power Socket N` +is the whole-module reading (Grace, both Blackwell GPUs, HBM, LPDDR5X, and regulator loss); +`Grace Power Socket N` or `CPU:cpuSidePowerUsageW` is the Grace-side socket total (CPU, SoC, +and LPDDR5X); `CPU:cpuPowerUsageW` is DCGM field 1130, the CPU rail only. Component rails +(`CPU Power Socket N`, `SysIO Power Socket N`, DRAM) never feed a published metric. + +Per socket the headline series is chosen in the order module, Grace socket total, DCGM CPU rail, +and one kind must be present for every socket. Every fed series is integrated over the same bound +formal window as GPU energy with the same trapezoid, linear boundary interpolation, and 3.0 s +maximum sample gap. Expected sockets are two per distinct worker host in the manifest topology +(one compute tray per host). The additive keys, rounded like the GPU keys and stripped on re-run, +are: + +| Key | Meaning | +| --- | --- | +| `cpu_power_valid` | `1` or `0` for the CPU-side leg; absent when the package has no `cpu/` | +| `avg_cpu_socket_power_w` | Mean over sockets of each socket's window-mean Grace-side watts | +| `avg_total_cpu_power_w` | Sum over sockets of window-mean Grace-side watts | +| `total_cpu_energy_j` | Grace-side energy over the window, all sockets | +| `avg_total_module_power_w`, `total_module_energy_j` | Module watts and energy; only when a module series exists for every socket | + +The Grace-side keys come from the Grace socket total (or, without ACPI, the DCGM CPU rail) even +when the module sensor is preferred, so a module reading is never published as a Grace-side one. +The sidecar's `cpu` block and the aggregate's `power_audit.cpu` record `sensor_kind` (`module`, +`grace_socket`, or `dcgm_cpu_rail`), `source` (`acpi` or `dcgm`), expected and observed socket +counts, the parsed row count, and reason codes. `power_metric_schema_version` stays `2`. + +The leg is best effort and its verdict is independent of `power_valid`: it borrows only the bound +formal window and the worker-host topology from the GPU package, and no GPU verdict reaches it, so +an unpinned producer or failed GPU coverage withholds GPU energy while `cpu_power_valid` still +judges the CPU samples on their own. Any CPU-leg failure records `cpu_power_valid: 0` with no CPU +keys, leaves every GPU field byte-identical, and never fails `REQUIRE_POWER=1`. Reason codes: +`cpu_samples_missing`, `cpu_samples_header_mismatch`, `cpu_samples_malformed`, +`cpu_manifest_invalid`, `cpu_socket_count_mismatch`, `cpu_sensor_kind_mixed`, +`cpu_sample_gap_exceeded`, `cpu_window_not_bracketed`, and `cpu_window_unavailable` (no completed +window binds to the result, or the window's own contract checks failed). An overflowed CPU +integration reuses the GPU leg's `non_finite_power_metric`, and `aggregate_result_missing` or +`aggregate_result_unwritable` appears in both audits when the aggregate itself cannot be patched. +At the pinned v2.2.1 the exporter does not classify the Module label, +so current packages yield the Grace socket total; module keys appear once the upstream exporter +classifies it. A package without `cpu/` produces an aggregate and sidecar identical to today. + ## Verification and stop conditions A handoff is verified only when all applicable checks pass. diff --git a/docs/results-and-ingestion_zh.md b/docs/results-and-ingestion_zh.md index 3c7bfd7fc1..e287ceb223 100644 --- a/docs/results-and-ingestion_zh.md +++ b/docs/results-and-ingestion_zh.md @@ -101,7 +101,7 @@ InferenceX-app 将路由字段作为列或配置维度,并把数值测量存 服务客户端在保存原始结果前写入 `benchmark_outcome`,保留现有的 5% 最大请求失败率,以及请求总数、完成数和失败数。处理器检查该记录并复制到聚合结果中;即使遥测有效,请求失败率超限仍返回失败。零成功请求会保留诊断聚合结果,但不会生成不存在的延迟倒数。请求计数无效时,失败诊断状态保留原始 `requested`/`completed` 值和 `error`,不生成无依据的失败数或失败率;客户端先保存原始 JSON 再退出,处理器仍拒绝该结果。没有状态元数据的历史结果仍可区分;功耗有效不能证明基准成功或答案质量。 -`power_invalid_reasons` 和 `power_audit` 在数值指标旁携带有界摘要,包括可用的测量窗口、预期与观测 GPU 数、采样诊断、观测设备标识和生产者版本。`source` 指向保留的 `power_validation_*.json` 工件名称。设备标识保留采集器原有语义,本地 SMI 序号不是物理 UUID 的证明。 +`power_invalid_reasons` 和 `power_audit` 在数值指标旁携带有界摘要,包括可用的测量窗口、预期与观测 GPU 数、采样诊断、观测设备标识和生产者版本。`source` 指向保留的 `power_validation_*.json` 工件名称。设备标识保留采集器原有语义,本地 SMI 序号不是物理 UUID 的证明。当产物包包含 Grace CPU 侧测量环节时,`power_audit.cpu` 会附带其传感器类型、来源、socket 数、行数和原因码;见 [NVL72 Grace CPU 侧实测功耗](#nvl72-grace-cpu-侧实测功耗)。 对于多节点固定序列任务,`utils/process_result.py --all` 先处理所有已有结果,再返回失败。它接受 `_c_gpus_...`、`_conc_gpus_...` 和 AMD 的 `_concurrency__req_rate__gpus_...` 文件名,也支持 `inf` 请求速率。它将结果并发度与 `CONC_LIST` 比较,拒绝重复或矛盾的点身份,并将遗漏和错误记录到 `result_processing_.json`。共享工作池通过 `AGGREGATE_GPUS` 及零值角色 GPU 数进行遥测验证;独立的 prefill/decode 能耗保持缺失。当 `DISAGG=true` 的配置组中某个点没有 decode worker 时,聚合行会有意设置 `disagg: false` 并记录 `num_aggregate_gpu`;文件名、工件名和工作流输入仍保留配置组身份。下游应按聚合行的拓扑解释测量结果。 @@ -423,6 +423,49 @@ rm -rf -- "$tmp" 各设备分位数的平均值。遥测验证失败时,四项指标都不发布。旧结果需要使用原始 遥测重新计算;不能从平均功耗推算 P75 或 P90。验证 sidecar 会记录 `power_percentile_method`。 +## NVL72 Grace CPU 侧实测功耗 + +启用了 srt-slurm `telemetry.cpu_power_exporter` 的 GB200 和 GB300 NVL72 recipe 会在 GPU DCGM +产物包旁写入 `LOGS/power/cpu/samples.csv` 和一份非权威的 `cpu_manifest.json`。多节点校验器接受 +当前固定版本 v2.2.1 写出的长格式(每个传感器读数一行,表头为 +`schema_version,timestamp_unix,hostname,source,sensor,socket_id,power_w,total_power_w`),以及后续 +版本的宽格式(每次采集、每台主机、每个 socket 一行,另带 `cpu_rail_w`、`soc_w`、`dram_w` 参考列)。 +它按 `sensor` 单元格对每行分类:`Module Power Socket N` 是整模块读数(Grace、两颗 Blackwell GPU、 +HBM、LPDDR5X 及稳压损耗);`Grace Power Socket N` 或 `CPU:cpuSidePowerUsageW` 是 Grace 侧 socket +总功耗(CPU、SoC 和 LPDDR5X);`CPU:cpuPowerUsageW` 是 DCGM 字段 1130,仅含 CPU 供电轨。组件供电轨 +(`CPU Power Socket N`、`SysIO Power Socket N`、DRAM)永远不会进入发布指标。 + +每个 socket 的主序列按模块、Grace socket 总功耗、DCGM CPU 供电轨的顺序选取,且所有 socket 必须具备 +同一种传感器类型。每条参与计算的序列都在与 GPU 能耗相同的正式测量窗口内积分,采用同样的梯形法、 +线性边界插值和 3.0 秒最大采样间隔。预期 socket 数为 manifest 拓扑中每台不同 worker 主机两个(每台 +主机一个 compute tray)。新增字段与 GPU 字段采用相同的取整规则,重新运行时会先清除旧值: + +| 字段 | 含义 | +| --- | --- | +| `cpu_power_valid` | CPU 侧测量的有效性,`1` 或 `0`;产物包没有 `cpu/` 时不出现 | +| `avg_cpu_socket_power_w` | 各 socket 窗口平均 Grace 侧功耗的均值 | +| `avg_total_cpu_power_w` | 各 socket 窗口平均 Grace 侧功耗之和 | +| `total_cpu_energy_j` | 窗口内所有 socket 的 Grace 侧能耗 | +| `avg_total_module_power_w`、`total_module_energy_j` | 模块功耗与能耗;仅当每个 socket 都有模块序列时输出 | + +即使优先选用模块传感器,Grace 侧字段仍来自 Grace socket 总功耗(没有 ACPI 时来自 DCGM CPU 供电轨), +因此模块读数永远不会被当作 Grace 侧读数发布。sidecar 的 `cpu` 块和聚合结果的 `power_audit.cpu` +记录 `sensor_kind`(`module`、`grace_socket` 或 `dcgm_cpu_rail`)、`source`(`acpi` 或 `dcgm`)、 +预期与观测 socket 数、解析行数和原因码。`power_metric_schema_version` 保持为 `2`。 + +该测量环节尽力而为,其结论与 `power_valid` 相互独立:它只从 GPU 产物包借用已绑定的正式测量窗口和 +worker 主机拓扑,GPU 侧的任何结论都不会传导过来,因此 producer 固定版本校验失败或 GPU 覆盖不足只会 +使 GPU 能耗不予发布,`cpu_power_valid` 仍按 CPU 采样自身给出结论。CPU 侧的任何失败都会记录 +`cpu_power_valid: 0` 且不输出 CPU 字段,所有 GPU 字段保持逐字节不变,也不会使 `REQUIRE_POWER=1` 失败。 +原因码包括 `cpu_samples_missing`、`cpu_samples_header_mismatch`、`cpu_samples_malformed`、 +`cpu_manifest_invalid`、`cpu_socket_count_mismatch`、`cpu_sensor_kind_mixed`、 +`cpu_sample_gap_exceeded`、`cpu_window_not_bracketed` 以及 `cpu_window_unavailable`(没有已完成的窗口 +与该结果绑定,或窗口自身的契约检查失败)。CPU 积分溢出沿用 GPU 侧的 `non_finite_power_metric`;聚合 +结果本身无法写入时,`aggregate_result_missing` 或 `aggregate_result_unwritable` 会同时出现在两侧的审计 +中。在当前固定的 v2.2.1 版本中,exporter 尚不识别 Module 标签,因此现有产物包只会得到 Grace +socket 总功耗;上游 exporter 支持该标签后才会出现模块字段。没有 `cpu/` 的产物包产生的聚合结果和 +sidecar 与现在完全一致。 + ## 验证和停止条件 只有全部适用检查通过,交接才算验证完成。 diff --git a/infx/results/power/__init__.py b/infx/results/power/__init__.py index a396f9684b..6e5fb04510 100644 --- a/infx/results/power/__init__.py +++ b/infx/results/power/__init__.py @@ -34,6 +34,21 @@ "decode_joules_per_output_token", ) ALL_POWER_METRIC_KEYS = WHOLE_METRIC_KEYS + ROLE_METRIC_KEYS +# Additive Grace CPU-side metrics from the multinode cpu/ sub-package. They +# share 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_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( @@ -43,17 +58,23 @@ 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. """ # Do not coerce malformed JSON (such as a list of pairs) into an object. # Leave non-mappings to fail on the same operations as the existing adapters. 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) @@ -61,8 +82,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/infx/results/power/audit.py b/infx/results/power/audit.py index 443b18f3e5..a42134d852 100644 --- a/infx/results/power/audit.py +++ b/infx/results/power/audit.py @@ -7,6 +7,36 @@ from collections.abc import Mapping from typing import Any +from .cpu_side import HEADLINE_PREFERENCE + +_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] = {} + kind = cpu.get("sensor_kind") + if kind in HEADLINE_PREFERENCE: + summary["sensor_kind"] = kind + 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 +79,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/infx/results/power/common.py b/infx/results/power/common.py index 671f215989..f9e333f801 100644 --- a/infx/results/power/common.py +++ b/infx/results/power/common.py @@ -53,6 +53,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]], *, @@ -242,6 +252,8 @@ def patch_power_metrics( metric_keys: Iterable[str], power_valid: bool, metrics: Mapping[str, float], + cpu_power_valid: bool | None = None, + cpu_metrics: Mapping[str, float] | None = None, ) -> None: """Validate replacement metrics before atomically updating an aggregate.""" data = json.loads(path.read_text(encoding="utf-8")) @@ -251,5 +263,7 @@ def patch_power_metrics( schema_version=POWER_METRIC_SCHEMA_VERSION, power_valid=power_valid, metrics=metrics, + cpu_power_valid=cpu_power_valid, + cpu_metrics=cpu_metrics, ) _write_json_atomic(path, data) diff --git a/infx/results/power/cpu_side.py b/infx/results/power/cpu_side.py new file mode 100644 index 0000000000..ca9dc07016 --- /dev/null +++ b/infx/results/power/cpu_side.py @@ -0,0 +1,323 @@ +"""Best-effort Grace CPU-side power leg of the multinode srt-slurm package. + +srt-slurm's ``cpu_power_exporter`` leg writes ``power/cpu/samples.csv`` and a +non-authoritative ``power/cpu/cpu_manifest.json`` beside the GPU DCGM package +(srt-slurm ``src/srtctl/core/power/cpu_session.py``). The pinned v2.2.1 +producer writes the v1 long format, one row per sensor reading; v2.15.0 +writes the v2 wide format, one row per (scrape, host, socket) with the ACPI +component rails as reference columns. Both are accepted here. + +Each (hostname, socket) headline 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 leg is +best-effort by contract: any failure records ``cpu_power_valid=0`` with +reason codes and leaves every GPU field untouched, and ``REQUIRE_POWER`` +never fails a run because of it. A package without ``cpu/`` emits nothing. +""" + +from __future__ import annotations + +import csv +import itertools +import json +import math +import re +from collections.abc import Iterable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .common import _append_reason, _bracketing_sequence, _integrate_device, audit_metrics + +CPU_DIRNAME = "cpu" +CPU_SAMPLES_FILENAME = "samples.csv" +CPU_MANIFEST_FILENAME = "cpu_manifest.json" + +# srt-slurm contract.py: CPU_SAMPLES_HEADER_V1 (written by v2.2.1) and the +# wide CPU_SAMPLES_HEADER (written by v2.15.0, which reads both). +CPU_SAMPLES_HEADER_V1 = ( + "schema_version", + "timestamp_unix", + "hostname", + "source", + "sensor", + "socket_id", + "power_w", + "total_power_w", +) +CPU_SAMPLES_HEADER_V2 = ( + *CPU_SAMPLES_HEADER_V1[:7], + "cpu_rail_w", + "soc_w", + "dram_w", + "total_power_w", +) +_HEADERS = {1: CPU_SAMPLES_HEADER_V1, 2: CPU_SAMPLES_HEADER_V2} + +# One GB200/GB300 compute tray per worker host, two Grace sockets per tray. +SOCKETS_PER_HOST = 2 +# Shared with the GPU leg (srt-slurm contract.MAX_SAMPLE_GAP_SECONDS). +MAX_SAMPLE_GAP_SECONDS = 3.0 + +SENSOR_MODULE = "module" +SENSOR_GRACE = "grace_socket" +SENSOR_DCGM = "dcgm_cpu_rail" +# Headline preference per socket. Component rails (cpu_rail, soc, dram) are +# reference breakdowns and never feed a published metric. +HEADLINE_PREFERENCE = (SENSOR_MODULE, SENSOR_GRACE, SENSOR_DCGM) +# Grace-side kinds, in fallback order, that feed the ``*_cpu_*`` keys. +_GRACE_SIDE_PREFERENCE = (SENSOR_GRACE, SENSOR_DCGM) + +# ``CPU:`` sensor names (srt-slurm cpu_rails.SENSOR_SUFFIXES) +# and firmware OEM labels (srt-slurm cpu_rails.ACPI_LABEL_PATTERNS for the +# Grace total). The Module label is the NVL72 tray firmware's whole-module +# sensor, which srt-slurm does not classify yet; a v2.2.1 package therefore +# yields the Grace socket total until the exporter learns the label. +_SUFFIX_KINDS = {"cpuSidePowerUsageW": SENSOR_GRACE, "cpuPowerUsageW": SENSOR_DCGM} +_LABEL_KINDS = ( + (SENSOR_MODULE, re.compile(r"\bModule\s+Power\s+Socket\s+\d+\b", re.IGNORECASE)), + (SENSOR_GRACE, re.compile(r"\bGrace\s+Power\s+Socket\s+\d+\b", re.IGNORECASE)), + ( + SENSOR_GRACE, + re.compile(r"\bTotal(?:\s+Input)?\s+Power(?:\s+in\s+uW)?\s+Socket\s+\d+\b", re.IGNORECASE), + ), +) + +_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 + + +@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, + "headline_preference": list(HEADLINE_PREFERENCE), + "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 classify_sensor(sensor: str) -> str | None: + """Headline kind for a ``sensor`` cell from either writer version, else None.""" + _, _, suffix = sensor.partition(":") + if suffix in _SUFFIX_KINDS: + return _SUFFIX_KINDS[suffix] + return next((kind for kind, pattern in _LABEL_KINDS if pattern.search(sensor)), None) + + +def _parse_row(raw: list[str], version: int) -> CpuSampleRow | None: + if len(raw) != len(_HEADERS[version]): + return None + try: + schema_version = int(raw[0]) + timestamp_unix = float(raw[1]) + socket_id = int(raw[5]) + power_w = float(raw[6]) + except ValueError: + return None + hostname, source, sensor = raw[2], raw[3], raw[4] + 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 + return CpuSampleRow( + timestamp_unix=timestamp_unix, + hostname=hostname, + source=source, + sensor=sensor, + socket_id=socket_id, + power_w=power_w, + ) + + +def read_cpu_samples(path: Path) -> tuple[tuple[CpuSampleRow, ...], tuple[str, ...]]: + """Strictly parse either sample generation; 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 = next(reader, None) + version = next((v for v, h in _HEADERS.items() if header == list(h)), None) + if version is None: + return (), ("cpu_samples_header_mismatch",) + for raw in reader: + row = _parse_row(raw, 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) + + +def _manifest_is_object(path: Path) -> bool: + try: + return isinstance(json.loads(path.read_text(encoding="utf-8")), dict) + except (OSError, UnicodeDecodeError, ValueError): + return False + + +SocketKey = tuple[str, int] + + +def _headline_series( + rows: Iterable[CpuSampleRow], +) -> tuple[dict[SocketKey, dict[str, dict[float, float]]], dict[str, set[str]], bool]: + """Group headline rows as (host, socket) -> kind -> {timestamp: watts}.""" + series: dict[SocketKey, dict[str, dict[float, float]]] = {} + sources: dict[str, set[str]] = {} + duplicate = False + for row in rows: + kind = classify_sensor(row.sensor) + if kind is None: + continue + points = series.setdefault((row.hostname, row.socket_id), {}).setdefault(kind, {}) + if row.timestamp_unix in points: + duplicate = True + points[row.timestamp_unix] = row.power_w + sources.setdefault(kind, set()).add(row.source) + return series, sources, duplicate + + +def _select_feeds(common_kinds: set[str]) -> tuple[str | None, dict[str, str]]: + """Pick the headline kind and which kind feeds each metric family.""" + selected = next((kind for kind in HEADLINE_PREFERENCE if kind in common_kinds), None) + if selected is None: + return None, {} + if selected != SENSOR_MODULE: + return selected, {selected: "cpu"} + feeds = {SENSOR_MODULE: "module"} + grace = next((kind for kind in _GRACE_SIDE_PREFERENCE if kind in common_kinds), None) + if grace is not None: + feeds[grace] = "cpu" + return selected, feeds + + +def validate_cpu_leg( + cpu_dir: Path, + *, + window: tuple[float, float] | None, + expected_hosts: Iterable[str], +) -> CpuPowerAudit | None: + """Validate and integrate the CPU sub-package; None when the run carried none.""" + if not cpu_dir.is_dir(): + return None + audit = CpuPowerAudit() + if not _manifest_is_object(cpu_dir / CPU_MANIFEST_FILENAME): + audit.invalidate("cpu_manifest_invalid") + rows, sample_reasons = read_cpu_samples(cpu_dir / CPU_SAMPLES_FILENAME) + audit.sample_row_count = len(rows) + audit.invalidate(*sample_reasons) + series, sources, duplicate = _headline_series(rows) + if duplicate: + 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 + common_kinds = set.intersection(*(set(kinds) for kinds in series.values())) + selected, feeds = _select_feeds(common_kinds) + if selected is None or len(sources[selected]) != 1: + audit.invalidate("cpu_sensor_kind_mixed") + return audit + audit.sensor_kind = selected + audit.source = next(iter(sources[selected])) + + start, end = window + energy: dict[str, float] = {} + for (host, socket), by_kind in sorted(series.items()): + for kind, family in feeds.items(): + label = f"{host}/socket{socket}/{kind}" + samples = sorted(by_kind[kind].items()) + sequence = _bracketing_sequence(tuple(t for t, _ in samples), start, end) + if sequence is None: + audit.invalidate("cpu_window_not_bracketed") + continue + gap = max( + (later - earlier for earlier, later in itertools.pairwise(sequence)), + default=0.0, + ) + audit.per_series_max_sample_gap_s[label] = gap + if gap > MAX_SAMPLE_GAP_SECONDS: + 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: dict[str, float] = {} + if "cpu" in energy: + metrics["avg_cpu_socket_power_w"] = energy["cpu"] / duration / len(expected_keys) + metrics["avg_total_cpu_power_w"] = energy["cpu"] / duration + metrics["total_cpu_energy_j"] = energy["cpu"] + if "module" in energy: + metrics["avg_total_module_power_w"] = energy["module"] / duration + metrics["total_module_energy_j"] = energy["module"] + 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/infx/results/power/multinode.py b/infx/results/power/multinode.py index 756449b85a..c46758c978 100644 --- a/infx/results/power/multinode.py +++ b/infx/results/power/multinode.py @@ -24,6 +24,12 @@ ``power_valid=0`` (and no energy metrics) in the aggregate plus a validation sidecar, but never fails the benchmark. Power studies set ``REQUIRE_POWER=1`` to fail after those audit artifacts exist. + +Grace CPU-side leg: when the package also carries ``power/cpu/`` (see +:mod:`.cpu_side`), each socket's Grace-side (and, when exposed, whole-module) +power is integrated over the same bound formal window and published under +``cpu_power_valid``. That verdict is independent of ``power_valid`` and of +``REQUIRE_POWER``; a package without ``cpu/`` produces identical output. """ from __future__ import annotations @@ -44,12 +50,14 @@ # import them through this module. from . import ( ALL_POWER_METRIC_KEYS as _ALL_POWER_METRIC_KEYS, + CPU_METRIC_KEYS as CPU_METRIC_KEYS, ROLE_METRIC_KEYS as ROLE_METRIC_KEYS, WHOLE_METRIC_KEYS as WHOLE_METRIC_KEYS, ) from .common import ( BenchmarkData, _append_reason, + _bracketing_sequence, _integrate_device, _load_benchmark_data, _percentile_total_power, @@ -58,6 +66,7 @@ benchmark_window_payload, patch_power_metrics, ) +from .cpu_side import CPU_DIRNAME, CpuPowerAudit, validate_cpu_leg # --- srt-slurm dcgm-power v1 wire contract (mirrored constants) ------------- @@ -711,15 +720,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, @@ -896,6 +896,11 @@ class MultinodePowerAudit: observed_gpu_count: int = 0 expected_gpu_count: int = 0 metrics: dict[str, float] = field(default_factory=dict) + # Inputs the CPU-side leg borrows from the GPU package; not serialized. + expected_worker_hosts: tuple[str, ...] = () + formal_window_trusted: bool = False + # None when the package carried no cpu/ sub-package. + cpu: CpuPowerAudit | None = None def _canonical_sha256(payload: dict) -> str: @@ -921,7 +926,45 @@ def validate_and_integrate( aggregate_gpus: int, expected_producer_sha: str | None, ) -> MultinodePowerAudit: - """Recompute package validity, cross-check verdicts, and integrate energy.""" + """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 audit.formal_window_trusted: + 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, + ) + 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: audit = MultinodePowerAudit( expected_producer_git_commit=expected_producer_sha, ) @@ -972,6 +1015,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] @@ -1078,6 +1122,8 @@ def validate_and_integrate( window = _select_window_for_result( audit, parsed_windows, logs_root, bench_result_path, benchmark ) + if window is not None: + audit.formal_window_trusted = _window_contract_holds(window, validations) if audit.reasons or window is None or benchmark is None: return audit @@ -1235,18 +1281,47 @@ def _select_window_for_result( return window +# Per-window audit codes that describe GPU sample coverage rather than the +# window's own contract; the CPU-side leg re-checks its own coverage. +_GPU_COVERAGE_REASONS = frozenset( + {"measurement_window_not_bracketed", "sample_gap_exceeded", "gpu_uuid_changed"} +) + + +def _window_contract_holds(window: ParsedWindow, 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 _patch_agg(agg_path: Path, audit: MultinodePowerAudit) -> None: + cpu = audit.cpu patch_power_metrics( agg_path, 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, ) +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( *, audit: MultinodePowerAudit, @@ -1254,7 +1329,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), @@ -1284,6 +1359,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( @@ -1322,20 +1401,17 @@ def run( 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, @@ -1359,6 +1435,16 @@ 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 not audit.power_valid: print( f"[aggregate_power_multinode] Power validation failed: " diff --git a/perf-changelog.yaml b/perf-changelog.yaml index 9419a68f0b..f5574820bc 100644 --- a/perf-changelog.yaml +++ b/perf-changelog.yaml @@ -8531,3 +8531,17 @@ - "为 GB300 vLLM DeepSeek-V4.1-Flash AgentX 配方在现有 TP4 臂旁新增 TP2 臂,Engram 表继续通过 --engram-config cpu_offload 放在固定页主机 DRAM;每张 277 GiB GPU 的权重升至约 175 GiB" - "在 dsv41flash_fp4_vllm_mtp.sh 中将 B200 TP2 的上限推广到所有 TP2 臂:--max-num-batched-tokens 4096(上游 16384 时 indexer 的 [batched-tokens, 1M] fp8 缓冲区达 32 GiB)、--max-num-seqs 为并发的两倍(16-256)、CUDA graph 捕获上限 512,为每张 GPU 留出约 36 GiB KV;TP4 与 TP8 臂沿用上游默认值" pr-link: https://github.com/SemiAnalysisAI/InferenceX/pull/3321 + +- config-keys: + - qwen3.5-fp8-gb200-dynamo-sglang + - qwen3.5-fp8-gb300-dynamo-sglang + scenario-type: + - fixed-seq-len + description: + - "Enable srt-slurm's independent, best-effort CPU power leg on the 1P1D TP4/TP4 STP Qwen3.5 FP8 GB200 and GB300 recipes: telemetry.cpu_power_exporter with port 9405 (the schema default) and source acpi. Nothing else in either recipe changes; image, serving flags, topology, concurrency list, and the GPU DCGM exporter (port 9401 on GB200, 19401 on GB300, required: true) are untouched." + - "At the pinned srt-slurm v2.2.1 (984180e5) srtctl launches cpu-power-exporter --port 9405 --source acpi through srun on the bare host of every worker node, falling back to the ACPI-only Python exporter when make setup did not install the release binary, and the head-node collector writes LOGS/power/cpu/samples.csv plus cpu_manifest.json beside the GPU DCGM package. The leg has no required flag: a missing hwmon sensor or a dead exporter only leaves gaps in samples.csv and cannot fail the job or the GPU power verdict." + - "The multinode power validator integrates the Grace Power Socket series over the same formal window as GPU energy and adds cpu_power_valid plus the CPU-side keys (docs/results-and-ingestion.md, Measured Grace CPU-side power); power_valid and every GPU field stay unchanged. The v2.2.1 exporter classifies only the CPU, Grace, and SysIO socket labels, so these runs yield the Grace-side total rather than the module reading." + - "These are the smallest active non-Kimi-K3 NVL72 power lanes (2 nodes, 1 prefill + 1 decode host, 4 Grace sockets). The DSV4 8k1k GB200/GB300 lanes are not used: Single-turn 8k1k on DeepSeek-V4-Pro was deprecated after 2026-09-08 (MODELS.md; configs/deprecated/nvidia-dsv4-8k1k-master.yaml), so no sweep would exercise them." + - "在 Qwen3.5 FP8 GB200 与 GB300 的 1P1D TP4/TP4 STP recipe 上启用 srt-slurm 独立、尽力而为的 CPU 功耗测量环节:telemetry.cpu_power_exporter,port 9405(schema 默认值),source acpi。两份 recipe 其余内容不变:镜像、服务参数、拓扑、并发列表以及 GPU DCGM exporter(GB200 端口 9401、GB300 端口 19401,required: true)均未改动。在固定版本 srt-slurm v2.2.1(984180e5)下,srtctl 通过 srun 在每个 worker 节点的裸主机上启动 cpu-power-exporter --port 9405 --source acpi,若 make setup 未安装发布的二进制则回退到仅支持 ACPI 的 Python exporter;头节点采集器把 LOGS/power/cpu/samples.csv 和 cpu_manifest.json 写在 GPU DCGM 产物包旁。该环节没有 required 开关:缺少 hwmon 传感器或 exporter 失效只会让 samples.csv 出现空缺,不会导致作业失败或改变 GPU 功耗结论。多节点功耗校验器在与 GPU 能耗相同的正式测量窗口内对 Grace Power Socket 序列积分,新增 cpu_power_valid 与 CPU 侧字段(见 docs/results-and-ingestion_zh.md「NVL72 Grace CPU 侧实测功耗」);power_valid 与所有 GPU 字段保持不变。v2.2.1 的 exporter 只识别 CPU、Grace、SysIO 三种 socket 标签,因此这些运行得到的是 Grace 侧总功耗而非模块读数。" + - "选择这两条 lane 是因为它们是当前活跃、非 Kimi-K3 的最小 NVL72 功耗 lane(2 个节点,1 台 prefill 主机 + 1 台 decode 主机,4 个 Grace socket)。未采用 DSV4 8k1k 的 GB200/GB300 lane:DeepSeek-V4-Pro 的单轮 8k1k 场景已于 2026-09-08 之后弃用(见 MODELS.md 与 configs/deprecated/nvidia-dsv4-8k1k-master.yaml),sweep 不会再执行它们。" + pr-link: https://github.com/SemiAnalysisAI/InferenceX/pull/3296 diff --git a/utils/srt-slurm b/utils/srt-slurm index 984180e5b8..473ecaa8fb 160000 --- a/utils/srt-slurm +++ b/utils/srt-slurm @@ -1 +1 @@ -Subproject commit 984180e5b8755aef85e9995048b5a16cb5336bce +Subproject commit 473ecaa8fbd1d409867d492c0271ba154e1dcaa4 diff --git a/utils/test_aggregate_power_multinode.py b/utils/test_aggregate_power_multinode.py index 3235b88ebb..b7b17468ea 100644 --- a/utils/test_aggregate_power_multinode.py +++ b/utils/test_aggregate_power_multinode.py @@ -653,3 +653,295 @@ def test_v2_samples_reject_mixed_versions_and_invalid_utilization(tmp_path, row) rows, reasons = apm.read_samples(path) assert not rows assert reasons == ("samples_csv_malformed",) + + +# --- Grace CPU-side power leg (power/cpu/) ----------------------------------- + +CPU_HOSTS = tuple(sorted({host for host, *_ in DEVICES})) +CPU_HEADER_V1 = [ + "schema_version", "timestamp_unix", "hostname", "source", "sensor", + "socket_id", "power_w", "total_power_w", +] +CPU_HEADER_V2 = CPU_HEADER_V1[:7] + ["cpu_rail_w", "soc_w", "dram_w", "total_power_w"] +GRACE_W = {"node-d": 500.0, "node-p": 700.0} +MODULE_W = 1500.0 +DCGM_W = 300.0 +# Component rails never feed a headline metric; the values are chosen so any +# leak into a published key is visible against the Grace/module constants. +RAILS_W = {"cpu_rail": 200.0, "soc": 50.0, "dram": 30.0} +GRACE_KINDS = {"grace": None, **RAILS_W} + +_SENSORS = { + # (source, v1 firmware OEM label, v2 collector sensor name) + "grace": ("acpi", "Grace Power Socket {s}", "CPU{s}:cpuSidePowerUsageW"), + "module": ("acpi", "Module Power Socket {s}", "Module Power Socket {s}"), + "dcgm": ("dcgm", "CPU{s}:cpuPowerUsageW", "CPU{s}:cpuPowerUsageW"), + "cpu_rail": ("acpi", "CPU Power Socket {s}", "CPU{s}:cpuRailPowerUsageW"), + "soc": ("acpi", "SysIO Power Socket {s}", "CPU{s}:socPowerUsageW"), + "dram": ("acpi", "DRAM Power Socket {s}", "CPU{s}:dramPowerUsageW"), +} + + +def _cpu_row(ts, host, socket, kind, watts, fmt): + source, label_v1, label_v2 = _SENSORS[kind] + sensor = (label_v1 if fmt == "v1" else label_v2).format(s=socket) + if fmt == "v1": + return [1, repr(ts), host, source, sensor, socket, repr(watts), ""] + rails = ["", "", ""] if source == "dcgm" else [repr(w) for w in RAILS_W.values()] + return [2, repr(ts), host, source, sensor, socket, repr(watts), *rails, ""] + + +def _cpu_rows(kinds_by_host=None, fmt="v1", *, 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_KINDS 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, fmt)) + ts += 1.0 + return rows + + +def add_cpu_package(pkg, rows, header=CPU_HEADER_V1, manifest_text=None): + 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) + if manifest_text is None: + manifest_text = json.dumps( + { + "schema_version": 1, + "producer": "srt-slurm.cpu-power", + "producer_git_commit": PRODUCER_SHA, + "started_at_unix": 990.0, + "stopped_at_unix": 1070.0, + "nodes": { + host: {"resolved_mode": "acpi", "scrape_count": 65, "error_count": 0} + for host in CPU_HOSTS + }, + }, + indent=2, + ) + (cpu_dir / "cpu_manifest.json").write_text(manifest_text) + return cpu_dir + + +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: + @pytest.mark.parametrize("fmt, header", [("v1", CPU_HEADER_V1), ("v2", CPU_HEADER_V2)]) + def test_grace_total_emits_cpu_keys_and_provenance(self, tmp_path, fmt, header): + from infx.results.power.audit import audit_summary + + pkg = build_package(tmp_path) + kinds = None if fmt == "v1" else {host: {"grace": None} for host in CPU_HOSTS} + rows = _cpu_rows(kinds, fmt) + add_cpu_package(pkg, rows, header) + assert pkg.run(require_power=True) == 0 + + agg = pkg.agg() + assert_grace_keys(agg) + assert "avg_total_module_power_w" not in agg + assert "total_module_energy_j" not in agg + 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}/grace_socket" for host in CPU_HOSTS for socket in (0, 1) + } + 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_dcgm_only_package_uses_cpu_rail_kind(self, tmp_path): + pkg = build_package(tmp_path) + add_cpu_package(pkg, _cpu_rows({host: {"dcgm": DCGM_W} for host in CPU_HOSTS})) + assert pkg.run() == 0 + agg = pkg.agg() + assert agg["cpu_power_valid"] == 1 + assert agg["avg_cpu_socket_power_w"] == 300.0 + assert agg["avg_total_cpu_power_w"] == 1200.0 + assert agg["total_cpu_energy_j"] == 72000.0 + cpu = pkg.sidecar()["cpu"] + assert (cpu["sensor_kind"], cpu["source"]) == ("dcgm_cpu_rail", "dcgm") + + def test_module_on_every_socket_is_preferred_and_grace_keys_stay_grace(self, tmp_path): + pkg = build_package(tmp_path) + add_cpu_package( + pkg, _cpu_rows({host: {**GRACE_KINDS, "module": MODULE_W} for host in CPU_HOSTS}) + ) + assert pkg.run() == 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 + cpu = pkg.sidecar()["cpu"] + assert cpu["sensor_kind"] == "module" + 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 ("module", "grace_socket") + } + + def test_module_on_some_sockets_falls_back_to_grace(self, tmp_path): + pkg = build_package(tmp_path) + kinds = {"node-d": {**GRACE_KINDS, "module": MODULE_W}, "node-p": GRACE_KINDS} + add_cpu_package(pkg, _cpu_rows(kinds)) + assert pkg.run() == 0 + agg = pkg.agg() + assert_grace_keys(agg) + assert "avg_total_module_power_w" not in agg + assert "total_module_energy_j" not in agg + assert pkg.sidecar()["cpu"]["sensor_kind"] == "grace_socket" + + 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 test_stale_cpu_keys_are_stripped_on_rerun(self, tmp_path): + pkg = build_package(tmp_path) + pkg.agg_result.write_text( + json.dumps({"hw": "gb200", "conc": 4, "cpu_power_valid": 1, "total_cpu_energy_j": 1.0}) + ) + assert pkg.run() == 0 + assert _gpu_fields(pkg.agg()) == _reference_agg(tmp_path) + assert "cpu_power_valid" not in pkg.agg() + assert "total_cpu_energy_j" not in pkg.agg() + + 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, manifest_text = CPU_HEADER_V1, None + if kind == "header": + header = [*CPU_HEADER_V1[:6], "watts", "total_power_w"] + elif 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 == "unbracketed": + rows = self._drop(rows, "node-d", 0, lambda ts: ts <= WINDOW_START) + elif kind == "mixed": + rows = _cpu_rows({"node-d": {"grace": None}, "node-p": {"dcgm": DCGM_W}}) + elif kind == "manifest": + manifest_text = "{broken" + elif kind == "malformed": + rows = [*rows, [1, repr(FIRST_TS), "node-d", "acpi", "Grace Power Socket 0", 0, "n/a", ""]] + cpu_dir = add_cpu_package(pkg, rows, header, manifest_text) + if kind == "samples_missing": + (cpu_dir / "samples.csv").unlink() + return pkg + + @pytest.mark.parametrize( + "tampered, reason", + [ + ("header", "cpu_samples_header_mismatch"), + ("socket", "cpu_socket_count_mismatch"), + ("gap", "cpu_sample_gap_exceeded"), + ("unbracketed", "cpu_window_not_bracketed"), + ("mixed", "cpu_sensor_kind_mixed"), + ("manifest", "cpu_manifest_invalid"), + ("malformed", "cpu_samples_malformed"), + ("samples_missing", "cpu_samples_missing"), + ], + indirect=["tampered"], + ) + def test_cpu_leg_failures_leave_gpu_fields_untouched(self, tmp_path, tampered, reason): + pkg = tampered + # REQUIRE_POWER guards the GPU leg only; a broken CPU leg never fails the run. + 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 + + @pytest.mark.parametrize( + "gpu_gap, sha, gpu_reason", + [ + (True, PRODUCER_SHA, "package_recompute_invalid"), + (False, "b" * 40, "producer_commit_mismatch"), + (False, None, "producer_pin_missing"), + ], + ) + def test_cpu_leg_survives_an_invalid_gpu_leg(self, tmp_path, gpu_gap, sha, gpu_reason): + """No GPU verdict, the producer pin included, reaches cpu_power_valid.""" + pkg = build_package(tmp_path) + add_cpu_package(pkg, _cpu_rows()) + if gpu_gap: + _rewrite_samples( + pkg, + lambda body: [ + row + for row in body + if not (row[3] == "node-p" and row[4] == "0" and 20 <= int(row[2]) <= 24) + ], + ) + assert pkg.run(sha=sha) == 0 + agg = pkg.agg() + assert agg["power_valid"] == 0 + assert "total_gpu_energy_j" not in agg + assert_grace_keys(agg) + sidecar = pkg.sidecar() + assert gpu_reason in sidecar["reasons"] + assert sidecar["cpu"]["reason_codes"] == [] + + def test_window_unavailable_when_result_binds_to_no_window(self, tmp_path): + pkg = build_package(tmp_path) + add_cpu_package(pkg, _cpu_rows()) + tampered = dict(BENCH_FIELDS, max_concurrency=8) + pkg.original_result.write_text(json.dumps(tampered, indent=2)) + pkg.bench_result.write_text(json.dumps(tampered, indent=2)) + assert pkg.run() == 0 + assert pkg.agg()["cpu_power_valid"] == 0 + assert "cpu_window_unavailable" in pkg.sidecar()["cpu"]["reason_codes"]