diff --git a/.gitignore b/.gitignore index c33f2a170..e4935f04f 100644 --- a/.gitignore +++ b/.gitignore @@ -51,6 +51,7 @@ bin/ configs/nats-server configs/etcd configs/etcdctl +configs/process-exporter configs/*.whl configs/*.deb configs/*.tar.gz diff --git a/Makefile b/Makefile index ada226969..648ced42e 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,7 @@ NATS_VERSION ?= v2.10.28 ETCD_VERSION ?= v3.5.21 +PROCESS_EXPORTER_VERSION ?= 0.8.7 LOGS_DIR ?= logs ARCH ?= $(shell uname -m) TACHOMETER_RELEASE ?= latest @@ -128,6 +129,26 @@ setup: tachometer-scraper-download echo "✅ ETCD installed to configs/etcd"; \ fi; \ echo ""; \ + echo "--- process-exporter $(PROCESS_EXPORTER_VERSION) (Tachometer per-process/thread telemetry) ---"; \ + if [ -f configs/process-exporter ] && file configs/process-exporter | grep -q "$$ARCH_FILE_PATTERN"; then \ + echo "✅ process-exporter already installed at configs/process-exporter ($(ARCH))"; \ + else \ + echo "⬇️ Downloading process-exporter ($(PROCESS_EXPORTER_VERSION)) for $$ARCH_SHORT..."; \ + PE_NAME="process-exporter-$(PROCESS_EXPORTER_VERSION).linux-$$ARCH_SHORT"; \ + PE_TAR="$$PE_NAME.tar.gz"; \ + PE_URL="https://github.com/ncabatoff/process-exporter/releases/download/v$(PROCESS_EXPORTER_VERSION)/$$PE_TAR"; \ + if ! wget -q --show-progress --tries=3 --waitretry=5 "$$PE_URL" -O "configs/$$PE_TAR"; then \ + rm -f "configs/$$PE_TAR"; \ + echo "❌ Failed to download process-exporter from $$PE_URL"; \ + exit 1; \ + fi; \ + echo "📁 Extracting process-exporter binary..."; \ + tar -xzf "configs/$$PE_TAR" --strip-components=1 -C configs "$$PE_NAME/process-exporter"; \ + chmod +x configs/process-exporter; \ + rm "configs/$$PE_TAR"; \ + echo "✅ process-exporter installed to configs/process-exporter"; \ + fi; \ + echo ""; \ echo "--- uv (compute node arch: $(ARCH)) ---"; \ if [ -f bin/uv ] && file bin/uv | grep -q "$$ARCH_FILE_PATTERN"; then \ echo "✅ uv already installed at bin/uv ($(ARCH))"; \ diff --git a/docs/config-reference.md b/docs/config-reference.md index dd661e32c..18e7fe765 100644 --- a/docs/config-reference.md +++ b/docs/config-reference.md @@ -1277,7 +1277,7 @@ The legacy in-job Python RAW scraper is retired: a recipe still carrying `scrape The component perf dashboard is **not** configured here. It is built in post-processing on every run; `enabled` decides which capture legs exist and therefore which tabs the page carries. See [Component Performance Dashboard](component-dashboard.md). -Tachometer collects every worker rank, frontend, DCGM, and node metrics by default (minus the client-polled complement described above) — the exporters launch from pinned multi-arch registry images with no configuration. Air-gapped clusters override the images via the `containers:` alias map in `srtslurm.yaml`; `default_exporters: false` disables the built-ins: +Tachometer collects every worker rank, frontend, DCGM, node, and process metrics by default (minus the client-polled complement described above) — the exporters launch from pinned multi-arch registry images with no configuration. Air-gapped clusters override the images via the `containers:` alias map in `srtslurm.yaml`; `default_exporters: false` disables the built-ins: ```yaml observability: @@ -1296,6 +1296,10 @@ observability: node_exporter: container_image: /containers/node-exporter.sqsh port: 9100 + process_exporter: + binary: /opt/srt/configs/process-exporter # host-native (default mode); or set container_image instead + container_image: "" + port: 9256 ``` | Tachometer field | Type | Default | Description | @@ -1307,11 +1311,14 @@ observability: | `compaction_threads` | int | `4` | Value passed as `POLARS_MAX_THREADS` | | `storage_subdir` | string | `tachometer` | Output directory below the run log directory | | `extra_metadata` | dict | `{}` | Static string metadata added to every endpoint | -| `default_exporters` | bool | `true` | Launch the built-in DCGM + node exporters when no explicit blocks are set (sweep path only) | +| `default_exporters` | bool | `true` | Launch the built-in DCGM + node + process exporters when no explicit blocks are set (sweep path only) | | `dcgm_exporter` | object/null | built-in | Defaults to `nvcr.io#nvidia/k8s/dcgm-exporter:3.3.9-3.6.1-ubuntu22.04` on port 9401; an explicit block overrides | -| `node_exporter` | object/null | built-in | Defaults to `quay.io#prometheus/node-exporter:v1.8.2` on port 9101; an explicit block overrides | +| `node_exporter` | object/null | built-in | Defaults to `quay.io#prometheus/node-exporter:v1.8.2` on port 9101 with `--collector.disable-defaults` plus the `cpu`, `infiniband`, `meminfo`, `stat` (`node_procs_running/blocked`, context switches), `vmstat` (with `--collector.vmstat.fields` widened to include `pgsteal_*` next to `pgmajfault`), `pressure` (PSI; absent on kernels without `CONFIG_PSI`), `meminfo_numa` (`node_memory_numa_*`, kept per NUMA node by the scraper as `numa_node=N`) and `processes` (`node_processes_threads`, per-state `node_processes_state`) collectors; an explicit block overrides | +| `process_exporter` | object/null | built-in | Defaults to the **host-native** `configs/process-exporter` binary (ncabatoff/process-exporter 0.8.7, installed by `make setup` for the compute arch, like `configs/nats-server` and `configs/etcd`) on port 9256, launched with plain `srun` (no container) on every node that hosts a backend rank or a frontend replica. Reads the host `/proc` and publishes per-process-group CPU seconds by mode, thread count, per-thread-name CPU and count (`-threads=true`), context switches, RSS and open fds. Groups (frontend, `dynamo_trtllm` / `dynamo_sglang` / `dynamo_vllm` handlers + engine ranks, launcher, client, infra daemons) come from `/process-exporter.yml`, written at launch. If the binary is missing the leg is skipped with a warning (submit warns too). An explicit block may set `binary` (absolute, or relative to the srtctl checkout) or instead a `container_image` with `binary` unset to run it containerized; the upstream `FROM scratch` image is not used by default because pyxis/enroot on some clusters cannot start shell-less images | -`make setup ARCH=` downloads and checksum-verifies the matching Tachometer binary from the latest srt-slurm release. The scraper runs as a native `srun` process on the head node; configured exporters remain containerized on worker nodes. Run `make tachometer-scraper` to build from source instead. +Every exporter block accepts `container_image`, `port`, `command` and `binary`. `binary` selects host-native launch (the executable runs directly under `srun`, `container_image` is ignored and may be `""`); without it the exporter runs from `container_image`. One of the two must be set. + +`make setup ARCH=` downloads and checksum-verifies the matching Tachometer binary from the latest srt-slurm release and installs the process-exporter binary for the same arch. The scraper and the process exporter run as native `srun` processes; the DCGM and node exporters remain containerized on worker nodes. Run `make tachometer-scraper` to build the scraper from source instead. Tachometer writes its Parquet stream under `//raw/scrape/` (the leaf is created by the scraper itself — srtctl pre-creates only the parent, because the scraper refuses a pre-existing storage directory), compacting to `final.parquet` there on shutdown. Intermediate files remain in `//local` until shutdown compaction completes. Rows carry an epoch `timestamp_ns` column, so they join directly with AIPerf records and Dynamo spans; the post-processing ingest converts the Parquet into the dashboard's `server_metrics_export.jsonl`. diff --git a/docs/host-attribution-metrics.md b/docs/host-attribution-metrics.md new file mode 100644 index 000000000..9d26e7567 --- /dev/null +++ b/docs/host-attribution-metrics.md @@ -0,0 +1,80 @@ +# Host attribution metrics: pinning vs frontend placement + +srt-slurm's host sampler collects the telemetry needed to answer, **from a +baseline run alone**, two questions that end-to-end serving metrics cannot: + +1. Are worker ranks losing CPU time to scheduler contention/migration? + (remedy: `backend.numa_cpu_bind: true`) +2. Is the frontend/etcd interfering with the workers sharing its node? + (remedy: `frontend.dedicated_node` / `infra.etcd_nats_dedicated_node`) + +Both effects are real and independently worth ~1% output throughput each at +high concurrency on GB300 disaggregated serving — but they are invisible in +throughput/TTFT alone, which is why the collectors below exist. + +## Collection + +With `observability.enabled: true`, the `/proc` host sampler runs on the +orchestrator node (in-process) **and on every other allocated node** +(`observability.host_sampler_all_nodes`, default true: one persistent +`srun --overlap` per node group running `host_sampler.py` standalone). Each +node writes `host_samples_.jsonl` into the run's log dir; the ingest +merges them into `host_series.json` with a per-node `hosts` map. This closes +the previous gaps: worker nodes had no per-process host telemetry, and a +dedicated frontend node had none at all. + +## Metric set + +Per sampled process (workers, frontend, benchmark client — matched by cmdline): + +| Field (raw JSONL) | Ingest series | Diagnoses | Points at | +|---|---|---|---| +| `run_delay_ns` (`/proc/pid/schedstat`) | `run_delay_ms_per_s` | Task runnable but not running: scheduler contention on its cores | pinning | +| `nr_migrations` (`/proc/pid/sched`) | `migrations_rate` | Cross-core churn; near-zero when pinned | pinning | +| `affinity_ncpus` (`sched_getaffinity`) | `affinity_ncpus` | Direct pinning-state observable (144 = floating, 36 = pinned rank on GB200/GB300) | pinning (config state) | +| `ctx_invol` (`/proc/pid/status`) | `ctx_invol_rate` | Involuntary descheduling (lock convoys, neighbor pressure) | pinning / placement | +| `cpu_jiffies` | `cpu_pct` | Per-process CPU use — splits a shared node's load into frontend vs etcd vs ranks | placement | +| host `procs_running/blocked` (`/proc/stat`) | `procs_runnable` | Whole-node run-queue pressure vs core count | either (localizes with the per-process rows) | +| `t` + `t_mono` | — | Per-node clock-offset estimation; cross-node wall clocks have been observed seconds apart | metric hygiene | + +## Decision rubric (thresholds from the c1010 validation matrix, GB300/oci-aga) + +You are looking at `host_series.json` from ONE run. You do not need to know what +"taskset" or "frontend placement" are — the rubric names the config change. + +**Step 1 — is the bottleneck host-CPU-side at all?** +Look at the per-node `procs` map for the busiest process per worker node +(highest `cpu_pct`). If every worker node shows `run_delay_ms_per_s` p50 +< 0.1 and `migrations_rate` ≈ 0, host-CPU scheduling is NOT the problem — +stop here. (Validated: clean nodes sit at 0.00–0.01 ms/s.) + +**Step 2 — check the pinning state directly.** +`affinity_ncpus` of the worker ranks equals the node's full logical-CPU count +(e.g. 144 or 288) → the ranks are NOT pinned. Remedy: +`backend.numa_cpu_bind: true`. Validated effect at c1010: +4.0% output +throughput on 288-CPU GB300 nodes (+1.1% on 144-CPU nodes in the reference +campaign — the gain grows with core count). If `affinity_ncpus` equals +(CPUs ÷ GPUs per node), the ranks are already pinned. + +**Step 3 — look for the single-node asymmetry.** +Compare each worker rank's `run_delay_ms_per_s` p50 against the median of its +peers on other nodes. Threshold: **>10× the peer median AND >0.5 ms/s absolute, +on exactly the node(s) that also host a non-worker process with +`affinity_ncpus` = full width and `cpu_pct` > 1000** (the frontend: measured +~4,100–5,000% of one core at c1010). That is co-location interference. +Remedy: `frontend.dedicated_node: true`. Validated effect: +0.5% throughput +(+0.85% in the reference campaign) — and the asymmetry itself is huge even +when the throughput cost is small: measured 140–350× on the shared node, +collapsing to 1× in all three dedicated-frontend runs. +Note the dissociation, confirmed both ways across 7 runs: this asymmetry is +UNCHANGED by pinning (190× with ranks pinned), and `affinity_ncpus` is +UNCHANGED by moving the frontend. Each signal names exactly one remedy. + +**Expected-gain estimate**: single-node asymmetry affecting 1 of N prefill +groups → small-percent gain (≈ its share of prefill capacity); full-width +affinity on all ranks → the pinning gain for your node's core count. + +Cross-node timing comparisons must estimate per-node clock offsets first +(pair `t` with `t_mono`, or use a constant frontend→worker dispatch offset); +raw cross-node wall-clock deltas are unreliable at millisecond scale — +observed inter-node skew up to 2.1 s. diff --git a/docs/profiling.md b/docs/profiling.md index 1ae328d50..959a9fec4 100644 --- a/docs/profiling.md +++ b/docs/profiling.md @@ -109,17 +109,103 @@ Profiling has specific requirements: ### nsys-specific behavior -When using `nsys`, workers are wrapped with: +**TRT-LLM workers (`backend.type: trtllm`)** are wrapped with the capture recipe from the Dynamo +Benchmark Playbook §9.5.1.1 "Dynamo + TRTLLM", the flag set that produced every usable multi-node +capture on nsys 2026.3.x / VR200 disaggregated serving: ```bash -nsys profile -t cuda,nvtx --cuda-graph-trace=node \ - -c cudaProfilerApi --capture-range-end stop \ - [extra_nsys_args...] \ - -o /logs/profiles/{mode}/{name} \ - python3 -m sglang.launch_server ... +nsys profile --force-overwrite=true \ + -t cuda-sw,nvtx,python-gil --cuda-graph-trace=graph \ + --sample=none --cpuctxsw=none --python-sampling=false --python-sampling-frequency=1000 \ + --gpu-metrics-devices=none --flush-on-cudaprofilerstop=false --cuda-flush-interval=0 \ + -c cudaProfilerApi --capture-range-end=stop \ + [extra_nsys_args...] --kill none --wait all \ + -o /logs/profiles/{mode}/{leader}_{mode}_w{index}_profile_rank%q{SLURM_PROCID} \ + trtllm-llmapi-launch python3 -m dynamo.trtllm ... ``` -You can pass extra arguments via `profiling.extra_nsys_args` (e.g. `["--stats=true", "--trace=osrt"]`). +One `nsys` runs per srun task, i.e. per engine rank; `%q{SLURM_PROCID}` is nsys's own env-var +substitution, expanded inside the launched process, so each rank gets its own `.nsys-rep`. +The capture window is the TRT-LLM iteration range from `prefill`/`decode` `start_step`/`stop_step` +(`TLLM_PROFILE_START_STOP`): the PyExecutor calls `cudaProfilerStart`/`cudaProfilerStop` on every rank +at those iterations and `--capture-range-end=stop` finalises the report the moment the range closes. +Reports appear under `/profiles//` while the benchmark is still running. + +Why each flag (from the playbook): + +| Flag | Reason | +| ---- | ------ | +| `-t cuda-sw`, not `cuda` | on nsys 2026.x plain `cuda` selects the HES hardware trace, which SIGSEGVs the KV transceiver's device-to-device `cudaMemcpyAsync`; `cuda-sw` forces the software tracer | +| `--cuda-graph-trace=graph` | decode runs CUDA graphs; with `node` the launches are invisible (~0.3 % GPU busy reads as idle), with `graph` each iteration is one entry and utilisation is real | +| `--sample=none --cpuctxsw=none` | `--sample=process-tree` wedged workers at `cudaProfilerStop` across 8 runs. Cost: no scheduler/CPU-time data, every duration is wall-clock. `nsys_cpuctxsw: process-tree` alone (context switches, no IP sampling) is the safer experiment; both need `kernel.perf_event_paranoid <= 2` on the compute node | +| `-c cudaProfilerApi --capture-range-end=stop` | finalise-at-process-exit has never produced a report on this stack | +| `--flush-on-cudaprofilerstop=false --cuda-flush-interval=0` | matches the known-good captures; nsys 2026.4 flips the flush default, so it is set explicitly | +| `--gpu-metrics-devices=none` | a second collection path with its own failure modes | + +Worker environment set alongside: `TLLM_PROFILE_START_STOP=-`, `TLLM_LLMAPI_ENABLE_NVTX=1`, +`TLLM_PROFILE_LOG_RANKS=` (default `all`), `DYN_ENABLE_RUST_NVTX=1` (Dynamo's Rust NVTX +ranges; effective only on a wheel built with the `nvtx` cargo feature) and, when +`nvtx_injection_path` is set, `NVTX_INJECTION64_PATH` pointing at nsys's injection library inside the +container (TRT-LLM containers ship it next to nsys, e.g. +`/usr/local/cuda-0.gpgpu/NsightSystems-cli-2026.3.0/target-linux-sbsa-armv8/libToolsInjection64.so`). + +Knobs (all optional, defaults = the recipe above): `nsys_trace`, `nsys_cuda_graph_trace` (`graph`|`node`), +`nsys_sample` and `nsys_cpuctxsw` (`none`|`process-tree`|`system-wide`), `nsys_python_sampling`, +`nsys_python_sampling_frequency`, `nsys_gpu_metrics_devices`, `nvtx_injection_path`, `log_ranks`, +plus `extra_nsys_args` (appended before `-o`). `nsys-time` uses the same trace flags with +`--delay`/`--duration` instead of the cudaProfilerApi window. The `profiling:` block is independent of +`observability.enabled`, so the capture cost can be measured against a default-visibility run. + +Sizing the window: `start_step` counts engine iterations, which begin with traffic, not with worker +start. On the 8-node AgentX recipe the decode worker ran ~20 iterations/s and each prefill worker +~2.7/s in steady state, so `decode: 6000-6600` and `prefill: 1200-1300` both capture about 30-40 s +roughly 12 minutes into the benchmark. + +**SGLang / vLLM workers** keep the time-based or `--trace-fork-before-exec` prefixes described in the +examples below; `extra_nsys_args` applies to them as well. + +### Profiling the Dynamo frontend too (`profiling.frontend`) + +The worker windows are iteration based (cudaProfilerApi). The Dynamo frontend does no CUDA +work and never calls `cudaProfilerStart`, so it gets a time window instead: `profiling.frontend` +wraps every `python3 -m dynamo.frontend` process in +`nsys profile -t --delay --duration --kill none --wait all` +and exports `DYN_ENABLE_RUST_NVTX=1` (plus `NVTX_INJECTION64_PATH` when `nvtx_injection_path` is +set) so Dynamo's Rust NVTX ranges are recorded. Reports land in +`/logs/profiles/frontend/_frontend_.nsys-rep`; nsys stays attached for the whole run +and the frontend keeps serving after the window closes. + +```yaml +profiling: + type: nsys + frontend: + delay_secs: 1500 # seconds after the frontend starts (workers need ~10-15 min to load first) + duration_secs: 120 + trace: nvtx # or "nvtx,osrt"; CUDA tracing is pointless on this process + prefill: {start_step: 1200, stop_step: 1300} + decode: {start_step: 6000, stop_step: 6600} +``` + +Requires `frontend.type: dynamo` and an nsys profiling type; other frontends are not wrapped. + +### Time-windowed sessions and the srun task (`nsys-time`, `profiling.frontend`) + +`nsys profile --delay D --duration T --kill none ` writes its report when the window closes and then +**exits**, leaving the application running as an orphan. Under Slurm the exiting nsys is the srun task, so the +step ends and slurmstepd kills the orphan — on hecate job 595056 the frontend and the decode workers died three +seconds after their reports were written. srtctl therefore wraps every time-windowed launch (the profiled +frontend, and workers in `nsys-time` mode) so the task waits for the profiled process after nsys exits +(`srtctl.core.nsys_keepalive`). Iteration-based captures need no wrapper: nsys stays attached until the engine exits. + +### Teardown grace for open capture ranges (`profiling.teardown_grace_secs`) + +nsys writes a report when its capture range closes. If the range is still open when the run ends — +`stop_step` was never reached, or an `nsys-time` duration outlasts the benchmark — the report is written +only after the engine exits, and the default 10-second SIGTERM→SIGKILL grace that srtctl gives every +process loses it. When `profiling.type` is `nsys`/`nsys-time`, srtctl therefore waits +`profiling.teardown_grace_secs` (default 180) after SIGTERM before SIGKILL for the nsys-wrapped worker +sruns and the profiled frontend. The waits overlap (every process is signalled first, then each is +reaped), so a run pays the grace once, not once per worker. Unprofiled processes keep the 10 s grace. ## Example Configurations diff --git a/src/ingest/ingest.py b/src/ingest/ingest.py index a6cb27ea3..d85e90105 100644 --- a/src/ingest/ingest.py +++ b/src/ingest/ingest.py @@ -1007,13 +1007,43 @@ def run_host_samples(run_dir: Path, bundle: Path) -> dict: single core, so >100 means the process is genuinely using more than one. """ src = Path(run_dir) / "host_samples.jsonl" - if not src.exists(): - _log("L2 host", "no host_samples.jsonl; host CPU / fd / client-bottleneck " + per_node = sorted(Path(run_dir).glob("host_samples_*.jsonl")) + if not src.exists() and not per_node: + _log("L2 host", "no host_samples*.jsonl; host CPU / fd / client-bottleneck " "signals unavailable for this run") return {} + out: dict = {} + if src.exists(): + out = _host_series_from_rows(_read_host_rows(src)) or {} + + # Per-node samplers (observability.host_sampler_all_nodes) write one file per + # node; keyed by hostname so worker nodes and a dedicated frontend node are + # separable downstream. The orchestrator-node series stays at the top level + # for backward compatibility with existing consumers. + hosts: dict[str, dict] = {} + for path in per_node: + series = _host_series_from_rows(_read_host_rows(path)) + if series: + hosts[series["host"] or path.stem.removeprefix("host_samples_")] = series + if hosts: + out.setdefault("hosts", {}).update(hosts) + if not out: + _log("L2 host", "host sample files present but none had >= 2 rows") + return {} + + with open(bundle / "host_series.json", "w") as f: + json.dump(out, f) + peak_cpu = max((v for _, v in out.get("host_cpu_pct", [])), default=None) + _log("L2 host", f"{out.get('samples', 0)} samples on {out.get('host')} (+{len(hosts)} remote node(s)): " + f"peak host CPU {peak_cpu}%, " + f"{len(out.get('procs', {}))} process(es) tracked") + return out + + +def _read_host_rows(path: Path) -> list[dict]: rows = [] - with open(src, errors="replace") as fh: + with open(path, errors="replace") as fh: for line in fh: line = line.strip() if not line: @@ -1022,14 +1052,24 @@ def run_host_samples(run_dir: Path, bundle: Path) -> dict: rows.append(json.loads(line)) except Exception: continue + return rows + + +def _host_series_from_rows(rows: list[dict]) -> dict | None: + """Difference cumulative counters into rate series for one node's samples.""" if len(rows) < 2: - _log("L2 host", f"only {len(rows)} host sample(s); a rate needs two") - return {} + return None - host_cpu, fds, conns, mem = [], [], [], [] + host_cpu, fds, conns, mem, runq, blocked_series = [], [], [], [], [], [] procs: dict = {} for prev, cur in zip(rows, rows[1:]): - dt = (cur.get("t") or 0) - (prev.get("t") or 0) + # Rate denominators prefer the monotonic clock: NTP steps can make + # wall-clock dt negative (pair dropped) or inflated (rates deflated). + # Wall-clock t stays as the series x-axis for cross-source alignment. + if cur.get("t_mono") is not None and prev.get("t_mono") is not None: + dt = cur["t_mono"] - prev["t_mono"] + else: + dt = (cur.get("t") or 0) - (prev.get("t") or 0) if dt <= 0: continue t = cur["t"] @@ -1042,19 +1082,35 @@ def run_host_samples(run_dir: Path, bundle: Path) -> dict: mem.append([t, round(100.0 * (1 - m["MemAvailable"] / m["MemTotal"]), 2)]) if cur.get("established_conns") is not None: conns.append([t, cur["established_conns"]]) + if cur.get("procs_running") is not None: + runq.append([t, cur["procs_running"]]) + if cur.get("procs_blocked") is not None: + blocked_series.append([t, cur["procs_blocked"]]) prev_by_pid = {p["pid"]: p for p in (prev.get("procs") or [])} for p in cur.get("procs") or []: q = prev_by_pid.get(p["pid"]) key = f"{p.get('name', 'proc')}:{p['pid']}" e = procs.setdefault(key, {"cpu_pct": [], "rss_kb": [], "ctx_invol_rate": [], - "open_fds": [], "threads": []}) + "open_fds": [], "threads": [], + "run_delay_ms_per_s": [], "migrations_rate": [], + "affinity_ncpus": []}) if q and p.get("cpu_jiffies") is not None and q.get("cpu_jiffies") is not None: # Jiffies are 1/100 s on Linux; /dt gives percent of ONE core. e["cpu_pct"].append([t, round((p["cpu_jiffies"] - q["cpu_jiffies"]) / dt, 1)]) if q and p.get("ctx_invol") is not None and q.get("ctx_invol") is not None: e["ctx_invol_rate"].append( [t, round((p["ctx_invol"] - q["ctx_invol"]) / dt, 1)]) + if q and p.get("run_delay_ns") is not None and q.get("run_delay_ns") is not None: + # ms of run-queue wait accumulated per wall second: the direct + # scheduler-contention rate that CPU pinning is the remedy for. + e["run_delay_ms_per_s"].append( + [t, round((p["run_delay_ns"] - q["run_delay_ns"]) / 1e6 / dt, 2)]) + if q and p.get("nr_migrations") is not None and q.get("nr_migrations") is not None: + e["migrations_rate"].append( + [t, round((p["nr_migrations"] - q["nr_migrations"]) / dt, 1)]) + if p.get("affinity_ncpus") is not None: + e["affinity_ncpus"].append([t, p["affinity_ncpus"]]) if p.get("rss_kb") is not None: e["rss_kb"].append([t, p["rss_kb"]]) if p.get("open_fds") is not None: @@ -1067,10 +1123,12 @@ def run_host_samples(run_dir: Path, bundle: Path) -> dict: fd_limit = rows[-1].get("fd_limit") peak_fds = max((v for _, v in fds), default=0) - out = { + return { "host_cpu_pct": host_cpu, "host_mem_used_pct": mem, "established_conns": conns, + "procs_runnable": runq, + "procs_blocked": blocked_series, "open_fds_total": fds, "fd_limit": fd_limit, # The number that matters for PERF-40: how close the run came to the ceiling. @@ -1082,14 +1140,6 @@ def run_host_samples(run_dir: Path, bundle: Path) -> dict: "samples": len(rows), "host": rows[-1].get("host"), } - with open(bundle / "host_series.json", "w") as f: - json.dump(out, f) - peak_cpu = max((v for _, v in host_cpu), default=None) - _log("L2 host", f"{len(rows)} samples on {out['host']}: peak host CPU {peak_cpu}%, " - f"peak open fds {peak_fds}/{fd_limit} " - f"({out['fd_headroom_pct']}% of limit), " - f"{len(procs)} process(es) tracked") - return out def run_provenance(run_dir: Path, bundle: Path) -> list[str]: diff --git a/src/srtctl/analysis/host_sampler.py b/src/srtctl/analysis/host_sampler.py index 1e26fd21b..9e8e1456f 100644 --- a/src/srtctl/analysis/host_sampler.py +++ b/src/srtctl/analysis/host_sampler.py @@ -34,12 +34,15 @@ import json import logging import os +import re import threading import time from pathlib import Path logger = logging.getLogger(__name__) +_NR_MIGRATIONS_RE = re.compile(r"se\.nr_migrations\s*:\s*(\d+)") + # Process names worth sampling individually. Matched as substrings against the # process's own cmdline, so a wrapper script does not hide the real one. _PROC_PATTERNS = ("dynamo", "aiperf", "http-server", "trtllm") @@ -73,6 +76,25 @@ def _cpu_totals() -> tuple[int, int] | None: return None +def _procs_running_blocked() -> tuple[int | None, int | None]: + """(procs_running, procs_blocked) from ``/proc/stat``. + + Instantaneous run-queue pressure for the whole host: sustained + procs_running above the core count is contention no per-process + counter can localize. + """ + txt = _read("/proc/stat") or "" + running = blocked = None + for line in txt.splitlines(): + if line.startswith("procs_running"): + with contextlib.suppress(IndexError, ValueError): + running = int(line.split()[1]) + elif line.startswith("procs_blocked"): + with contextlib.suppress(IndexError, ValueError): + blocked = int(line.split()[1]) + return running, blocked + + def _meminfo() -> dict: txt = _read("/proc/meminfo") or "" out = {} @@ -84,6 +106,36 @@ def _meminfo() -> dict: return out +def _pressure() -> dict: + """PSI stall totals from ``/proc/pressure/{cpu,memory,io}``. + + PSI is the signal that separates "this box is busy" from "real work spent + time BLOCKED waiting for a resource" -- the question the sampler exists to + answer, and the one metric the retired steady_probe.sh flagged as most + important. Node_exporter's pressure collector covers worker nodes, but the + exporters are launched on backend nodes only; the sampler runs on the + orchestrator/head node, so this read is that node's only PSI source. + + Reported as cumulative ``total=`` microseconds (like the cpu jiffies above): + the consumer differences consecutive rows over any window it likes. Absent + on kernels without ``CONFIG_PSI`` -> the resource key is simply omitted. + """ + out: dict = {} + for resource in ("cpu", "memory", "io"): + txt = _read(f"/proc/pressure/{resource}") + if not txt: + continue + for line in txt.splitlines(): + kind = line.split(" ", 1)[0] # "some" or "full" + if kind not in ("some", "full"): + continue + for field in line.split(): + if field.startswith("total="): + with contextlib.suppress(ValueError): + out[f"{resource}_{kind}_total_us"] = int(field.split("=", 1)[1]) + return out + + # Launcher processes whose cmdline mentions a worker without BEING one. On a real node # `srun` appears once per launched process, so a naive cmdline match spends most of the # budget on wrappers: observed 14 of 24 slots on theia0019 (job 2753007), crowding out @@ -107,8 +159,11 @@ def _interesting_pids(limit: int = 32) -> list[int]: entries = os.listdir("/proc") except OSError: return [] + own_pid = os.getpid() for e in entries: - if not e.isdigit(): + if not e.isdigit() or int(e) == own_pid: + # Never sample self: the standalone sampler's own cmdline carries the + # log-dir path, which routinely contains a pattern word ("dynamo"). continue cmd = _read(f"/proc/{e}/cmdline") if not cmd or not any(p in cmd for p in _PROC_PATTERNS): @@ -143,6 +198,28 @@ def _proc_sample(pid: int) -> dict | None: # utime+stime, fields 14/15 after the (possibly space-containing) comm field. tail = stat[stat.rfind(")") + 2 :].split() d["cpu_jiffies"] = int(tail[11]) + int(tail[12]) + schedstat = _read(f"/proc/{pid}/schedstat") + if schedstat: + with contextlib.suppress(IndexError, ValueError): + parts = schedstat.split() + # Field 2 is cumulative RUN-QUEUE WAIT: time the task was runnable but + # not running. THE scheduler-contention signal — a rising rate here with + # idle CPUs elsewhere means this task is losing its core to neighbors, + # which is exactly what CPU pinning remedies. + d["cpu_ns"] = int(parts[0]) + d["run_delay_ns"] = int(parts[1]) + sched = _read(f"/proc/{pid}/sched") + if sched: + m = _NR_MIGRATIONS_RE.search(sched) + if m: + # Cross-core migrations: near-zero when pinned, so this doubles as a + # direct observable of whether pinning is in effect AND of scheduler + # churn when it is not. + d["nr_migrations"] = int(m.group(1)) + with contextlib.suppress(OSError): + # Allowed-CPU count: 144 = free-floating on GB200/GB300, 36 = taskset-pinned + # rank. Reads the pinning CONFIG state directly, no inference needed. + d["affinity_ncpus"] = len(os.sched_getaffinity(pid)) try: d["open_fds"] = len(os.listdir(f"/proc/{pid}/fd")) except OSError: @@ -216,13 +293,21 @@ def _loop(self, stop_event: threading.Event) -> None: while not stop_event.is_set() and not self._own_stop.is_set(): try: cpu = _cpu_totals() + running, blocked = _procs_running_blocked() row = { "t": time.time(), + # Monotonic companion: cross-node wall clocks can disagree by + # seconds (observed 2.1s NVL72 skew); pairs of (t, t_mono) let + # the consumer estimate per-node offsets post-hoc. + "t_mono": time.monotonic(), "host": os.uname().nodename, "cpu_busy_jiffies": cpu[0] if cpu else None, "cpu_total_jiffies": cpu[1] if cpu else None, + "procs_running": running, + "procs_blocked": blocked, "loadavg": (_read("/proc/loadavg") or "").split()[:3], "mem": _meminfo(), + "psi": _pressure(), "fd_limit": fd_limit, "established_conns": _established_connections(), "procs": [s for s in (_proc_sample(p) for p in _interesting_pids()) if s], @@ -260,3 +345,69 @@ def try_start_host_sampler(log_dir: Path, observability, stop_event: threading.E except Exception as exc: # noqa: BLE001 - best effort logger.warning("Host sampler failed to start (continuing without it): %s", exc) return None + + +def plan_remote_sampler_nodes(all_nodes: list[str], local_host: str) -> list[str]: + """Nodes that need a standalone sampler: every allocated node except the one + already covered by the in-process sampler. Pure so it is unit-testable. + + ``local_host`` may be a short hostname while the nodelist carries FQDNs (or + vice versa); match on the first dot-separated label to be safe. + """ + local_label = local_host.split(".")[0] + seen: set[str] = set() + out: list[str] = [] + for node in all_nodes: + if not node or node in seen: + continue + seen.add(node) + if node.split(".")[0] != local_label: + out.append(node) + return out + + +def _main() -> int: + """Standalone per-node entry point (``python3 host_sampler.py --log-dir D``). + + Runs on bare compute nodes with nothing but a system python3: this module + deliberately imports only the standard library. Writes + ``host_samples_.jsonl`` so per-node files never collide on the + shared log dir, and exits cleanly on SIGTERM/SIGINT (srun teardown). + """ + import argparse + import signal + + ap = argparse.ArgumentParser(description="Standalone /proc sampler for one node") + ap.add_argument("--log-dir", required=True) + ap.add_argument("--interval", type=float, default=2.0, help="seconds between samples (floored to 1.0)") + ap.add_argument("--max-samples", type=int, default=0, help="stop after N samples (0 = until signaled)") + args = ap.parse_args() + + logging.basicConfig(level=logging.INFO, format="[host_sampler %(levelname)s] %(message)s") + stop = threading.Event() + for sig in (signal.SIGTERM, signal.SIGINT): + signal.signal(sig, lambda *_: stop.set()) + + sampler = HostSampler( + Path(args.log_dir), + interval_seconds=args.interval, + output_name=f"host_samples_{os.uname().nodename}.jsonl", + ) + sampler.start(stop) + while not stop.is_set(): + if args.max_samples and sampler.samples >= args.max_samples: + stop.set() + break + if sampler._thread is not None and not sampler._thread.is_alive(): + # Sampler thread died (e.g. unwritable log dir): exit nonzero so the + # failure is visible in the srun step output instead of idling forever. + logger.error("sampler thread exited after %d samples; aborting", sampler.samples) + sampler.stop() + return 1 + stop.wait(0.5) + sampler.stop() + return 0 + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/src/srtctl/cli/mixins/benchmark_stage.py b/src/srtctl/cli/mixins/benchmark_stage.py index 31367f960..b87f68816 100644 --- a/src/srtctl/cli/mixins/benchmark_stage.py +++ b/src/srtctl/cli/mixins/benchmark_stage.py @@ -8,7 +8,9 @@ """ import logging +import os import shlex +import subprocess import threading import time from pathlib import Path @@ -420,28 +422,34 @@ def _run_benchmark_script( # truthy, and plain truthiness would silently switch it on there. observability = getattr(self.config, "observability", None) host_sampler = None + remote_samplers: list[subprocess.Popen] = [] if getattr(observability, "enabled", False) is True: from srtctl.analysis.host_sampler import try_start_host_sampler host_sampler = try_start_host_sampler(self.runtime.log_dir, observability, stop_event) - - bench_node = self._benchmark_node() - proc = start_srun_process( - command=cmd, - nodelist=[bench_node], - output=str(log_file), - container_image=str(container_image), - container_mounts=container_mounts, - env_to_set=env_to_set, - srun_options=self.runtime.srun_options, - het_group=self.runtime.nodes.het_group_for(bench_node), - ) + if getattr(observability, "host_sampler_all_nodes", False) is True: + remote_samplers = self._start_remote_host_samplers() # The signal handler raises SystemExit, so only finally can establish # how the local srun client stopped before telemetry finalizes. + # proc is created INSIDE the try: _benchmark_node()/start_srun_process can + # raise (bad client_placement, fork failure), and the finally must still + # tear down the remote samplers launched above. self.benchmark_child_reaped = False self.benchmark_child_allows_window_mutation = False + proc: subprocess.Popen | None = None try: + bench_node = self._benchmark_node() + proc = start_srun_process( + command=cmd, + nodelist=[bench_node], + output=str(log_file), + container_image=str(container_image), + container_mounts=container_mounts, + env_to_set=env_to_set, + srun_options=self.runtime.srun_options, + het_group=self.runtime.nodes.het_group_for(bench_node), + ) while proc.poll() is None: if stop_event.is_set(): logger.info("Stop requested, terminating benchmark") @@ -451,7 +459,7 @@ def _run_benchmark_script( self.benchmark_child_allows_window_mutation = True return proc.returncode or 0 finally: - if proc.poll() is None: + if proc is not None and proc.poll() is None: outcome = terminate_and_reap( proc, terminate_timeout=_BENCHMARK_TERMINATE_TIMEOUT, @@ -461,7 +469,7 @@ def _run_benchmark_script( # Reaping a force-killed local srun client does not prove that # its remote Slurm step can no longer write the window. self.benchmark_child_allows_window_mutation = outcome.reaped and not outcome.force_killed - elif self.benchmark_child_reaped is False: + elif proc is not None and self.benchmark_child_reaped is False: proc.wait() self.benchmark_child_reaped = True self.benchmark_child_allows_window_mutation = True @@ -469,6 +477,79 @@ def _run_benchmark_script( snapshotter.stop() if host_sampler is not None: host_sampler.stop() + for sampler_proc in remote_samplers: + if sampler_proc.poll() is not None: + if sampler_proc.returncode != 0: + # e.g. a bare node without python3: contained, but say so + # instead of letting the gap surface as missing files later. + logger.warning( + "A remote host sampler exited early (rc=%s); see host_sampler_remote*.out", + sampler_proc.returncode, + ) + continue + # SIGTERM lets the remote sampler flush its final JSONL row; + # terminate_and_reap escalates to SIGKILL and logs if it wedges, + # so a hung srun can't keep appending rows past the window. + terminate_and_reap(sampler_proc, terminate_timeout=10, kill_timeout=5) + + def _start_remote_host_samplers(self) -> list[subprocess.Popen]: + """Launch the standalone /proc sampler on every allocated node except this one. + + One persistent srun per het group (not per sample), no container: the + sampler is stdlib-only and reads the HOST /proc either way, and the + srtctl checkout lives on a shared filesystem the compute nodes see. + Best-effort like all host telemetry — a node without python3 logs a + failure into host_sampler_remote.out and the benchmark proceeds. + """ + from srtctl.analysis import host_sampler as host_sampler_module + from srtctl.analysis.host_sampler import plan_remote_sampler_nodes + + nodes = self.runtime.nodes + all_nodes = list(dict.fromkeys([nodes.head, nodes.bench, nodes.infra, *nodes.worker])) + targets = plan_remote_sampler_nodes(all_nodes, os.uname().nodename) + if not targets: + return [] + + script = Path(host_sampler_module.__file__).resolve() + groups: dict[int | None, list[str]] = {} + for node in targets: + groups.setdefault(nodes.het_group_for(node), []).append(node) + + # Same cadence as the in-process sampler (try_start_host_sampler): the + # single scrape knob, observability.tachometer.collect_interval_ms, so + # head-node and remote rows line up at the same rate. HostSampler clamps + # to >= 1 s on the receiving side; mirror it here so the CLI value is + # the one actually used. + tachometer = getattr(self.config.observability, "tachometer", None) + interval_ms = getattr(tachometer, "collect_interval_ms", 1000) if tachometer else 1000 + interval = f"{max(1.0, interval_ms / 1000.0):g}" + + procs: list[subprocess.Popen] = [] + launched_nodes = 0 + for het_group, group_nodes in sorted(groups.items(), key=lambda kv: (kv[0] is not None, kv[0])): + suffix = "" if het_group is None else f".g{het_group}" + try: + proc = start_srun_process( + command=["python3", str(script), "--log-dir", str(self.runtime.log_dir), "--interval", interval], + nodes=len(group_nodes), + ntasks=len(group_nodes), + nodelist=group_nodes, + output=str(self.runtime.log_dir / f"host_sampler_remote{suffix}.out"), + srun_options=self.runtime.srun_options, + het_group=het_group, + use_bash_wrapper=False, + ) + except Exception as exc: # noqa: BLE001 - best effort, never blocks the benchmark + logger.warning("Remote host samplers failed to start on %s: %s", group_nodes, exc) + continue + procs.append(proc) + launched_nodes += len(group_nodes) + if procs: + logger.info( + "Remote host samplers started on %d node(s) -> host_samples_.jsonl", + launched_nodes, + ) + return procs def _get_benchmark_profiling_env( self, diff --git a/src/srtctl/cli/mixins/telemetry_stage.py b/src/srtctl/cli/mixins/telemetry_stage.py index e6173a697..20785f065 100644 --- a/src/srtctl/cli/mixins/telemetry_stage.py +++ b/src/srtctl/cli/mixins/telemetry_stage.py @@ -38,6 +38,83 @@ # this is allowed but warned about at launch. DCGM_PROVEN_SAFE_INTERVAL_MS = 1000 +# process-exporter (ncabatoff) reads host /proc and publishes per-group CPU +# seconds by mode, thread count, per-THREAD-NAME CPU/count, context switches, +# RSS and open fds. Groups are defined by the YAML below, written into the run's +# log dir. ``-threads=true`` is what exposes +# namedprocess_namegroup_thread_{count,cpu_seconds_total}{threadname}: a runaway +# thread pool shows up as a step in thread_count and a CPU cluster on one +# threadname, which no application-level metric can express. +# ``-children=false``: a process is counted only by its own matcher, never +# folded into its parent's group (engine ranks stay separate from the launcher). +# +# Default launch is HOST-NATIVE (binary from `make setup`, config read at its +# host path); the container template applies only when a recipe gives a +# container_image (config then reached through the /logs mount). +PROCESS_EXPORTER_CONFIG_NAME = "process-exporter.yml" +PROCESS_EXPORTER_FLAGS = "-web.listen-address=:{port} -threads=true -children=false -recheck=false" +PROCESS_EXPORTER_COMMAND_TEMPLATE = ( + f"/bin/process-exporter -config.path /logs/{PROCESS_EXPORTER_CONFIG_NAME} {PROCESS_EXPORTER_FLAGS}" +) +PROCESS_EXPORTER_HOST_COMMAND_TEMPLATE = "{binary} -config.path {config_path} " + PROCESS_EXPORTER_FLAGS + + +def process_exporter_config_yaml() -> str: + """Process groups for process-exporter, first match wins. + + ``cmdline`` regexps run against the full argv; ``comm`` is the 15-char kernel + task name. The Dynamo frontend (``python3 -m dynamo.frontend``) gets its own + group because it is the process every frontend CPU pathology lives in; the + TRT-LLM/SGLang/vLLM worker handlers and their engine ranks share the + ``dynamo.`` module name and are grouped per backend; the MPI + launcher, the benchmark client and the infra daemons are named so their CPU + is attributable rather than silently dropped. Unmatched processes are not + exported (no catch-all): the per-thread breakdown of every process on a + 352-CPU node would be high-cardinality noise. + """ + return """# Generated by srtctl (telemetry_stage.process_exporter_config_yaml). First match wins. +process_names: + - name: frontend + cmdline: + - 'dynamo\\.frontend' + - name: trtllm_llmapi_launch + cmdline: + - '^trtllm-llmapi-launch' + - name: dynamo_trtllm + cmdline: + - 'dynamo\\.trtllm' + - name: dynamo_sglang + cmdline: + - 'dynamo\\.sglang' + - name: dynamo_vllm + cmdline: + - 'dynamo\\.vllm' + - name: trtllm_serve + cmdline: + - 'trtllm-serve' + - name: aiperf + cmdline: + - 'aiperf' + - name: agentperf + cmdline: + - 'agentperf' + - name: etcd + comm: + - etcd + - name: nats + comm: + - nats-server + - name: tachometer + comm: + - tachometer-scra + - name: node_exporter + comm: + - node_exporter + - name: dcgm_exporter + comm: + - dcgm-exporter +""" + def tachometer_dcgm_command_template(tachometer: TachometerConfig) -> str: """DCGM exporter command for the tachometer-owned launch. @@ -51,6 +128,52 @@ def tachometer_dcgm_command_template(tachometer: TachometerConfig) -> str: return f"dcgm-exporter --collect-interval={tachometer.collect_interval_ms} --address :{{port}}" +# Node-exporter collectors the tachometer launch enables. Beyond the original +# cpu/infiniband/meminfo trio, this adds the host scheduler-pressure family that +# distinguishes "the box is busy" from "real work is blocked waiting for a +# resource" -- the signal set the retired steady_probe.sh sampler carried and +# tachometer did not: +# stat -> node_procs_running / node_procs_blocked / node_context_switches_total +# vmstat -> node_vmstat_pgmajfault / node_vmstat_pgsteal_* (memory reclaim) +# pressure -> node_pressure_{cpu,memory,io}_* (PSI stall time) +# meminfo_numa -> node_memory_numa_MemFree_bytes (per-NUMA-node free memory) +# All four are cheap procfs/sysfs reads (/proc/{stat,vmstat,pressure}, +# /sys/devices/system/node/*/meminfo); unlike dense NVML sampling they carry no +# measured decode-latency cost. Scraper side: the vendored NodeExporterFilter +# passes the label-free stat/vmstat/pressure families through unchanged, keeps +# the NUMA `node` label of memory_numa_* as `numa_node=N`, and keeps `state` / +# `thread_state` for the processes collector (filters.rs; without those two +# rules the per-NUMA and per-state series collapsed into one). An explicit +# recipe ``node_exporter.command`` still wins (resolved in +# :func:`resolve_exporter_command`). +# processes -> node_processes_threads (host-wide thread total), node_processes_state +# (the cheapest "how many threads exist on this box" signal; #413) +NODE_EXPORTER_COLLECTORS = ("cpu", "infiniband", "meminfo", "stat", "vmstat", "pressure", "meminfo_numa", "processes") + +# node_exporter's vmstat collector defaults to ``^(oom_kill|pgpg|pswp|pg.*fault).*``, +# which ships pgmajfault but NOT pgsteal_* (page-reclaim). steady_probe.sh carried +# both major faults and reclaim, so widen the field filter to add pgsteal. Verified +# against node-exporter v1.8.2: without this, node_vmstat_pgsteal_* is absent. +NODE_EXPORTER_VMSTAT_FIELDS = "^(oom_kill|pgpg|pswp|pgsteal|pg.*fault).*" + + +def tachometer_node_exporter_command_template() -> str: + """node_exporter command for the tachometer-owned launch. + + Enables exactly :data:`NODE_EXPORTER_COLLECTORS` on top of + ``--collector.disable-defaults`` so the scrape surface is explicit and + stable regardless of the node_exporter image's built-in default set. The + pressure collector is a no-op on kernels built without ``CONFIG_PSI`` + (e.g. hecate's ``6.17.0-nvidia-64k``, verified 2026-09-09) -- node_exporter + simply omits the family, which downstream tolerates. + """ + collectors = " ".join(f"--collector.{name}" for name in NODE_EXPORTER_COLLECTORS) + return ( + f"/bin/node_exporter --web.listen-address=:{{port}} --collector.disable-defaults " + f"{collectors} --collector.vmstat.fields={NODE_EXPORTER_VMSTAT_FIELDS}" + ) + + def resolve_exporter_command(exporter_config: TelemetryExporterConfig, default_template: str) -> str: """The exact command string an exporter is launched with. @@ -127,6 +250,9 @@ def _start_exporter_container( else: chunks = [(-1, nodelist)] # sentinel: no --het-group + # Host-native exporters (``binary`` set) run straight on the node: no + # container image, no mounts -- the command already carries host paths. + host_native = bool(exporter_config.binary) managed: list[ManagedProcess] = [] for group_id, nodes in chunks: het_group = group_id if group_id >= 0 else None @@ -137,8 +263,8 @@ def _start_exporter_container( ntasks=len(nodes), nodelist=nodes, output=str(chunk_log), - container_image=exporter_config.container_image, - container_mounts=self.runtime.container_mounts, + container_image=None if host_native else exporter_config.container_image, + container_mounts=None if host_native else self.runtime.container_mounts, srun_options=self.runtime.srun_options, het_group=het_group, use_bash_wrapper=use_bash_wrapper, @@ -284,6 +410,28 @@ def finalize_power_telemetry(self, exit_code: int, *, interrupted: bool = False) return 1 return exit_code + def _resolve_host_binary(self, binary: str) -> Path | None: + """Resolve an exporter's host-native ``binary`` to an executable path, or None. + + Absolute paths are taken verbatim. Relative ones resolve against the + srtctl checkout root (``SRTCTL_SOURCE_DIR`` from the sbatch script, else + this file's repo root) -- where ``make setup`` installs host binaries. + The path must exist on the compute nodes too; the checkout lives on the + shared filesystem in every supported deployment, exactly like + ``configs/nats-server`` and ``configs/etcd``. + """ + p = Path(binary) + candidates = [p] if p.is_absolute() else [] + if not p.is_absolute(): + source_dir = os.environ.get("SRTCTL_SOURCE_DIR") + if source_dir: + candidates.append(Path(source_dir) / p) + candidates.append(Path(__file__).resolve().parents[4] / p) + for candidate in candidates: + if candidate.is_file() and os.access(candidate, os.X_OK): + return candidate + return None + def _resolve_tachometer_binary(self, binary_path: str) -> str: """Resolve the default bare binary name against the checkout's bin/. @@ -373,14 +521,46 @@ def start_tachometer(self) -> list[ManagedProcess]: name="tachometer_node_exporter", nodelist=worker_nodes, log_file=self.runtime.log_dir / "tachometer_node_exporter.out", - default_command_template=( - "/bin/node_exporter --web.listen-address=:{port} " - "--collector.disable-defaults --collector.cpu --collector.infiniband --collector.meminfo" - ), + default_command_template=tachometer_node_exporter_command_template(), use_bash_wrapper=False, critical=False, ) ) + process_exporter = tachometer.resolved_process_exporter + if process_exporter is not None: + # Every node with a backend rank OR a frontend replica: a dedicated / + # head-placed frontend node hosts no backend process, and it is + # exactly the node whose process telemetry matters most. + exporter_nodes = sorted(set(worker_nodes) | set(topology.frontend_nodes)) + pe_config = self.runtime.log_dir / PROCESS_EXPORTER_CONFIG_NAME + pe_config.write_text(process_exporter_config_yaml()) + template: str | None = PROCESS_EXPORTER_COMMAND_TEMPLATE + if process_exporter.binary: + binary = self._resolve_host_binary(process_exporter.binary) + if binary is None: + logger.warning( + "process exporter: host binary %r not found under the srtctl root; run " + "`make setup ARCH=` to install configs/process-exporter. " + "Skipping the process-exporter leg (per-process CPU/thread telemetry).", + process_exporter.binary, + ) + template = None + else: + template = PROCESS_EXPORTER_HOST_COMMAND_TEMPLATE.replace("{binary}", str(binary)).replace( + "{config_path}", str(pe_config) + ) + if template is not None: + processes.extend( + self._start_exporter_container( + exporter_config=process_exporter, + name="tachometer_process_exporter", + nodelist=exporter_nodes, + log_file=self.runtime.log_dir / "tachometer_process_exporter.out", + default_command_template=template, + use_bash_wrapper=False, + critical=False, + ) + ) cmd = [ self._resolve_tachometer_binary(tachometer.binary_path), diff --git a/src/srtctl/cli/mixins/worker_stage.py b/src/srtctl/cli/mixins/worker_stage.py index 5652c900c..de589622b 100644 --- a/src/srtctl/cli/mixins/worker_stage.py +++ b/src/srtctl/cli/mixins/worker_stage.py @@ -15,6 +15,7 @@ from srtctl.core.fingerprint import generate_capture_script from srtctl.core.health import wait_for_health +from srtctl.core.nsys_keepalive import keepalive_command from srtctl.core.processes import ManagedProcess, NamedProcesses from srtctl.core.schema import build_otel_env, installs_dynamo from srtctl.core.slurm import CONTAINER_REMAP_ROOT_EXPORT, get_hostname_ip, start_srun_process @@ -231,6 +232,9 @@ def __missing__(self, key: str) -> str: logger.info("Log: %s", worker_log) if profiling.enabled: logger.info("Profiling: %s mode", profiling.type) + if nsys_prefix and profiling.is_nsys_time: + # nsys --duration exits after writing the report; keep the task alive for the engine. + cmd = keepalive_command(cmd) # Build bash preamble (setup script + dynamo install + fingerprint) bash_preamble = self._build_worker_preamble() @@ -267,6 +271,8 @@ def __missing__(self, key: str) -> str: log_file=worker_log, node=process.node, critical=True, + # nsys writes a still-open capture range only after the engine exits; give it time. + terminate_timeout=profiling.teardown_grace_secs if nsys_prefix else 10.0, ) def start_endpoint_worker(self, endpoint_processes: list["Process"]) -> ManagedProcess: @@ -382,6 +388,9 @@ def start_endpoint_worker(self, endpoint_processes: list["Process"]) -> ManagedP logger.info("Log: %s", worker_log) if profiling.enabled: logger.info("Profiling: %s mode", profiling.type) + if nsys_prefix and profiling.is_nsys_time: + # nsys --duration exits after writing the report; keep the task alive for the engine. + cmd = keepalive_command(cmd) # Build bash preamble (setup script + dynamo install + fingerprint) bash_preamble = self._build_worker_preamble() @@ -427,6 +436,8 @@ def start_endpoint_worker(self, endpoint_processes: list["Process"]) -> ManagedP log_file=worker_log, node=leader.node, critical=True, + # nsys writes a still-open capture range only after the engine exits; give it time. + terminate_timeout=profiling.teardown_grace_secs if nsys_prefix else 10.0, ) def _wait_for_worker_ready(self, leader: "Process") -> None: diff --git a/src/srtctl/cli/submit.py b/src/srtctl/cli/submit.py index e54409209..7058eb32c 100755 --- a/src/srtctl/cli/submit.py +++ b/src/srtctl/cli/submit.py @@ -445,6 +445,10 @@ def show_config_details(config: SrtConfig) -> None: if tachometer.resolved_node_exporter is not None: node = tachometer.resolved_node_exporter details.add_row("observability", "node_exporter", f"{node.container_image} :{node.port}") + if tachometer.resolved_process_exporter is not None: + proc = tachometer.resolved_process_exporter + launch = f"host binary {proc.binary}" if proc.binary else proc.container_image + details.add_row("observability", "process_exporter", f"{launch} :{proc.port}") if config.telemetry.enabled: exporter = config.telemetry.dcgm_exporter @@ -508,6 +512,15 @@ def validate_setup(srtctl_source: Path) -> None: console.print(" make setup ARCH=x86_64 [dim]# for x86_64 compute nodes[/]\n") raise SystemExit(1) + # Optional: the default process exporter is host-native and skipped at launch + # (with a warning in the sweep log) when its binary is absent. Surface that at + # submit time so the gap is not discovered after the run. + if not (configs_dir / "process-exporter").exists(): + console.print( + "[yellow]WARNING:[/] configs/process-exporter not found; Tachometer will run without per-process/" + "per-thread CPU telemetry. Re-run [bold]make setup ARCH=[/] to install it." + ) + def generate_minimal_sbatch_script( config: SrtConfig, diff --git a/src/srtctl/core/nsys_keepalive.py b/src/srtctl/core/nsys_keepalive.py new file mode 100644 index 000000000..95b38d68a --- /dev/null +++ b/src/srtctl/core/nsys_keepalive.py @@ -0,0 +1,45 @@ +"""Keep an srun task alive after a time-windowed nsys session exits. + +``nsys profile --delay D --duration T --kill none `` stops collecting after ``T`` seconds, writes +its report and then **exits**, leaving ```` running as an orphan. Verified on hecate job 595056: +the frontend report landed at 02:09:45, the frontend srun step completed (exit 0) at 02:09:48 and the +orphaned frontend was killed with the step; the decode worker step went the same way seconds later. +Under Slurm the exiting nsys *is* the task, so the step ends and slurmstepd kills everything left in it. + +The wrapper below runs nsys in the background, records the PID of its child (the profiled app), waits +for nsys, and then keeps the shell (= the task) alive while that child still exists. Iteration-based +captures (``-c cudaProfilerApi --capture-range-end=stop``) do not need this: nsys stays attached until +the app exits. +""" + +from __future__ import annotations + +import shlex + +# Poll interval while waiting for the orphaned application (seconds). +_APP_POLL_SECS = 5 +# How long to wait for nsys to fork the application before giving up on tracking it (seconds). +_CHILD_LOOKUP_SECS = 600 + + +def keepalive_command(command: list[str]) -> list[str]: + """Wrap an nsys-prefixed ``command`` so the task outlives the nsys session. + + Returns ``["bash", "-c", script]``. The script exits with nsys's own exit code once the + application has gone away, so a failed nsys launch still surfaces as a failed process. If + ``pgrep`` is unavailable or nsys never forks a child, the script degrades to today's + behaviour (exit when nsys exits). + """ + launch = shlex.join(command) + script = ( + f"{launch} & NSYS=$!; APP=''; " + f"for _ in $(seq 1 {_CHILD_LOOKUP_SECS}); do " + 'APP=$(pgrep -P "$NSYS" 2>/dev/null | head -n1); ' + '[ -n "$APP" ] && break; kill -0 "$NSYS" 2>/dev/null || break; sleep 1; done; ' + 'wait "$NSYS"; rc=$?; ' + 'if [ -n "$APP" ]; then ' + 'echo "[srtctl] nsys (pid $NSYS) exited with $rc; keeping task alive while pid $APP runs" >&2; ' + f'while kill -0 "$APP" 2>/dev/null; do sleep {_APP_POLL_SECS}; done; fi; ' + 'exit "$rc"' + ) + return ["bash", "-c", script] diff --git a/src/srtctl/core/processes.py b/src/srtctl/core/processes.py index 7c1672d42..30c86d609 100644 --- a/src/srtctl/core/processes.py +++ b/src/srtctl/core/processes.py @@ -61,6 +61,9 @@ class ManagedProcess: log_file: Path to the process log file node: Node hostname where the process runs critical: If True, failure triggers full cleanup + terminate_timeout: Seconds ``cleanup()`` waits after SIGTERM before SIGKILL. Raise + it for processes that must flush state on exit (an nsys-wrapped worker whose + capture range is still open writes its report only after the engine exits). """ name: str @@ -68,6 +71,7 @@ class ManagedProcess: log_file: Path | None = None node: str | None = None critical: bool = True + terminate_timeout: float = 10.0 @property def is_running(self) -> bool: @@ -79,11 +83,16 @@ def exit_code(self) -> int | None: """Get exit code if process has exited, None otherwise.""" return self.popen.poll() - def terminate(self, timeout: float = 10.0) -> None: - """Terminate the process gracefully, then kill if needed.""" + def terminate(self, timeout: float | None = None) -> None: + """Terminate the process gracefully, then kill if needed. + + ``timeout`` defaults to ``self.terminate_timeout``. + """ if not self.is_running: return + if timeout is None: + timeout = self.terminate_timeout outcome = terminate_and_reap(self.popen, terminate_timeout=timeout, kill_timeout=5) if not outcome.reaped: logger.error("Process %s was not reaped after SIGKILL", self.name) @@ -148,6 +157,7 @@ def add_processes(self, processes: NamedProcesses) -> None: log_file=proc.log_file, node=proc.node, critical=proc.critical, + terminate_timeout=proc.terminate_timeout, ) self.add_process(proc) @@ -172,16 +182,41 @@ def check_failures(self) -> bool: return len(self._failed_processes) > 0 def cleanup(self) -> None: - """Terminate all registered processes.""" + """Terminate all registered processes. + + Two phases so the grace periods overlap instead of adding up: SIGTERM every running + process first, then wait for each one up to its own ``terminate_timeout`` before + escalating to SIGKILL. A profiled worker with a 3-minute grace therefore costs at + most 3 minutes in total, not 3 minutes per worker. + """ with self._lock: logger.info("Cleaning up %d processes...", len(self._processes)) + running: list[ManagedProcess] = [] for name, proc in self._processes.items(): - if proc.is_running: - logger.debug("Terminating process: %s", name) + if not proc.is_running: + continue + logger.debug("Terminating process: %s", name) + try: + proc.popen.terminate() + running.append(proc) + except Exception as e: # noqa: BLE001 + logger.warning("Failed to terminate %s: %s", name, e) + for proc in running: + try: + proc.popen.wait(timeout=proc.terminate_timeout) + except subprocess.TimeoutExpired: + logger.warning( + "Process %s did not exit within %.0fs of SIGTERM, killing...", + proc.name, + proc.terminate_timeout, + ) + proc.popen.kill() try: - proc.terminate() - except Exception as e: # noqa: BLE001 - logger.warning("Failed to terminate %s: %s", name, e) + proc.popen.wait(timeout=5) + except subprocess.TimeoutExpired: + logger.error("Process %s was not reaped after SIGKILL", proc.name) + except Exception as e: # noqa: BLE001 + logger.warning("Failed to reap %s: %s", proc.name, e) def print_failure_details(self, tail_lines: int = 50) -> None: """Print detailed failure information including log tails. diff --git a/src/srtctl/core/schema.py b/src/srtctl/core/schema.py index 90e84859d..d62d8128f 100755 --- a/src/srtctl/core/schema.py +++ b/src/srtctl/core/schema.py @@ -18,6 +18,7 @@ import math import os import shlex +import warnings from collections.abc import Iterator, Mapping from dataclasses import field from enum import Enum @@ -843,6 +844,26 @@ def vllm_nsys_max_iterations(self) -> int: Schema: ClassVar[builtins.type[Schema]] = Schema +@dataclass(frozen=True) +class ProfilingFrontendConfig: + """nsys on the Dynamo frontend process (Rust/tokio HTTP frontend + router). + + The frontend has no CUDA work and never calls cudaProfilerStart, so it gets a time window + instead of the workers' iteration window: ``--delay`` seconds after the frontend process + starts, capture for ``--duration`` seconds, then nsys writes the report and leaves the + frontend running (``--kill none``). What is worth tracing there is NVTX (Dynamo's Rust + ranges via DYN_ENABLE_RUST_NVTX + the NVTX injection library) and optionally OS runtime + calls; CUDA tracing is meaningless on this process and is not enabled. + """ + + delay_secs: int = 900 # from frontend start; the workers need ~10-15 min to load before traffic flows + duration_secs: int = 120 + trace: str = "nvtx" # nsys -t for the frontend: "nvtx" or "nvtx,osrt" + extra_nsys_args: list[str] | None = None + + Schema: ClassVar[builtins.type[Schema]] = Schema + + @dataclass(frozen=True) class ProfilingConfig: """Profiling configuration. @@ -869,6 +890,49 @@ class ProfilingConfig: duration_secs: int | None = None # nsys --duration: seconds to capture after delay benchmark_duration_secs: int = 300 # total traffic generation duration (must cover delay + duration) + # Seconds srtctl waits after SIGTERM before SIGKILL for every nsys-wrapped process (workers and + # the profiled frontend). A capture range that is still open when the run ends (stop_step never + # reached, or nsys-time whose duration outlasts the benchmark) is written only after the engine + # exits; the default 10 s process grace loses those reports. Ignored unless type is nsys/nsys-time. + teardown_grace_secs: int = 180 + + # ---- TRT-LLM nsys capture recipe ------------------------------------------------- + # Defaults follow the Dynamo Benchmark Playbook §9.5.1.1 "Dynamo + TRTLLM", the set + # that produced every usable multi-node capture on nsys 2026.3.x / VR200 disagg: + # -t cuda-sw NOT `cuda`: on nsys 2026.x plain `cuda` selects the HES hardware + # trace, which SIGSEGVs the KV transceiver's device-to-device + # cudaMemcpyAsync; cuda-sw forces the software tracer. + # --cuda-graph-trace=graph decode runs CUDA graphs; with `node` the graph launches are + # invisible (~0.3% GPU busy reads as idle), with `graph` each + # iteration is one GRAPH_TRACE entry and utilisation is real. + # --sample=none --cpuctxsw=none --sample=process-tree wedged workers at cudaProfilerStop + # across 8 runs; the price is no scheduler/CPU-time data (every + # duration is wall-clock). --cpuctxsw=process-tree alone (context + # switches, no IP sampling) is the safer experiment; both need + # kernel.perf_event_paranoid <= 2 on the compute node. + # --flush-on-cudaprofilerstop=false --cuda-flush-interval=0 matches the known-good + # captures; nsys 2026.4 flips the flush default, so set it explicitly. + # --gpu-metrics-devices=none a second collection path with its own failure modes. + # `-c cudaProfilerApi --capture-range-end=stop` (iteration window from prefill/decode + # start_step/stop_step via TLLM_PROFILE_START_STOP) is kept: finalising the report when the + # range closes is the only path that has produced reports on this stack. + nsys_trace: str = "cuda-sw,nvtx,python-gil" + nsys_cuda_graph_trace: str = "graph" # "graph" | "node" + nsys_sample: str = "none" # "none" | "process-tree" | "system-wide" (nsys --sample) + nsys_cpuctxsw: str = "none" # "none" | "process-tree" | "system-wide" (nsys --cpuctxsw) + nsys_python_sampling: bool = False + nsys_python_sampling_frequency: int = 1000 + nsys_gpu_metrics_devices: str = "none" + # NVTX injection library inside the worker container, exported as NVTX_INJECTION64_PATH so + # Dynamo's Rust NVTX ranges (DYN_ENABLE_RUST_NVTX=1; needs a wheel built with the `nvtx` + # cargo feature) reach nsys. The TRT-LLM containers ship it next to nsys, e.g. + # /usr/local/cuda-0.gpgpu/NsightSystems-cli-2026.3.0/target-linux-sbsa-armv8/libToolsInjection64.so + nvtx_injection_path: str | None = None + # TLLM_PROFILE_LOG_RANKS: which engine ranks log the profiling start/stop iterations. + log_ranks: str = "all" + # Also attach nsys to the Dynamo frontend process(es), with a time window (see ProfilingFrontendConfig). + frontend: ProfilingFrontendConfig | None = None + @property def enabled(self) -> bool: """Check if profiling is enabled.""" @@ -936,6 +1000,15 @@ def get_env_vars(self, mode: str, profile_dir: str) -> dict[str, str]: env["TLLM_PROFILE_START_STOP"] = f"{phase_config.start_step}-{phase_config.stop_step}" env["TLLM_LLMAPI_ENABLE_NVTX"] = "1" + if self.is_nsys: + # Playbook environment for the profiled workers. TLLM_* keys are ignored by + # other engines; DYN_ENABLE_RUST_NVTX is a no-op on wheels built without the + # nvtx cargo feature; NVTX_INJECTION64_PATH only when the recipe names the lib. + env["TLLM_PROFILE_LOG_RANKS"] = self.log_ranks + env["DYN_ENABLE_RUST_NVTX"] = "1" + if self.nvtx_injection_path: + env["NVTX_INJECTION64_PATH"] = self.nvtx_injection_path + return env @property @@ -949,53 +1022,92 @@ def nsys_binary(self) -> str: """ return os.environ.get("SRTCTL_NSYS_BIN", "nsys") + def _nsys_common_flags(self) -> list[str]: + """Flags shared by the iteration- and time-based TRT-LLM captures (see the field docs).""" + return [ + "-t", + self.nsys_trace, + f"--cuda-graph-trace={self.nsys_cuda_graph_trace}", + f"--sample={self.nsys_sample}", + f"--cpuctxsw={self.nsys_cpuctxsw}", + f"--python-sampling={'true' if self.nsys_python_sampling else 'false'}", + f"--python-sampling-frequency={self.nsys_python_sampling_frequency}", + f"--gpu-metrics-devices={self.nsys_gpu_metrics_devices}", + "--flush-on-cudaprofilerstop=false", + "--cuda-flush-interval=0", + ] + + @property + def profiles_frontend(self) -> bool: + """Whether the Dynamo frontend process is wrapped in nsys too.""" + return self.is_nsys and self.frontend is not None + + def get_frontend_nsys_prefix(self, output_file: str) -> list[str]: + """``nsys profile ...`` prefix for the Dynamo frontend command (time window, NVTX-centric). + + Empty unless ``profiling.frontend`` is set with an nsys type. ``--wait all`` keeps nsys + alive as long as the frontend runs; the report is written when the window closes. + """ + if not self.profiles_frontend: + return [] + assert self.frontend is not None + fe = self.frontend + return [ + self.nsys_binary, + "profile", + "--force-overwrite=true", + "-t", + fe.trace, + f"--sample={self.nsys_sample}", + f"--cpuctxsw={self.nsys_cpuctxsw}", + "--python-sampling=false", + f"--gpu-metrics-devices={self.nsys_gpu_metrics_devices}", + "--delay", + str(fe.delay_secs), + "--duration", + str(fe.duration_secs), + *(fe.extra_nsys_args or []), + "--kill", + "none", + "--wait", + "all", + "-o", + output_file, + ] + + def get_frontend_env_vars(self) -> dict[str, str]: + """Environment that makes the frontend's Rust NVTX ranges reach nsys.""" + if not self.profiles_frontend: + return {} + env = {"DYN_ENABLE_RUST_NVTX": "1"} + if self.nvtx_injection_path: + env["NVTX_INJECTION64_PATH"] = self.nvtx_injection_path + return env + def _get_nsys_prefix_trtllm(self, output_file: str) -> list[str]: - """Get nsys command prefix for TRTLLM workers. + """nsys command prefix for TRT-LLM workers (Dynamo Benchmark Playbook §9.5.1.1 recipe). - Supports both iteration-based (cudaProfilerApi trigger via TLLM_PROFILE_START_STOP) - and time-based (--delay/--duration) capture modes. + Iteration-based (default): TLLM_PROFILE_START_STOP makes the PyExecutor call + cudaProfilerStart/Stop on every rank, `-c cudaProfilerApi --capture-range-end=stop` + records exactly that window and finalises the report as soon as it closes. Time-based + (nsys-time): --delay/--duration instead. One nsys per srun task (= engine rank); + the caller's ``output_file`` carries ``%q{SLURM_PROCID}`` so ranks never collide. """ + cmd = [self.nsys_binary, "profile", "--force-overwrite=true"] + self._nsys_common_flags() if self.is_nsys_time: - cmd = [ - self.nsys_binary, - "profile", - "-t", - "cuda,nvtx,ucx", - "--sample=none", - "--cuda-graph-trace=node", - ] if self.delay_secs is not None: cmd += ["--delay", str(self.delay_secs)] if self.duration_secs is not None: cmd += ["--duration", str(self.duration_secs)] else: - # Iteration-based: TLLM_PROFILE_START_STOP env var triggers cudaProfilerStart/Stop - cmd = [ - self.nsys_binary, - "profile", - "-t", - "cuda,nvtx,ucx", - "--sample=none", - "--cuda-graph-trace=node", - "-c", - "cudaProfilerApi", - "--capture-range-end", - "stop", - ] + cmd += ["-c", "cudaProfilerApi", "--capture-range-end=stop"] if self.extra_nsys_args: cmd.extend(self.extra_nsys_args) - cmd += [ - "--kill", - "none", - "--wait", - "all", - "--force-overwrite", - "true", - "-o", - output_file, - ] + # --kill none / --wait all: nsys must neither kill the engine when its own + # session ends nor exit before the MPI ranks trtllm-llmapi-launch spawns. + cmd += ["--kill", "none", "--wait", "all", "-o", output_file] return cmd def get_nsys_prefix( @@ -1075,11 +1187,25 @@ def get_nsys_prefix( @dataclass(frozen=True) class TelemetryExporterConfig: - """Configuration for a metrics exporter deployed on worker nodes.""" + """Configuration for a metrics exporter deployed on worker nodes. + + Two launch modes. With ``binary`` unset the exporter runs as a pyxis + container from ``container_image``. With ``binary`` set it runs + **host-native** -- the executable is started by ``srun`` directly on the + node with no container; ``container_image`` is ignored (set it to ``""``). + Relative ``binary`` paths resolve against the srtctl checkout root, which is + where ``make setup`` installs the host binaries (``configs/nats-server``, + ``configs/etcd``, ``configs/process-exporter``). Host-native exists because + some enroot deployments cannot start shell-less ``FROM scratch`` images + (observed on hecate: ``enroot-switchroot: failed to change directory: /root``, + then ``/bin/sh: No such file or directory`` with the home mounted), and a + static Go exporter needs no container at all. + """ container_image: str port: int command: str | None = None + binary: str | None = None Schema: ClassVar[type[Schema]] = Schema @@ -1104,6 +1230,25 @@ class TelemetryExporterConfig: container_image="quay.io#prometheus/node-exporter:v1.8.2", port=9101, ) +# Per-process and per-thread host telemetry from /proc: CPU seconds by mode, +# thread count and thread CPU by thread name, context switches, RSS, open fds -- +# for the frontend, the worker handlers, the engine ranks and the client, grouped +# by command line (see telemetry_stage.process_exporter_config_yaml). This is the +# signal the Prometheus surface cannot carry: Dynamo publishes no process_* or +# thread metrics, and node_exporter only sees the machine. +# +# Launched HOST-NATIVE from the static Go binary `make setup` installs at +# configs/process-exporter (ncabatoff/process-exporter release tarball for the +# compute arch), like nats-server and etcd. The upstream image is FROM scratch +# (no shell, no /root) and pyxis/enroot on hecate refuses to start it; the binary +# needs neither a container nor privileges and reads the host /proc directly. A +# recipe may still point `process_exporter.container_image` at an image that has +# a shell and leave `binary` unset to get the container launch. +DEFAULT_PROCESS_EXPORTER = TelemetryExporterConfig( + container_image="", + port=9256, + binary="configs/process-exporter", +) @dataclass(frozen=True) @@ -1117,10 +1262,11 @@ class TachometerConfig: observability expansion is what turns their content on); the frontend and the exporters are always worth capturing. - DCGM and node exporters default ON via the ``resolved_*`` properties - (sweep path only): an explicit ``dcgm_exporter``/``node_exporter`` block - always wins, ``default_exporters: false`` disables the built-ins, and the - raw fields stay ``None`` unless the recipe set them — which is what the + DCGM, node and process exporters default ON via the ``resolved_*`` + properties (sweep path only): an explicit ``dcgm_exporter`` / + ``node_exporter`` / ``process_exporter`` block always wins, + ``default_exporters: false`` disables the built-ins, and the raw fields + stay ``None`` unless the recipe set them — which is what the power-telemetry sharing validation and the --bash gate key on. """ @@ -1137,6 +1283,7 @@ class TachometerConfig: default_exporters: bool = True dcgm_exporter: TelemetryExporterConfig | None = None node_exporter: TelemetryExporterConfig | None = None + process_exporter: TelemetryExporterConfig | None = None Schema: ClassVar[type[Schema]] = Schema @@ -1154,6 +1301,13 @@ def resolved_node_exporter(self) -> TelemetryExporterConfig | None: return self.node_exporter return DEFAULT_NODE_EXPORTER if self.default_exporters else None + @property + def resolved_process_exporter(self) -> TelemetryExporterConfig | None: + """User-configured process exporter, else the built-in default.""" + if self.process_exporter is not None: + return self.process_exporter + return DEFAULT_PROCESS_EXPORTER if self.default_exporters else None + @dataclass(frozen=True) class ObservabilityConfig: @@ -1223,6 +1377,11 @@ class ObservabilityConfig: enabled: bool = False enable_otel: bool = False otel_endpoint: str | None = None + # Run the /proc host sampler on EVERY allocated node (one persistent srun per + # node group, files host_samples_.jsonl), not just the orchestrator + # node. Closes the gap where worker nodes — and a dedicated frontend node — + # had no per-process host-CPU/scheduler telemetry at all. Follows ``enabled``. + host_sampler_all_nodes: bool = True tachometer: TachometerConfig = field(default_factory=TachometerConfig) @@ -1917,7 +2076,7 @@ def _validate_dynamo_sidecar(self) -> None: return if self.frontend.type != "dynamo": raise ValidationError("dynamo.sidecar: true requires frontend.type: dynamo") - if not isinstance(self.backend, (SGLangProtocol, VLLMProtocol, TRTLLMProtocol)): + if not isinstance(self.backend, SGLangProtocol | VLLMProtocol | TRTLLMProtocol): raise ValidationError("dynamo.sidecar: true supports sglang, vllm, and trtllm backends only") if isinstance(self.backend, VLLMProtocol): self.backend.validate_sidecar_dp_config() @@ -2207,6 +2366,42 @@ def _validate_profiling(self): if prof.is_torch and backend_type == "trtllm": raise ValidationError("torch profiling is not supported for the trtllm backend; use nsys instead") + if prof.is_nsys: + for name, value, allowed in ( + ("nsys_cuda_graph_trace", prof.nsys_cuda_graph_trace, ("graph", "node")), + ("nsys_sample", prof.nsys_sample, ("none", "process-tree", "system-wide")), + ("nsys_cpuctxsw", prof.nsys_cpuctxsw, ("none", "process-tree", "system-wide")), + ): + if value not in allowed: + raise ValidationError(f"profiling.{name} must be one of {allowed}, got {value!r}") + if not prof.nsys_trace.strip(): + raise ValidationError( + "profiling.nsys_trace must be a non-empty nsys -t list, e.g. 'cuda-sw,nvtx,python-gil'" + ) + if prof.nsys_sample != "none": + # Not an error: the playbook measured wedged workers at cudaProfilerStop with + # --sample=process-tree; the user asked for it explicitly. + warnings.warn( + "profiling.nsys_sample != 'none': CPU IP sampling wedged TRT-LLM workers at " + "cudaProfilerStop in the playbook campaign and needs kernel.perf_event_paranoid <= 2; " + "prefer nsys_cpuctxsw: process-tree alone.", + stacklevel=2, + ) + + if prof.frontend is not None: + if not prof.is_nsys: + raise ValidationError("profiling.frontend requires profiling.type nsys or nsys-time") + if self.frontend.type != "dynamo": + raise ValidationError( + f"profiling.frontend is implemented for frontend.type: dynamo only (got {self.frontend.type!r})" + ) + if prof.frontend.delay_secs < 0 or prof.frontend.duration_secs <= 0: + raise ValidationError("profiling.frontend.delay_secs must be >= 0 and duration_secs > 0") + if not prof.frontend.trace.strip(): + raise ValidationError("profiling.frontend.trace must be a non-empty nsys -t list, e.g. 'nvtx'") + if prof.teardown_grace_secs <= 0: + raise ValidationError("profiling.teardown_grace_secs must be > 0 seconds") + # nsys-time (time-based capture via nsys --delay/--duration) is supported # for all backends. get_nsys_prefix() emits a time-based command for the # non-TRTLLM (vllm/sglang) path too, which is the only option for @@ -2362,12 +2557,14 @@ def _validate_observability(self): "observability.tachometer.storage_subdir and telemetry.storage_subdir must be different" ) - for name in ("dcgm_exporter", "node_exporter"): + for name in ("dcgm_exporter", "node_exporter", "process_exporter"): exporter = getattr(tachometer, name) if exporter is None: continue - if not exporter.container_image: - raise ValidationError(f"observability.tachometer.{name}.container_image must be non-empty") + if not exporter.container_image and not exporter.binary: + raise ValidationError( + f"observability.tachometer.{name}: set container_image (container launch) or binary (host-native)" + ) if not 1 <= exporter.port <= 65535: raise ValidationError(f"observability.tachometer.{name}.port must be in 1..65535") if not tachometer.binary_path: diff --git a/src/srtctl/core/telemetry.py b/src/srtctl/core/telemetry.py index 68b3c49af..2ab33231f 100644 --- a/src/srtctl/core/telemetry.py +++ b/src/srtctl/core/telemetry.py @@ -180,6 +180,26 @@ def generate_tachometer_config( ) ) + process_exporter = tachometer.resolved_process_exporter + if process_exporter is not None: + # Per-process / per-thread host telemetry on every node that hosts a + # backend rank OR a frontend replica. The frontend node is the one the + # other exporters can miss (a dedicated or `orchestrator_placement: + # head` frontend hosts no backend process), and it is where frontend + # CPU pathologies live. No filter: groupname/threadname/mode labels + # pass through verbatim. + for node in sorted(set(physical_nodes) | set(frontend_nodes)): + node_metadata = {"hostname": node, "job_id": runtime.job_id, "run_name": runtime.run_name} + node_metadata.update(tachometer.extra_metadata) + endpoints.append( + TelemetryEndpoint( + name=f"process_exporter_{node}", + url=f"http://{node}:{process_exporter.port}/metrics", + collect_interval_ms=tachometer.collect_interval_ms, + node_metadata=node_metadata, + ) + ) + return _dump_toml( endpoints=endpoints, storage=str(runtime.log_dir / tachometer.storage_subdir / TACHOMETER_STORAGE_PARENT / TACHOMETER_STORAGE_LEAF), diff --git a/src/srtctl/frontends/dynamo.py b/src/srtctl/frontends/dynamo.py index 6e579a796..7967bd61e 100644 --- a/src/srtctl/frontends/dynamo.py +++ b/src/srtctl/frontends/dynamo.py @@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any from srtctl.core.health import WorkerHealthResult, check_dynamo_health +from srtctl.core.nsys_keepalive import keepalive_command from srtctl.core.schema import build_otel_env from srtctl.core.slurm import CONTAINER_REMAP_ROOT_EXPORT, start_srun_process from srtctl.ports import ETCD_CLIENT_PORT, NATS_PORT @@ -90,6 +91,25 @@ def start_frontends( frontend_log = runtime.log_dir / f"{node}_frontend_{idx}.out" cmd = ["python3", "-m", "dynamo.frontend", f"--http-port={topology.frontend_port}"] cmd.extend(self.get_frontend_args_list(config.frontend.args)) + # profiling.frontend: run the frontend under nsys with a time window (see ProfilingFrontendConfig). + # The report goes next to the workers' reports: /logs/profiles/frontend/_frontend_.nsys-rep + profiling = getattr(config, "profiling", None) # tests pass reduced configs without a profiling block + nsys_prefix = ( + profiling.get_frontend_nsys_prefix(f"/logs/profiles/frontend/{node}_frontend_{idx}") + if profiling + else [] + ) + if nsys_prefix: + (runtime.log_dir / "profiles" / "frontend").mkdir(parents=True, exist_ok=True) + # Time-windowed nsys exits once its report is written; keep the srun task (and the + # frontend) alive until the run ends. See srtctl.core.nsys_keepalive. + cmd = keepalive_command([*nsys_prefix, *cmd]) + logger.info( + "Profiling: nsys on frontend %d (delay %ss, duration %ss)", + idx, + profiling.frontend.delay_secs, + profiling.frontend.duration_secs, + ) env_to_set = { "ETCD_ENDPOINTS": f"http://{runtime.nodes.infra}:{ETCD_CLIENT_PORT}", @@ -107,6 +127,10 @@ def start_frontends( # dynamo.wheel, before frontend-specific overrides. env_to_set.update(runtime.environment) + # NVTX plumbing for the profiled frontend (no-op unless profiling.frontend is set) + if profiling: + env_to_set.update(profiling.get_frontend_env_vars()) + # Add frontend env from config if config.frontend.env: env_to_set.update(config.frontend.env) @@ -138,6 +162,7 @@ def start_frontends( log_file=frontend_log, node=node, critical=True, + terminate_timeout=profiling.teardown_grace_secs if nsys_prefix else 10.0, ) ) diff --git a/src/tachometer/tachometer-scraper/src/filters.rs b/src/tachometer/tachometer-scraper/src/filters.rs index 4f0cbf6ec..9a35abb02 100644 --- a/src/tachometer/tachometer-scraper/src/filters.rs +++ b/src/tachometer/tachometer-scraper/src/filters.rs @@ -128,6 +128,13 @@ impl MetricFilter for NodeExporterFilter { base_metric.to_string() } } + // Per-NUMA memory (--collector.meminfo_numa): node_memory_numa_*{node="N"}. + // The `node` label IS the breakdown; dropping it (as the generic + // memory_ arm below does) collapses every NUMA node into one series. + m if m.starts_with("memory_numa_") => match sample.labels.get("node") { + Some(numa_node) => format!("{}{{numa_node={}}}", base_metric, numa_node), + None => base_metric.to_string(), + }, // Memory metrics - usually don't need labels m if m.starts_with("memory_") => base_metric.to_string(), // Disk metrics - simplify device names @@ -149,10 +156,22 @@ impl MetricFilter for NodeExporterFilter { if sample.labels.is_empty() { base_metric.to_string() } else { - // Keep only the most important labels (limit to 2-3) + // Keep only the most important labels (limit to 2-3). + // `state` / `thread_state` carry the per-state process and + // thread counts of --collector.processes + // (node_processes_state{state="R"}, node_processes_threads_state); + // without them the states collapse into one series. let mut important_labels = Vec::new(); - let priority_labels = - ["job", "instance", "device", "mountpoint", "fstype", "mode"]; + let priority_labels = [ + "job", + "instance", + "device", + "mountpoint", + "fstype", + "mode", + "state", + "thread_state", + ]; for key in priority_labels.iter() { if let Some(value) = sample.labels.get(*key) { @@ -398,3 +417,69 @@ pub fn get_filter( _ => Box::new(NoOpFilter), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::parse::MetricType; + + fn sample(name: &str, labels: &[(&str, &str)]) -> ParsedSample { + ParsedSample { + metric_name: name.to_string(), + labels: labels + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + value: 1.0, + metric_type: MetricType::Gauge, + } + } + + #[test] + fn numa_memory_keeps_the_node_breakdown() { + // --collector.meminfo_numa: one series per NUMA node. The generic memory_ + // arm used to drop every label, folding all nodes into one series. + let f = NodeExporterFilter::new(HashMap::new()); + let (m0, _) = f.filter(&sample("node_memory_numa_MemFree_bytes", &[("node", "0")])); + let (m1, _) = f.filter(&sample("node_memory_numa_MemFree_bytes", &[("node", "1")])); + assert_eq!(m0, "memory_numa_MemFree_bytes{numa_node=0}"); + assert_eq!(m1, "memory_numa_MemFree_bytes{numa_node=1}"); + // Host-wide meminfo stays label-free. + let (m, _) = f.filter(&sample("node_memory_MemFree_bytes", &[])); + assert_eq!(m, "memory_MemFree_bytes"); + } + + #[test] + fn process_state_counts_keep_their_state_label() { + // --collector.processes: node_processes_state{state="R"|"S"|"D"|...} + let f = NodeExporterFilter::new(HashMap::new()); + let (r, _) = f.filter(&sample("node_processes_state", &[("state", "R")])); + let (d, _) = f.filter(&sample("node_processes_state", &[("state", "D")])); + assert_eq!(r, "processes_state{state=R}"); + assert_eq!(d, "processes_state{state=D}"); + let (t, _) = f.filter(&sample( + "node_processes_threads_state", + &[("thread_state", "R")], + )); + assert_eq!(t, "processes_threads_state{thread_state=R}"); + } + + #[test] + fn label_free_scheduler_pressure_families_pass_through() { + // --collector.stat / vmstat / pressure carry no labels: bare names. + let f = NodeExporterFilter::new(HashMap::new()); + for (name, want) in [ + ("node_procs_running", "procs_running"), + ("node_procs_blocked", "procs_blocked"), + ("node_vmstat_pgmajfault", "vmstat_pgmajfault"), + ( + "node_pressure_cpu_waiting_seconds_total", + "pressure_cpu_waiting_seconds_total", + ), + ("node_processes_threads", "processes_threads"), + ] { + let (m, _) = f.filter(&sample(name, &[])); + assert_eq!(m, want); + } + } +} diff --git a/tests/test_component_dashboard.py b/tests/test_component_dashboard.py index 6c47c36a7..0dab3dc0c 100644 --- a/tests/test_component_dashboard.py +++ b/tests/test_component_dashboard.py @@ -2045,6 +2045,34 @@ def test_budget_drops_wrappers_not_workers(self, monkeypatch): assert 12 in got, "the one real worker must survive a budget full of wrappers" assert len(got) == 5 + def test_pressure_parses_psi_totals(self, monkeypatch): + from srtctl.analysis import host_sampler as hs + + psi = { + "/proc/pressure/cpu": "some avg10=0.00 avg60=0.10 avg300=0.05 total=123456\n", + "/proc/pressure/memory": ( + "some avg10=0.00 avg60=0.00 avg300=0.00 total=7890\n" + "full avg10=0.00 avg60=0.00 avg300=0.00 total=4200\n" + ), + "/proc/pressure/io": "some avg10=0.00 avg60=0.00 avg300=0.00 total=99\nfull avg10=0.00 total=55\n", + } + monkeypatch.setattr(hs, "_read", lambda path: psi.get(path)) + + got = hs._pressure() + assert got == { + "cpu_some_total_us": 123456, + "memory_some_total_us": 7890, + "memory_full_total_us": 4200, + "io_some_total_us": 99, + "io_full_total_us": 55, + } + + def test_pressure_absent_psi_yields_empty(self, monkeypatch): + from srtctl.analysis import host_sampler as hs + + monkeypatch.setattr(hs, "_read", lambda path: None) # kernel without CONFIG_PSI + assert hs._pressure() == {} + class TestAiperfJsonMetrics: """AIPerf's own server-metrics export as a second metrics source. diff --git a/tests/test_dry_run.py b/tests/test_dry_run.py index 5d4b4cf68..88f3dbb72 100644 --- a/tests/test_dry_run.py +++ b/tests/test_dry_run.py @@ -275,8 +275,15 @@ def test_tachometer_details_shown_without_explicit_block(self, capsys): # Built-in exporters are part of the default and must be visible. assert "dcgm_exporter" in output assert "node_exporter" in output + assert "process_exporter" in output assert ":9401" in output assert ":9101" in output + assert ":9256" in output + # The process exporter is host-native by default; dry-run must say so + # (and name the binary make setup installs) rather than print an image. + # Rich wraps the cell, so the two halves are asserted separately. + assert "host binary" in output + assert "configs/process-exporter" in output def test_dcgm_power_telemetry_details_shown(self, capsys): config = _make_config( @@ -584,8 +591,9 @@ class TestInfmaxWorkspaceMount: --container-mounts against the failed arm's showed this single missing entry. """ - AGENTIC = {"benchmark": {"type": "custom", - "command": "bash /infmax-workspace/benchmarks/multi_node/agentic_srt.sh"}} + AGENTIC = { + "benchmark": {"type": "custom", "command": "bash /infmax-workspace/benchmarks/multi_node/agentic_srt.sh"} + } def test_mount_is_shown_when_the_variable_is_set(self, capsys): config = _make_config(self.AGENTIC) diff --git a/tests/test_host_sampler_all_nodes.py b/tests/test_host_sampler_all_nodes.py new file mode 100644 index 000000000..df6f78538 --- /dev/null +++ b/tests/test_host_sampler_all_nodes.py @@ -0,0 +1,193 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the all-node host sampler: scheduler-contention fields, the +standalone per-node CLI mode, remote-launch planning, and the multi-file ingest.""" + +import json +import os +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +from srtctl.analysis import host_sampler +from srtctl.analysis.host_sampler import _proc_sample, plan_remote_sampler_nodes + + +class TestProcSampleSchedulerFields: + def test_own_process_has_scheduler_fields(self): + d = _proc_sample(os.getpid()) + assert d is not None + # /proc//schedstat: cumulative on-CPU ns and run-queue-wait ns. + assert isinstance(d.get("cpu_ns"), int) + assert isinstance(d.get("run_delay_ns"), int) + # se.nr_migrations from /proc//sched. + assert isinstance(d.get("nr_migrations"), int) + # sched_getaffinity width: the direct pinning-state observable. + assert isinstance(d.get("affinity_ncpus"), int) and d["affinity_ncpus"] >= 1 + + def test_procs_running_blocked(self): + running, blocked = host_sampler._procs_running_blocked() + assert isinstance(running, int) and running >= 1 + assert isinstance(blocked, int) and blocked >= 0 + + +class TestPlanRemoteSamplerNodes: + def test_excludes_local_and_dedups(self): + assert plan_remote_sampler_nodes(["n1", "n2", "n1", "n3"], "n2") == ["n1", "n3"] + + def test_fqdn_vs_short_hostname(self): + assert plan_remote_sampler_nodes(["n1.cluster.local", "n2"], "n1") == ["n2"] + assert plan_remote_sampler_nodes(["n1", "n2"], "n1.cluster.local") == ["n2"] + + def test_all_local(self): + assert plan_remote_sampler_nodes(["n1", "n1"], "n1") == [] + + +class TestStandaloneMode: + def test_writes_per_node_file_and_exits(self, tmp_path): + """python3 host_sampler.py --log-dir D --max-samples 2 writes host_samples_.jsonl.""" + script = Path(host_sampler.__file__).resolve() + r = subprocess.run( + [sys.executable, str(script), "--log-dir", str(tmp_path), "--interval", "1", "--max-samples", "2"], + capture_output=True, + text=True, + timeout=30, + ) + assert r.returncode == 0, r.stderr + out = tmp_path / f"host_samples_{os.uname().nodename}.jsonl" + assert out.exists() + rows = [json.loads(line) for line in out.read_text().splitlines() if line.strip()] + assert len(rows) >= 2 + assert rows[0]["host"] == os.uname().nodename + assert "t_mono" in rows[0] and "procs_running" in rows[0] + + +class TestRemoteLaunch: + def _mixin(self, worker_nodes, collect_interval_ms=None): + from srtctl.cli.mixins.benchmark_stage import BenchmarkStageMixin + + m = BenchmarkStageMixin.__new__(BenchmarkStageMixin) + runtime = MagicMock() + runtime.nodes.head = worker_nodes[0] + runtime.nodes.bench = worker_nodes[0] + runtime.nodes.infra = worker_nodes[0] + runtime.nodes.worker = tuple(worker_nodes) + runtime.nodes.het_group_for.return_value = None + runtime.log_dir = Path("/tmp/logs") + runtime.srun_options = {} + m.runtime = runtime + config = MagicMock() + if collect_interval_ms is None: + config.observability.tachometer = None + else: + config.observability.tachometer.collect_interval_ms = collect_interval_ms + m.config = config + return m + + def test_launches_on_all_nodes_except_local(self): + local = os.uname().nodename + mixin = self._mixin([local, "nodeB", "nodeC"]) + with patch("srtctl.cli.mixins.benchmark_stage.start_srun_process") as srun: + srun.return_value = MagicMock() + procs = mixin._start_remote_host_samplers() + assert len(procs) == 1 + kwargs = srun.call_args.kwargs + assert kwargs["nodelist"] == ["nodeB", "nodeC"] + assert kwargs["nodes"] == 2 and kwargs["ntasks"] == 2 + assert kwargs["command"][0] == "python3" + assert kwargs["command"][1].endswith("host_sampler.py") + assert "--log-dir" in kwargs["command"] + # No tachometer block -> the 1 s default of the scrape knob. + assert kwargs["command"][kwargs["command"].index("--interval") + 1] == "1" + + def test_remote_interval_follows_the_scrape_knob(self): + """Remote samplers use observability.tachometer.collect_interval_ms like the + in-process one (try_start_host_sampler), so head-node and remote rows share a + cadence; sub-second values clamp to 1 s exactly as HostSampler does.""" + local = os.uname().nodename + for interval_ms, expected in ((2000, "2"), (5000, "5"), (250, "1"), (1500, "1.5")): + mixin = self._mixin([local, "nodeB"], collect_interval_ms=interval_ms) + with patch("srtctl.cli.mixins.benchmark_stage.start_srun_process") as srun: + srun.return_value = MagicMock() + mixin._start_remote_host_samplers() + cmd = srun.call_args.kwargs["command"] + assert cmd[cmd.index("--interval") + 1] == expected, (interval_ms, cmd) + + def test_no_launch_when_only_local(self): + mixin = self._mixin([os.uname().nodename]) + with patch("srtctl.cli.mixins.benchmark_stage.start_srun_process") as srun: + procs = mixin._start_remote_host_samplers() + assert procs == [] and srun.call_count == 0 + + +def _rows(): + def proc(pid, cpu_j, invol, delay_ns, migr, aff): + return {"pid": pid, "name": "dynamo", "cpu_jiffies": cpu_j, "ctx_invol": invol, + "run_delay_ns": delay_ns, "nr_migrations": migr, "affinity_ncpus": aff, + "rss_kb": 100, "threads": 4, "open_fds": 10} + + return [ + {"t": 100.0, "host": "nodeB", "cpu_busy_jiffies": 1000, "cpu_total_jiffies": 10000, + "procs_running": 5, "procs_blocked": 0, "mem": {"MemTotal": 100, "MemAvailable": 50}, + "established_conns": 3, "fd_limit": 1024, "procs": [proc(7, 100, 10, 1_000_000_000, 50, 144)]}, + {"t": 102.0, "host": "nodeB", "cpu_busy_jiffies": 1100, "cpu_total_jiffies": 10200, + "procs_running": 8, "procs_blocked": 1, "mem": {"MemTotal": 100, "MemAvailable": 40}, + "established_conns": 4, "fd_limit": 1024, "procs": [proc(7, 140, 30, 1_200_000_000, 60, 144)]}, + ] + + +class TestIngestSchedulerSeries: + def test_rates_from_rows(self): + from src.ingest.ingest import _host_series_from_rows + + s = _host_series_from_rows(_rows()) + assert s is not None and s["host"] == "nodeB" + p = s["procs"]["dynamo:7"] + # 200ms of run-queue delta over 2s -> 100 ms/s + assert p["run_delay_ms_per_s"] == [[102.0, 100.0]] + # 10 migrations over 2s -> 5/s + assert p["migrations_rate"] == [[102.0, 5.0]] + assert p["affinity_ncpus"] == [[102.0, 144]] + assert s["procs_runnable"] == [[102.0, 8]] + assert s["procs_blocked"] == [[102.0, 1]] + + def test_rates_prefer_monotonic_clock(self): + """A forward NTP step in wall-clock t must not deflate the rates.""" + from src.ingest.ingest import _host_series_from_rows + + rows = _rows() + rows[0]["t_mono"], rows[1]["t_mono"] = 500.0, 502.0 + rows[1]["t"] = 112.0 # wall clock stepped +10s mid-window + s = _host_series_from_rows(rows) + # dt = 2s from t_mono, not 12s from t: 200ms delta -> 100 ms/s + assert s["procs"]["dynamo:7"]["run_delay_ms_per_s"] == [[112.0, 100.0]] + + def test_multi_file_hosts_map(self, tmp_path): + from src.ingest.ingest import run_host_samples + + run_dir, bundle = tmp_path / "run", tmp_path / "bundle" + run_dir.mkdir() + bundle.mkdir() + local = _rows() + for r in local: + r["host"] = "orchestrator" + (run_dir / "host_samples.jsonl").write_text("\n".join(json.dumps(r) for r in local)) + (run_dir / "host_samples_nodeB.jsonl").write_text("\n".join(json.dumps(r) for r in _rows())) + out = run_host_samples(run_dir, bundle) + assert out["host"] == "orchestrator" + assert "nodeB" in out["hosts"] + assert out["hosts"]["nodeB"]["procs"]["dynamo:7"]["run_delay_ms_per_s"] == [[102.0, 100.0]] + assert json.loads((bundle / "host_series.json").read_text())["hosts"]["nodeB"]["host"] == "nodeB" + + def test_per_node_files_only(self, tmp_path): + """Remote-node files alone (no orchestrator file) still produce a bundle.""" + from src.ingest.ingest import run_host_samples + + run_dir, bundle = tmp_path / "run", tmp_path / "bundle" + run_dir.mkdir() + bundle.mkdir() + (run_dir / "host_samples_nodeB.jsonl").write_text("\n".join(json.dumps(r) for r in _rows())) + out = run_host_samples(run_dir, bundle) + assert out["hosts"]["nodeB"]["samples"] == 2 diff --git a/tests/test_nsys_keepalive.py b/tests/test_nsys_keepalive.py new file mode 100644 index 000000000..9ef151f87 --- /dev/null +++ b/tests/test_nsys_keepalive.py @@ -0,0 +1,55 @@ +"""keepalive_command: the srun task must outlive a time-windowed nsys session.""" + +import os +import shlex +import stat +import subprocess +import time + +from srtctl.core.nsys_keepalive import keepalive_command + + +def test_wrapper_shape(): + cmd = keepalive_command(["nsys", "profile", "--delay", "5", "-o", "/logs/x y", "python3", "-m", "dynamo.frontend"]) + assert cmd[:2] == ["bash", "-c"] + script = cmd[2] + assert script.startswith( + shlex.join(["nsys", "profile", "--delay", "5", "-o", "/logs/x y", "python3", "-m", "dynamo.frontend"]) + " &" + ) + assert "pgrep -P" in script and 'wait "$NSYS"' in script and 'kill -0 "$APP"' in script + assert script.endswith('exit "$rc"') + + +def _fake_nsys(tmp_path, *, exit_code: int, child_secs: float): + """A stand-in for nsys: fork a child that lives child_secs, exit after 0.5 s with exit_code.""" + path = tmp_path / "nsys" + path.write_text("#!/usr/bin/env bash\n" f"sleep {child_secs} &\n" "sleep 0.5\n" f"exit {exit_code}\n") + path.chmod(path.stat().st_mode | stat.S_IXUSR) + return str(path) + + +def test_task_outlives_nsys_and_keeps_its_exit_code(tmp_path): + nsys = _fake_nsys(tmp_path, exit_code=0, child_secs=3) + t0 = time.monotonic() + proc = subprocess.run(keepalive_command([nsys]), capture_output=True, text=True, timeout=60) + elapsed = time.monotonic() - t0 + assert proc.returncode == 0 + # nsys itself exits after 0.5 s; the wrapper must stay until the child (3 s) is gone + assert elapsed >= 2.5, (elapsed, proc.stderr) + assert "keeping task alive while pid" in proc.stderr + + +def test_nsys_failure_is_propagated(tmp_path): + nsys = _fake_nsys(tmp_path, exit_code=3, child_secs=1) + proc = subprocess.run(keepalive_command([nsys]), capture_output=True, text=True, timeout=60) + assert proc.returncode == 3 + + +def test_no_child_degrades_to_plain_wait(tmp_path): + path = tmp_path / "nsys" + path.write_text("#!/usr/bin/env bash\nexit 0\n") + path.chmod(path.stat().st_mode | stat.S_IXUSR) + t0 = time.monotonic() + proc = subprocess.run(keepalive_command([str(path)]), capture_output=True, text=True, timeout=60) + assert proc.returncode == 0 and time.monotonic() - t0 < 5 + assert os.environ is not None # keep os imported for readers extending the fake diff --git a/tests/test_process_registry.py b/tests/test_process_registry.py index e37098510..5f884ab08 100644 --- a/tests/test_process_registry.py +++ b/tests/test_process_registry.py @@ -42,8 +42,9 @@ def test_managed_process_exit_code(self): log_file=Path("/tmp/test.log"), ) - # exit_code comes from popen.returncode - assert mock_popen.returncode == 1 + # exit_code comes from popen.poll() + assert mp.exit_code == 1 + assert not mp.is_running def test_terminate_does_not_raise_when_kill_wait_times_out(self): """A child that survives SIGKILL must not raise out of terminate().""" @@ -204,3 +205,61 @@ def test_cleanup(self): registry.cleanup() mock_popen.terminate.assert_called_once() + + +class TestCleanupGrace: + """cleanup() signals every process first, then waits per-process up to terminate_timeout.""" + + @staticmethod + def _running(pid: int, exits_after_terminate: bool = True) -> MagicMock: + popen = MagicMock(spec=Popen) + popen.poll.return_value = None + popen.pid = pid + if exits_after_terminate: + popen.wait.return_value = 0 + else: + popen.wait.side_effect = [TimeoutExpired("x", 1), 0] # survives SIGTERM, dies on SIGKILL + return popen + + def test_signals_all_before_waiting(self): + order: list[str] = [] + registry = ProcessRegistry(job_id="j") + for i in range(3): + popen = self._running(100 + i) + popen.terminate.side_effect = lambda i=i: order.append(f"term{i}") + popen.wait.side_effect = lambda timeout=None, i=i: order.append(f"wait{i}") or 0 + registry.add_process(ManagedProcess(name=f"p{i}", popen=popen)) + registry.cleanup() + assert order[:3] == ["term0", "term1", "term2"] + assert sorted(order[3:]) == ["wait0", "wait1", "wait2"] + + def test_waits_for_each_process_own_timeout(self): + registry = ProcessRegistry(job_id="j") + fast, slow = self._running(1), self._running(2) + registry.add_process(ManagedProcess(name="fast", popen=fast)) + registry.add_process(ManagedProcess(name="slow", popen=slow, terminate_timeout=180.0)) + registry.cleanup() + fast.wait.assert_called_once_with(timeout=10.0) + slow.wait.assert_called_once_with(timeout=180.0) + fast.kill.assert_not_called() + slow.kill.assert_not_called() + + def test_kills_after_grace_expires(self): + registry = ProcessRegistry(job_id="j") + stubborn = self._running(3, exits_after_terminate=False) + registry.add_process(ManagedProcess(name="stubborn", popen=stubborn, terminate_timeout=0.01)) + registry.cleanup() + stubborn.terminate.assert_called_once() + stubborn.kill.assert_called_once() + assert stubborn.wait.call_count == 2 + + def test_add_processes_keeps_terminate_timeout_when_renaming(self): + registry = ProcessRegistry(job_id="j") + popen = self._running(4) + registry.add_processes({"renamed": ManagedProcess(name="orig", popen=popen, terminate_timeout=42.0)}) + assert registry._processes["renamed"].terminate_timeout == 42.0 + + def test_terminate_defaults_to_own_timeout(self): + popen = self._running(5) + ManagedProcess(name="p", popen=popen, terminate_timeout=33.0).terminate() + popen.wait.assert_called_once_with(timeout=33.0) diff --git a/tests/test_profiling_playbook.py b/tests/test_profiling_playbook.py new file mode 100644 index 000000000..eda0c7223 --- /dev/null +++ b/tests/test_profiling_playbook.py @@ -0,0 +1,253 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""TRT-LLM nsys capture recipe from the Dynamo Benchmark Playbook §9.5.1.1. + +The flags are not stylistic: `-t cuda` SIGSEGVs the KV transceiver on nsys 2026.x, +`--cuda-graph-trace=node` hides CUDA-graph decode work, `--sample=process-tree` wedged +workers at cudaProfilerStop. These tests pin the recipe and the knobs that relax it. +""" + +import warnings + +import pytest +from marshmallow import ValidationError + +from srtctl.core.schema import ModelConfig, ProfilingConfig, ProfilingPhaseConfig, ResourceConfig, SrtConfig + +INJECTION = "/usr/local/cuda-0.gpgpu/NsightSystems-cli-2026.3.0/target-linux-sbsa-armv8/libToolsInjection64.so" + + +def _disagg_config(**profiling_kwargs) -> SrtConfig: + profiling_kwargs.setdefault("type", "nsys") + profiling_kwargs.setdefault("prefill", ProfilingPhaseConfig(start_step=1200, stop_step=1300)) + profiling_kwargs.setdefault("decode", ProfilingPhaseConfig(start_step=6000, stop_step=6600)) + return SrtConfig( + name="nsys-playbook", + model=ModelConfig(path="/model", container="/container", precision="fp8"), + resources=ResourceConfig( + gpu_type="gb200", prefill_nodes=1, decode_nodes=1, prefill_workers=1, decode_workers=1 + ), + profiling=ProfilingConfig(**profiling_kwargs), + ) + + +class TestPlaybookPrefix: + def test_trtllm_iteration_prefix_is_the_playbook_recipe(self): + prefix = ProfilingConfig(type="nsys").get_nsys_prefix( + "/logs/profiles/decode/n_decode_w0_profile_rank%q{SLURM_PROCID}", backend_type="trtllm" + ) + assert prefix[:3] == ["nsys", "profile", "--force-overwrite=true"] + assert prefix[prefix.index("-t") + 1] == "cuda-sw,nvtx,python-gil" + for flag in ( + "--cuda-graph-trace=graph", + "--sample=none", + "--cpuctxsw=none", + "--python-sampling=false", + "--python-sampling-frequency=1000", + "--gpu-metrics-devices=none", + "--flush-on-cudaprofilerstop=false", + "--cuda-flush-interval=0", + "--capture-range-end=stop", + ): + assert flag in prefix, flag + assert prefix[prefix.index("-c") + 1] == "cudaProfilerApi" + # The pre-playbook flags must be gone: HES trace and node-level graph trace. + assert "cuda,nvtx,ucx" not in prefix + assert "--cuda-graph-trace=node" not in prefix + # One nsys per srun task; the per-rank output template is passed through verbatim. + assert prefix[-2:] == ["-o", "/logs/profiles/decode/n_decode_w0_profile_rank%q{SLURM_PROCID}"] + assert prefix[prefix.index("--kill") + 1] == "none" + assert prefix[prefix.index("--wait") + 1] == "all" + + def test_trtllm_time_prefix_shares_the_recipe(self): + prefix = ProfilingConfig(type="nsys-time", delay_secs=600, duration_secs=60).get_nsys_prefix( + "/out/rank%q{SLURM_PROCID}", backend_type="trtllm" + ) + assert "cuda-sw,nvtx,python-gil" in prefix + assert "--cuda-graph-trace=graph" in prefix + assert prefix[prefix.index("--delay") + 1] == "600" + assert prefix[prefix.index("--duration") + 1] == "60" + assert "cudaProfilerApi" not in prefix + + def test_knobs_relax_the_recipe(self): + prefix = ProfilingConfig( + type="nsys", + nsys_trace="cuda-sw,nvtx,python-gil,ucx", + nsys_cpuctxsw="process-tree", + nsys_python_sampling=True, + nsys_python_sampling_frequency=500, + extra_nsys_args=["--stats=true"], + ).get_nsys_prefix("/out/rank%q{SLURM_PROCID}", backend_type="trtllm") + assert prefix[prefix.index("-t") + 1] == "cuda-sw,nvtx,python-gil,ucx" + assert "--cpuctxsw=process-tree" in prefix + assert "--sample=none" in prefix # IP sampling stays off unless asked + assert "--python-sampling=true" in prefix + assert "--python-sampling-frequency=500" in prefix + assert prefix.index("--stats=true") < prefix.index("-o") + + def test_non_trtllm_backends_keep_their_own_prefix(self): + prefix = ProfilingConfig(type="nsys").get_nsys_prefix("/out/x", backend_type="sglang", frontend_type="dynamo") + assert "--trace-fork-before-exec=true" in prefix + assert "cuda-sw" not in prefix + + +class TestPlaybookEnv: + def test_worker_env_carries_window_ranks_and_nvtx(self): + env = ProfilingConfig( + type="nsys", + prefill=ProfilingPhaseConfig(start_step=1200, stop_step=1300), + decode=ProfilingPhaseConfig(start_step=6000, stop_step=6600), + nvtx_injection_path=INJECTION, + ).get_env_vars("decode", "/logs/profiles") + assert env["TLLM_PROFILE_START_STOP"] == "6000-6600" + assert env["TLLM_LLMAPI_ENABLE_NVTX"] == "1" + assert env["TLLM_PROFILE_LOG_RANKS"] == "all" + assert env["DYN_ENABLE_RUST_NVTX"] == "1" + assert env["NVTX_INJECTION64_PATH"] == INJECTION + + def test_injection_path_only_when_configured(self): + env = ProfilingConfig(type="nsys", decode=ProfilingPhaseConfig(0, 50)).get_env_vars("decode", "/p") + assert "NVTX_INJECTION64_PATH" not in env + assert env["DYN_ENABLE_RUST_NVTX"] == "1" + + def test_log_ranks_override(self): + env = ProfilingConfig(type="nsys", log_ranks="0", decode=ProfilingPhaseConfig(0, 50)).get_env_vars( + "decode", "/p" + ) + assert env["TLLM_PROFILE_LOG_RANKS"] == "0" + + def test_disabled_profiling_sets_nothing(self): + assert ProfilingConfig().get_env_vars("decode", "/p") == {} + + +class TestPlaybookValidation: + def test_valid_playbook_config(self): + config = _disagg_config(nvtx_injection_path=INJECTION) + assert config.profiling.enabled and config.profiling.nsys_trace == "cuda-sw,nvtx,python-gil" + + @pytest.mark.parametrize( + ("field", "value"), + [("nsys_cuda_graph_trace", "kernel"), ("nsys_sample", "yes"), ("nsys_cpuctxsw", "all")], + ) + def test_enum_fields_are_checked(self, field, value): + with pytest.raises(ValidationError, match=f"profiling.{field}"): + _disagg_config(**{field: value}) + + def test_empty_trace_rejected(self): + with pytest.raises(ValidationError, match="nsys_trace"): + _disagg_config(nsys_trace=" ") + + def test_ip_sampling_warns_but_is_allowed(self): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + config = _disagg_config(nsys_sample="process-tree") + assert config.profiling.nsys_sample == "process-tree" + assert any("perf_event_paranoid" in str(w.message) for w in caught) + + +class TestFrontendProfiling: + """profiling.frontend: nsys on the Dynamo frontend with a time window.""" + + def test_prefix_env_and_defaults(self): + from srtctl.core.schema import ProfilingFrontendConfig + + p = ProfilingConfig( + type="nsys", + frontend=ProfilingFrontendConfig(delay_secs=1500, duration_secs=120), + nvtx_injection_path="/opt/nsight/libToolsInjection64.so", + ) + assert p.profiles_frontend + prefix = p.get_frontend_nsys_prefix("/logs/profiles/frontend/n1_frontend_0") + assert prefix[:3] == ["nsys", "profile", "--force-overwrite=true"] + assert prefix[prefix.index("-t") + 1] == "nvtx" # NVTX-centric: no CUDA tracing on a CUDA-less process + assert "--delay" in prefix and prefix[prefix.index("--delay") + 1] == "1500" + assert "--duration" in prefix and prefix[prefix.index("--duration") + 1] == "120" + assert prefix[-6:] == ["--kill", "none", "--wait", "all", "-o", "/logs/profiles/frontend/n1_frontend_0"] + assert "-c" not in prefix and "--capture-range-end=stop" not in prefix + assert p.get_frontend_env_vars() == { + "DYN_ENABLE_RUST_NVTX": "1", + "NVTX_INJECTION64_PATH": "/opt/nsight/libToolsInjection64.so", + } + # Without the block nothing changes for the frontend. + assert ProfilingConfig(type="nsys").get_frontend_nsys_prefix("/o") == [] + assert ProfilingConfig(type="nsys").get_frontend_env_vars() == {} + + def test_validation(self): + from srtctl.core.schema import ProfilingFrontendConfig + + cfg = _disagg_config(frontend=ProfilingFrontendConfig()) + assert cfg.profiling.profiles_frontend + with pytest.raises(ValidationError, match="requires profiling.type nsys"): + _disagg_config(type="torch", frontend=ProfilingFrontendConfig()) + with pytest.raises(ValidationError, match="duration_secs"): + _disagg_config(frontend=ProfilingFrontendConfig(duration_secs=0)) + with pytest.raises(ValidationError, match="trace"): + _disagg_config(frontend=ProfilingFrontendConfig(trace=" ")) + + def test_teardown_grace_default_and_validation(self): + assert _disagg_config().profiling.teardown_grace_secs == 180 + assert _disagg_config(teardown_grace_secs=600).profiling.teardown_grace_secs == 600 + with pytest.raises(ValidationError): + _disagg_config(teardown_grace_secs=0) + + def test_dynamo_frontend_is_wrapped(self, tmp_path): + from types import SimpleNamespace + from unittest.mock import MagicMock, patch + + from srtctl.core.schema import ProfilingFrontendConfig + from srtctl.frontends.base import get_frontend + + cfg = _disagg_config( + frontend=ProfilingFrontendConfig(delay_secs=10, duration_secs=5), + nvtx_injection_path="/opt/nsight/libToolsInjection64.so", + ) + topology = SimpleNamespace(frontend_nodes=["n1"], frontend_port=8000) + runtime = SimpleNamespace( + log_dir=tmp_path, + nodes=SimpleNamespace(infra="n0", het_group_for=lambda node: None), + container_image="/img.sqsh", + container_mounts={}, + environment={}, + ) + with patch("srtctl.frontends.dynamo.start_srun_process", return_value=MagicMock()) as srun: + procs = get_frontend("dynamo").start_frontends( + topology=topology, runtime=runtime, config=cfg, backend=None, backend_processes=[] + ) + assert len(procs) == 1 + kwargs = srun.call_args.kwargs + cmd = kwargs["command"] + # wrapped by keepalive_command: bash -c ' & ...wait for the orphaned frontend...' + assert cmd[:2] == ["bash", "-c"] + script = cmd[2] + assert script.startswith("nsys profile ") + assert "-o /logs/profiles/frontend/n1_frontend_0 python3 -m dynamo.frontend --http-port=8000" in script + assert 'kill -0 "$APP"' in script + assert kwargs["env_to_set"]["DYN_ENABLE_RUST_NVTX"] == "1" + assert kwargs["env_to_set"]["NVTX_INJECTION64_PATH"] == "/opt/nsight/libToolsInjection64.so" + assert (tmp_path / "profiles" / "frontend").is_dir() + # an open frontend capture is written only after the frontend exits: cleanup must wait for it + assert procs[0].terminate_timeout == 180.0 + + def test_dynamo_frontend_untouched_without_block(self, tmp_path): + from types import SimpleNamespace + from unittest.mock import MagicMock, patch + + from srtctl.frontends.base import get_frontend + + cfg = _disagg_config() + topology = SimpleNamespace(frontend_nodes=["n1"], frontend_port=8000) + runtime = SimpleNamespace( + log_dir=tmp_path, + nodes=SimpleNamespace(infra="n0", het_group_for=lambda node: None), + container_image="/img.sqsh", + container_mounts={}, + environment={}, + ) + with patch("srtctl.frontends.dynamo.start_srun_process", return_value=MagicMock()) as srun: + get_frontend("dynamo").start_frontends( + topology=topology, runtime=runtime, config=cfg, backend=None, backend_processes=[] + ) + cmd = srun.call_args.kwargs["command"] + assert cmd[:3] == ["python3", "-m", "dynamo.frontend"] + assert "DYN_ENABLE_RUST_NVTX" not in srun.call_args.kwargs["env_to_set"] diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index aa3e7f7da..e118e6b61 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -92,6 +92,68 @@ def test_exporters_resolve_to_built_in_defaults(self): assert "dcgm-exporter" in tachometer.resolved_dcgm_exporter.container_image assert tachometer.resolved_node_exporter.port == 9101 assert "node-exporter" in tachometer.resolved_node_exporter.container_image + assert tachometer.process_exporter is None + assert tachometer.resolved_process_exporter.port == 9256 + # Host-native by default: the upstream image is FROM scratch and some + # enroot deployments cannot start it (no /root, no /bin/sh). + assert tachometer.resolved_process_exporter.binary == "configs/process-exporter" + assert tachometer.resolved_process_exporter.container_image == "" + + def test_process_exporter_requires_binary_or_container_image(self): + with pytest.raises(ValidationError, match="observability.tachometer.process_exporter"): + _make_config( + tachometer=TachometerConfig( + enabled=True, + process_exporter=TelemetryExporterConfig(container_image="", port=9256), + ) + ) + + def test_process_exporter_container_override_is_accepted(self): + custom = TelemetryExporterConfig(container_image="/containers/process-exporter.sqsh", port=9300) + config = _make_config(tachometer=TachometerConfig(enabled=True, process_exporter=custom)) + assert config.observability.tachometer.resolved_process_exporter is custom + assert config.observability.tachometer.resolved_process_exporter.binary is None + + def test_process_exporter_groups_name_every_srtctl_process_class(self): + """The process-exporter config must isolate the frontend in its own group + (its CPU is the signal the Prometheus surface cannot carry) and expose + thread-name breakdowns; first match wins, so the launcher precedes the + module it wraps.""" + from srtctl.cli.mixins.telemetry_stage import ( + PROCESS_EXPORTER_COMMAND_TEMPLATE, + process_exporter_config_yaml, + resolve_exporter_command, + ) + + text = process_exporter_config_yaml() + names = [line.split("name:", 1)[1].strip() for line in text.splitlines() if "name:" in line] + assert names[0] == "frontend" + assert names.index("trtllm_llmapi_launch") < names.index("dynamo_trtllm") + for expected in ("dynamo_trtllm", "dynamo_sglang", "dynamo_vllm", "aiperf", "etcd", "nats"): + assert expected in names + assert "dynamo\\.frontend" in text + + cmd = resolve_exporter_command(TachometerConfig().resolved_process_exporter, PROCESS_EXPORTER_COMMAND_TEMPLATE) + assert "-config.path /logs/process-exporter.yml" in cmd + assert "-threads=true" in cmd + assert "-children=false" in cmd + assert ":9256" in cmd + + def test_process_exporter_host_command_uses_host_paths(self): + """Host-native launch: no /logs mount exists, so the binary and the + group file are both addressed by their host paths.""" + from srtctl.cli.mixins.telemetry_stage import ( + PROCESS_EXPORTER_HOST_COMMAND_TEMPLATE, + resolve_exporter_command, + ) + + template = PROCESS_EXPORTER_HOST_COMMAND_TEMPLATE.replace("{binary}", "/srt/configs/process-exporter").replace( + "{config_path}", "/lustre/out/logs/process-exporter.yml" + ) + cmd = resolve_exporter_command(TachometerConfig().resolved_process_exporter, template) + assert cmd.startswith("/srt/configs/process-exporter -config.path /lustre/out/logs/process-exporter.yml ") + assert "-web.listen-address=:9256" in cmd + assert "-threads=true" in cmd def test_dcgm_sampling_follows_the_scrape_knob(self): """One knob rules both cadences: the tachometer-owned DCGM exporter @@ -106,9 +168,7 @@ def test_dcgm_sampling_follows_the_scrape_knob(self): ) default = TachometerConfig() - cmd = resolve_exporter_command( - default.resolved_dcgm_exporter, tachometer_dcgm_command_template(default) - ) + cmd = resolve_exporter_command(default.resolved_dcgm_exporter, tachometer_dcgm_command_template(default)) assert "--collect-interval=1000" in cmd assert ":9401" in cmd @@ -121,12 +181,50 @@ def test_dcgm_sampling_follows_the_scrape_knob(self): container_image="dcgm:latest", port=9401, command="dcgm-exporter --custom --address :{port}" ) ) - assert resolve_exporter_command( - custom.resolved_dcgm_exporter, tachometer_dcgm_command_template(custom) - ) == "dcgm-exporter --custom --address :9401" + assert ( + resolve_exporter_command(custom.resolved_dcgm_exporter, tachometer_dcgm_command_template(custom)) + == "dcgm-exporter --custom --address :9401" + ) assert "--collect-interval=100 " in DCGM_EXPORTER_COMMAND_TEMPLATE + def test_node_exporter_enables_host_scheduler_collectors(self): + """The tachometer node_exporter must collect the host scheduler-pressure + family (PSI, procs/context-switch counters, memory-reclaim, per-NUMA free) + on top of cpu/infiniband/meminfo -- the steady_probe.sh signal set that was + otherwise uncollected. An explicit recipe command still wins.""" + from srtctl.cli.mixins.telemetry_stage import ( + NODE_EXPORTER_COLLECTORS, + resolve_exporter_command, + tachometer_node_exporter_command_template, + ) + + template = tachometer_node_exporter_command_template() + assert "--collector.disable-defaults" in template + for collector in ("cpu", "infiniband", "meminfo", "stat", "vmstat", "pressure", "meminfo_numa"): + assert f"--collector.{collector}" in template, collector + assert set(NODE_EXPORTER_COLLECTORS) >= {"stat", "vmstat", "pressure", "meminfo_numa"} + # vmstat's default field set omits pgsteal (page-reclaim); the override + # must add it while keeping pgmajfault. Verified against node-exporter v1.8.2. + assert "--collector.vmstat.fields=" in template + assert "pgsteal" in template + + default = TachometerConfig() + cmd = resolve_exporter_command(default.resolved_node_exporter, template) + assert "--collector.pressure" in cmd + assert f":{default.resolved_node_exporter.port}" in cmd + + # An explicit recipe command must still win over the derived template. + custom = TachometerConfig( + node_exporter=TelemetryExporterConfig( + container_image="node:latest", port=9101, command="/bin/node_exporter --custom :{port}" + ) + ) + assert ( + resolve_exporter_command(custom.resolved_node_exporter, template) + == "/bin/node_exporter --custom :9101" + ) + def test_host_sampler_follows_the_scrape_knob(self, tmp_path): """The host sampler's cadence derives from the same single knob.""" import threading @@ -146,6 +244,7 @@ def test_default_exporters_false_disables_built_ins(self): tachometer = TachometerConfig(default_exporters=False) assert tachometer.resolved_dcgm_exporter is None assert tachometer.resolved_node_exporter is None + assert tachometer.resolved_process_exporter is None def test_explicit_exporter_block_wins_over_default(self): custom = TelemetryExporterConfig(container_image="/containers/dcgm.sqsh", port=9500) @@ -626,6 +725,49 @@ def test_generate_config_without_exporters_targets_servers_only(self, _mock_get_ assert 'name = "frontend0"' in config_text assert "dcgm_" not in config_text assert "node_exporter_" not in config_text + assert "process_exporter_" not in config_text + + @patch("srtctl.core.telemetry.get_hostname_ip", return_value="10.0.0.1") + def test_process_exporter_targets_frontend_node_even_without_backend(self, _mock_get_hostname_ip): + """A head-placed or dedicated frontend node hosts no backend process, so the + per-node exporters would skip it -- yet it is where frontend CPU lives. + The process exporter must target the union of backend and frontend nodes, + unfiltered (groupname/threadname labels pass through).""" + tachometer = TachometerConfig(enabled=True) + runtime = MagicMock(job_id="12345", run_name="test_12345", network_interface="eth0") + runtime.log_dir = Path("/runs/12345/logs") + processes = [ + Process( + node="node-a", + gpu_indices=frozenset({0}), + sys_port=8081, + http_port=30000, + endpoint_mode="agg", + endpoint_index=0, + node_rank=0, + ) + ] + topology = FrontendTopology( + nginx_node=None, + frontend_nodes=["fe-node"], + frontend_port=8000, + public_port=8000, + ) + + config_text = generate_tachometer_config( + processes=processes, + frontend_topology=topology, + runtime=runtime, + tachometer=tachometer, + ) + + assert 'name = "process_exporter_node-a"' in config_text + assert 'name = "process_exporter_fe-node"' in config_text + assert 'url = "http://fe-node:9256/metrics"' in config_text + # DCGM/node exporters keep their backend-node scope. + assert 'name = "node_exporter_fe-node"' not in config_text + block = config_text.split('name = "process_exporter_fe-node"', 1)[1].split("[[endpoints]]", 1)[0] + assert "filter =" not in block @patch("srtctl.core.telemetry.get_hostname_ip") def test_vllm_frontend_targets_only_agg_leader_metrics(self, mock_get_hostname_ip): @@ -741,13 +883,30 @@ def _compute_frontend_topology(self): # Pin the default-name PATH fallback so the assertion below does not # depend on whether the developer's checkout has bin/tachometer-scraper. harness._resolve_tachometer_binary = lambda binary_path: binary_path + # Likewise pin the host-native process-exporter binary (installed by make setup). + harness._resolve_host_binary = lambda binary: Path("/srt/configs/process-exporter") procs = harness.start_tachometer() - assert len(procs) == 3 + assert len(procs) == 4 # dcgm + node + process exporters + scraper assert (tmp_path / "tachometer_config.toml").exists() assert (tmp_path / "tachometer" / "local").exists() - assert mock_srun.call_count == 3 + assert mock_srun.call_count == 4 + # The process exporter runs host-native: no container, host paths for + # the binary and its group file. + pe_call = mock_srun.call_args_list[2] + assert pe_call.kwargs["container_image"] is None + assert pe_call.kwargs["container_mounts"] is None + assert pe_call.kwargs["command"][:3] == [ + "/srt/configs/process-exporter", + "-config.path", + str(tmp_path / "process-exporter.yml"), + ] + assert "-web.listen-address=:9256" in pe_call.kwargs["command"] + # The container exporters keep their image + mounts. + for call in mock_srun.call_args_list[:2]: + assert call.kwargs["container_image"] in ("dcgm:latest", "node:latest") + assert call.kwargs["container_mounts"] == {Path(tmp_path): Path("/logs")} scraper_call = mock_srun.call_args_list[-1] assert scraper_call.kwargs["command"] == [ "tachometer-scraper", @@ -832,6 +991,7 @@ def _compute_frontend_topology(self): mock_srun.return_value = _running_exporter() harness = Harness() harness._resolve_tachometer_binary = lambda binary_path: binary_path + harness._resolve_host_binary = lambda binary: Path("/srt/configs/process-exporter") procs = harness.start_tachometer() @@ -839,9 +999,155 @@ def _compute_frontend_topology(self): assert [proc.name for proc in procs] == [ "tachometer_dcgm_exporter", "tachometer_node_exporter", + "tachometer_process_exporter", "tachometer", ] assert (tmp_path / "tachometer_config.toml").exists() + # The process-exporter group file lands in the log dir. + assert "name: frontend" in (tmp_path / "process-exporter.yml").read_text() + + @patch("srtctl.cli.mixins.telemetry_stage.start_srun_process") + def test_process_exporter_skipped_when_host_binary_missing(self, mock_srun, tmp_path, caplog): + """A checkout whose `make setup` predates the process exporter must still + run: the leg is skipped with a warning, the other exporters and the + scraper start as before.""" + import dataclasses + import logging + + class Harness(TelemetryStageMixin): + def __init__(self): + base = _make_config() + self.config = dataclasses.replace(base, observability=ObservabilityConfig(enabled=True)) + self.runtime = MagicMock() + self.runtime.log_dir = tmp_path + self.runtime.job_id = "12345" + self.runtime.run_name = "test_12345" + self.runtime.network_interface = "eth0" + self.runtime.nodes.head = "node-a" + self.runtime.nodes.het = False + self.runtime.srun_options = {} + self.runtime.container_mounts = {Path(tmp_path): Path("/logs")} + self._backend_processes = [ + Process( + node="node-a", + gpu_indices=frozenset({0}), + sys_port=8081, + http_port=30000, + endpoint_mode="agg", + endpoint_index=0, + node_rank=0, + ) + ] + + @property + def backend_processes(self): + return self._backend_processes + + def _compute_frontend_topology(self): + return FrontendTopology( + nginx_node=None, + frontend_nodes=["node-a"], + frontend_port=8000, + public_port=8000, + ) + + mock_srun.return_value = _running_exporter() + harness = Harness() + harness._resolve_tachometer_binary = lambda binary_path: binary_path + harness._resolve_host_binary = lambda binary: None + + with caplog.at_level(logging.WARNING, logger="srtctl.cli.mixins.telemetry_stage"): + procs = harness.start_tachometer() + + assert [proc.name for proc in procs] == [ + "tachometer_dcgm_exporter", + "tachometer_node_exporter", + "tachometer", + ] + assert any("configs/process-exporter" in record.getMessage() for record in caplog.records) + + @patch("srtctl.cli.mixins.telemetry_stage.start_srun_process") + def test_process_exporter_container_image_launches_in_container(self, mock_srun, tmp_path): + """An explicit container_image (and no binary) keeps the container path: + image + mounts on srun, group file addressed through /logs.""" + import dataclasses + + class Harness(TelemetryStageMixin): + def __init__(self): + base = _make_config() + self.config = dataclasses.replace( + base, + observability=ObservabilityConfig( + enabled=True, + tachometer=TachometerConfig( + enabled=True, + process_exporter=TelemetryExporterConfig(container_image="pe-with-shell:latest", port=9256), + ), + ), + ) + self.runtime = MagicMock() + self.runtime.log_dir = tmp_path + self.runtime.job_id = "12345" + self.runtime.run_name = "test_12345" + self.runtime.network_interface = "eth0" + self.runtime.nodes.head = "node-a" + self.runtime.nodes.het = False + self.runtime.srun_options = {} + self.runtime.container_mounts = {Path(tmp_path): Path("/logs")} + self._backend_processes = [ + Process( + node="node-a", + gpu_indices=frozenset({0}), + sys_port=8081, + http_port=30000, + endpoint_mode="agg", + endpoint_index=0, + node_rank=0, + ) + ] + + @property + def backend_processes(self): + return self._backend_processes + + def _compute_frontend_topology(self): + return FrontendTopology( + nginx_node=None, + frontend_nodes=["node-a"], + frontend_port=8000, + public_port=8000, + ) + + mock_srun.return_value = _running_exporter() + harness = Harness() + harness._resolve_tachometer_binary = lambda binary_path: binary_path + harness._resolve_host_binary = lambda binary: (_ for _ in ()).throw(AssertionError("not consulted")) + + procs = harness.start_tachometer() + + assert "tachometer_process_exporter" in [proc.name for proc in procs] + pe_call = mock_srun.call_args_list[2] + assert pe_call.kwargs["container_image"] == "pe-with-shell:latest" + assert pe_call.kwargs["container_mounts"] == {Path(tmp_path): Path("/logs")} + assert pe_call.kwargs["command"][:3] == ["/bin/process-exporter", "-config.path", "/logs/process-exporter.yml"] + + def test_resolve_host_binary(self, tmp_path, monkeypatch): + """Absolute paths verbatim; relative ones against SRTCTL_SOURCE_DIR (the + checkout root the sbatch script exports); missing or non-executable -> None.""" + stage = TelemetryStageMixin() + monkeypatch.setenv("SRTCTL_SOURCE_DIR", str(tmp_path)) + + assert stage._resolve_host_binary("configs/process-exporter") is None + + configs = tmp_path / "configs" + configs.mkdir() + binary = configs / "process-exporter" + binary.write_text("#!/bin/sh\n") + assert stage._resolve_host_binary("configs/process-exporter") is None # not executable yet + binary.chmod(0o755) + assert stage._resolve_host_binary("configs/process-exporter") == binary + assert stage._resolve_host_binary(str(binary)) == binary + assert stage._resolve_host_binary("/nonexistent/process-exporter") is None @patch("srtctl.cli.mixins.telemetry_stage.start_srun_process") def test_tachometer_explicit_false_opts_out(self, mock_srun, tmp_path): @@ -904,11 +1210,17 @@ def _compute_frontend_topology(self): ) mock_srun.return_value = _running_exporter() + harness = Harness() + harness._resolve_host_binary = lambda binary: Path("/srt/configs/process-exporter") - processes = Harness().start_tachometer() + processes = harness.start_tachometer() - assert [process.name for process in processes] == ["tachometer_node_exporter", "tachometer"] - assert mock_srun.call_count == 2 + assert [process.name for process in processes] == [ + "tachometer_node_exporter", + "tachometer_process_exporter", + "tachometer", + ] + assert mock_srun.call_count == 3 assert 'name = "dcgm_node-a"' in (tmp_path / "tachometer_config.toml").read_text() @patch("srtctl.cli.mixins.telemetry_stage.start_srun_process") @@ -959,13 +1271,14 @@ def _compute_frontend_topology(self): mock_srun.return_value = _running_exporter() harness = Harness() + harness._resolve_host_binary = lambda binary: Path("/srt/configs/process-exporter") harness.start_tachometer() exporter_calls = [ call for call in mock_srun.call_args_list if call.kwargs.get("nodelist") == ["node-a", "node-b"] ] - assert len(exporter_calls) == 2 + assert len(exporter_calls) == 3 # dcgm, node, process exporters for call in exporter_calls: assert call.kwargs["nodes"] == 2 assert call.kwargs["ntasks"] == 2