feat(observability): combined host-attribution coverage (#356 + #413 + #415) + scraper label fix

.gitignore

@@ -51,6 +51,7 @@ bin/
 configs/nats-server
 configs/etcd
 configs/etcdctl
+configs/process-exporter
 configs/*.whl
 configs/*.deb
 configs/*.tar.gz

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))"; \

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 `<log_dir>/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=<compute_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=<compute_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 `<log_dir>/<storage_subdir>/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 `<log_dir>/<storage_subdir>/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`.
 

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_<node>.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.

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]: