Dashboard Integration

Dashboard Integration

How to read a job's telemetry programmatically. This is what the HPC Dashboard does.

Start from the manifest

Its name is derivable from the job id alone — psutil_<jobid>.manifest.json — so a poller never has to glob a directory or parse hostnames out of filenames. It is written before any logger starts, for single-node jobs as well as multi-node, so there is no special case.

{
  "event": "job_manifest",
  "schema": "manifest-v1",
  "job_id": "12345.gadi-pbs",
  "nodes_expected": 2,
  "files": [
    {"hostname": "gadi-cpu-clx-0123", "node_rank": 0,
     "log": "psutil_12345.gadi-pbs_gadi-cpu-clx-0123.log",
     "summary": "psutil_12345.gadi-pbs_gadi-cpu-clx-0123.log.summary.json"},
    {"hostname": "gadi-cpu-clx-0124", "node_rank": 1, "...": "..."}
  ],
  "merged_summary": "psutil_12345.gadi-pbs.summary.json"
}

Filenames are relative to the directory the manifest was found in, so the job does not need to know where gdata is mounted on the reader's side.

nodes_expected is what makes partial state legible: a listed file that does not exist yet is a node still starting, and one that stops growing while others continue is a node whose logger died. Without it, neither is distinguishable from a node that was never part of the job.

Aggregating to a job-level series

# One read offset per file; re-poll each every 15-30s.
async def poll_job(dirpath, job_id, offsets):
    manifest = json.loads((dirpath / f"psutil_{job_id}.manifest.json").read_text())

    for entry in manifest["files"]:
        path = dirpath / entry["log"]
        if not path.exists():
            continue                      # node still starting
        with path.open() as f:
            f.seek(offsets.get(entry["log"], 0))
            for line in f:
                if line.endswith("\n"):   # ignore a partially-flushed tail
                    yield json.loads(line)
            offsets[entry["log"]] = f.tell()

# Bucket to the sampling interval, then sum across nodes.
def to_job_series(samples, bucket_sec=1.0):
    buckets = defaultdict(dict)           # t -> {hostname: cpu_pct_sum}
    for s in samples:
        if s.get("event") != "job_log":
            continue
        t = round(s["t"] / bucket_sec) * bucket_sec
        buckets[t][s["hostname"]] = s["cpu_pct_sum"]

    # Sum per bucket. Keep the node count so a bucket missing a node is visible
    # rather than silently reading as a dip in utilisation.
    return [
        {"t": t,
         "cpu_pct_sum": sum(by_host.values()),
         "nodes_reporting": len(by_host),
         "nodes_expected": len(manifest["files"])}
        for t, by_host in sorted(buckets.items())
    ]

Two things worth getting right:

• Do not treat a missing node in a bucket as zero. Loggers start a second or two apart and tick independently, so early and late buckets legitimately have fewer nodes. Carrying nodes_reporting alongside the sum lets the frontend show a partial bucket as partial instead of as a utilisation dip.
• Only parse complete lines. The logger flushes at least every 2s, so a poll can catch a partly-written final line. Checking for the trailing newline and leaving the offset before it is enough.

The same applies to per-GPU readings: keep them keyed by host. Concatenating each node's GPU array into one flat list gives a GPU an index that depends on how many GPUs every earlier node contributed to that bucket, so one node missing a bucket shifts every GPU after it.

Schema stability

The field names are a contract with the dashboard, enforced by tests/schema_contract.rs. New fields are added alongside old ones; every field added since the first release is #[serde(default)], so archived logs still parse.

test_sample.ndjson is a representative file, checked against the real types by that test.

Legacy fields

Retained for backwards compatibility. Do not use them for new work:

Legacy field	Use instead	Why
rss_bytes_sum	cgroup_mem_bytes	Summed RSS double-counts shared pages — badly wrong for MPI.
io_*	cgroup_io_*	Per-process I/O counters vanish when children exit.
swap_bytes	cgroup_swap_bytes	Same.
system_cpu_efficiency	cpu_efficiency_pct	The old field is a duplicate of cpu_pct_sum, not an efficiency.
tree_percpu_pct	system_percpu_pct	Identical; cgroup cpuacct cannot attribute per-CPU time to individual processes.

Service Units

The logger emits cpu_core_hours. Service Units are core-hours multiplied by the queue's charge rate, which lives in the dashboard's rate table — the logger has no way to know the rate for the queue it is running in.

On NCI the charge is:

Job Cost (SU) = charge rate x max(NCPUs, mem ÷ mem-per-core) x walltime USED

Note max(...): a job booking few cores and a lot of memory is charged for the memory, not the cores. And walltime used, not booked — overbooking walltime costs nothing directly, though it does hurt queue priority.