HPCE/slurm-efficiency-tools
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

allow later state filtering and include in aggregation

Browse Source
This commit is contained in:
feichtinger
2026-06-26 13:48:28 +02:00
parent 6831e2d1b4
commit 652acc0b9a
1 changed files with 59 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
slurm-eff-tool.py
+59 -9
View File
@@ -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)