From 0c120b55ad73c128ef35cb6edce91ffeb361089b Mon Sep 17 00:00:00 2001 From: Derek Feichtinger Date: Fri, 26 Jun 2026 18:40:46 +0200 Subject: [PATCH] Fix: move record filtering out of original record building --- slurm-eff-tool.py | 33 ++++++++++++++++++++------------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/slurm-eff-tool.py b/slurm-eff-tool.py index b340f17..a75c853 100755 --- a/slurm-eff-tool.py +++ b/slurm-eff-tool.py @@ -622,9 +622,7 @@ def is_top_level_row(row: dict[str, str]) -> bool: return "." not in jid and "_" not in jid -def build_job_records(rows: list[dict[str, str]], - filter_user: str | None = None, - filter_state: str | None = None) -> list[JobRecord]: +def build_job_records(rows: list[dict[str, str]]) -> 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: @@ -637,21 +635,13 @@ def build_job_records(rows: list[dict[str, str]], records: list[JobRecord] = [] - 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 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. @@ -766,6 +756,20 @@ def build_job_records(rows: list[dict[str, str]], return records +def filter_records(records: list[JobRecord], + filter_user: str | None = None, + filter_state: str | None = None) -> list[JobRecord]: + + allowed_states = [] + if filter_state: + allowed_states = filter_state.split(",") + records = [r for r in records if r.state not in allowed_states] + + if filter_user: + records = [r for r in records if r.username == filter_user] + + return records + def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> list[OutputRow]: """Aggregate records according to given grouping instructions.""" @@ -1142,8 +1146,11 @@ def main(argv: list[str] | None = None) -> int: if args.output_raw: write_cache_raw(args.output_raw, rows) - records = build_job_records(rows, filter_user=args.user, - filter_state=args.state) + records = build_job_records(rows) + + records = filter_records(records, + filter_user=args.user, + filter_state=args.state) if args.write_binary_cache: write_binary_cache(records, args.write_binary_cache)