From 652acc0b9abcae38e439fd60f58077026d942344 Mon Sep 17 00:00:00 2001 From: Derek Feichtinger Date: Fri, 26 Jun 2026 13:48:28 +0200 Subject: [PATCH] allow later state filtering and include in aggregation --- slurm-eff-tool.py | 68 ++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 59 insertions(+), 9 deletions(-) diff --git a/slurm-eff-tool.py b/slurm-eff-tool.py index c6cbc5f..8b001cc 100755 --- a/slurm-eff-tool.py +++ b/slurm-eff-tool.py @@ -61,6 +61,7 @@ SACCT_FIELDS = [ ALL_COLUMNS = [ "username", "JobID", + "state", "Count", "NTasks", "CPUs", @@ -85,6 +86,7 @@ PRESET_COLUMNS = { "default": [ "username", "JobID", + "state", "Count", "NTasks", "CPUs", @@ -151,6 +153,23 @@ NUMERIC_COLUMNS = { "waste_Mem", } +state_mappings = { + "BOOT_FAIL": 0, + "CANCELLED": 1, + "COMPLETED": 2, + "DEADLINE": 3, + "FAILED": 4, + "NODE_FAIL": 5, + "OUT_OF_MEMORY": 6, + "PENDING": 7, + "PREEMPTED": 8, + "RUNNING": 9, + "SUSPENDED": 10, + "TIMEOUT": 11, +} + +state_mappings_inv = dict(zip(state_mappings.values(), state_mappings.keys())) + ########################################## # Definitions for expression evaluations # ########################################## @@ -262,6 +281,7 @@ class JobRecord: reqtasks: int cpus: int nodes: int + state: int reqmem_gb: float | None mem_per_cpu_gb: float | None mem_alloc_tres: float | None @@ -333,6 +353,7 @@ class UsedTresStruct: class OutputRow: username: str JobID: str + state: str NTasks: int CPUs: int Nodes: int @@ -359,6 +380,7 @@ class OutputRow: d: dict[str, Any] = { "username": self.username, "JobID": self.JobID, + "state": self.state, "NTasks": self.NTasks, "CPUs": self.CPUs, "Nodes": self.Nodes, @@ -578,7 +600,8 @@ def is_top_level_row(row: dict[str, str]) -> bool: def build_job_records(rows: list[dict[str, str]], - only_user: str | None = None) -> list[JobRecord]: + filter_user: str | None = None, + filter_state: str | None = None) -> list[JobRecord]: """Collapse sacct top-level and step rows into one JobRecord per base job.""" grouped: dict[str, list[dict[str, str]]] = defaultdict(list) for row in rows: @@ -590,12 +613,24 @@ def build_job_records(rows: list[dict[str, str]], grouped[base_job_id(row.get("JobIDRaw") or row.get("JobID", ""))].append(row) records: list[JobRecord] = [] - # Each group consists of sacct rows belonging to a job + + allowed_states = [] + if filter_state: + allowed_states = filter_state.split(",") + + # Each group consists of sacct rows belonging to job steps of a single job for _, group in grouped.items(): top = next((r for r in group if is_top_level_row(r)), group[0]) - if only_user and top.get("User") != only_user: + if filter_user and top.get("User") != filter_user: continue + state = top.get("State") or "UNK" + if state.startswith("CANCELLED by"): + state = "CANCELLED" + if filter_state and state not in allowed_states: + continue + stateid = state_mappings[state] + # seff-style practical peak RSS: maximum MaxRSS across non-extern job steps. # This is intentionally not a sum over all ranks/tasks. step_rows = [ @@ -688,6 +723,7 @@ def build_job_records(rows: list[dict[str, str]], reqtasks=req_ntasks, cpus=alloc_cpus, nodes=nodes, + state = stateid, reqmem_gb=reqmem_gb, mem_used_tres=mem_used_tres_gb, mem_per_cpu_gb=mem_per_cpu_gb, @@ -723,8 +759,14 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis matched = False for pat, rx in compiled: if rx.search(rec.jobname): - key = (pat, rec.username, rec.reqtasks, rec.cpus, rec.nodes, - rec.reqmem_gb, rec.reqwall_hours) + key = (pat, + rec.username, + rec.state, + rec.reqtasks, + rec.cpus, + rec.nodes, + rec.reqmem_gb, + rec.reqwall_hours) buckets[key].append(rec) matched = True break @@ -741,10 +783,15 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis if args.aggr_user: buckets = defaultdict(list) for rec in records: - key = (rec.username, rec.reqtasks, rec.cpus, rec.nodes, - rec.reqmem_gb, rec.reqwall_hours) + key = (rec.username, + rec.state, + rec.reqtasks, + rec.cpus, + rec.nodes, + rec.reqmem_gb, + rec.reqwall_hours) buckets[key].append(rec) - return [make_aggregate_row(v, username=k[0], jobname=f"{common_prefix([r.jobname for r in v])}", dflt_mpcpu=default_mempercpu_gb) for k, v in buckets.items()] + return [make_aggregate_row(v, username=k[0],jobname=f"{common_prefix([r.jobname for r in v])}", dflt_mpcpu=default_mempercpu_gb) for k, v in buckets.items()] return [make_single_row(r, dflt_mpcpu=default_mempercpu_gb) for r in records] @@ -765,6 +812,7 @@ def make_single_row(rec: JobRecord, dflt_mpcpu: float) -> OutputRow: return OutputRow( username=rec.username, JobID=rec.raw.get("JobIDRaw", rec.raw.get("JobID", "")), + state=state_mappings_inv[rec.state], NTasks=rec.reqtasks, CPUs=rec.cpus, Nodes=rec.nodes, @@ -826,6 +874,7 @@ def make_aggregate_row(records: list[JobRecord], username: str, return OutputRow( username=username, JobID="", + state=state_mappings_inv[first.state], NTasks=first.reqtasks, CPUs=first.cpus, Nodes=first.nodes, @@ -1048,7 +1097,8 @@ def main(argv: list[str] | None = None) -> int: if args.output_cache: write_cache(args.output_cache, rows) - records = build_job_records(rows, only_user=args.user) + records = build_job_records(rows, filter_user=args.user, + filter_state=args.state) out_rows = aggregate_records(records, args) # , dflt_mpcpu=default_mempercpu_gb if args.expr: expr = SafeExpression(args.expr)