HPCE/slurm-efficiency-tools
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fbab3ecdf23d7643ccaf685120465eae86d6cd8a
slurm-efficiency-tools/slurm-eff-tool.py
T

1687 lines
54 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3.11
"""
slurm-eff-tool.py - Slurm job efficiency analysis and reporting and tool.
Also targeted at automation and backfeeding use cases.
Efficiency definitions:
CPU_Eff = CPU_time / walltime * 100
Time_Eff = walltime / timelimit * 100
Mem_Eff = 100 (slurm recorded mem usage) / (mem allocated by system)
Waste definitions:
waste_CPU = walltime * (100 - CPU_Eff) * n_cpus [hours]
waste_Mem = walltime * (100 - Mem_eff) * \
(req_mem - (n_cpus * default_mem_per_cpu )) [GB*hours]
waste_total = waste_CPU + waste_Mem/default_mem_per_cpu
2006 D. Feichtinger <derek.feichtinger@psi.ch>
"""
from __future__ import annotations
import argparse
import csv
import json
import math
import os
import re
import statistics
import subprocess
import sys
import signal
import tempfile
import time
from collections import defaultdict
from dataclasses import dataclass, field, fields, asdict
from pathlib import Path
from subprocess import run
import ast
import operator as op
from typing import Any, Iterable, Mapping, Union, List, Tuple, cast
import msgpack
import gzip
VERSION=0.3
# Default GB per CPUs for the cluster
default_mempercpu_gb = 2.2
SACCT_FIELDS = [
"JobIDRaw",
"JobID",
"User",
"Partition",
"JobName",
"State",
"AllocCPUS",
"NNodes",
"ReqMem",
"ElapsedRaw",
"TimelimitRaw",
"CPUTimeRAW",
"TotalCPU",
"MaxRSS",
"ReqTRes",
"AllocTRES",
"TRESUsageInTot",
"PlannedCPURAW"
]
ALL_COLUMNS = [
"username",
"JobID",
"state",
"Count",
"Partition",
"NTasks",
"CPUs",
"Nodes",
"CPU_Eff",
"waste_CPU",
"ReqMem",
"AllocMem",
"UsedMem",
"MaxRSS_max",
"MemPerCPU",
"Mem_Eff",
"Planned_Time",
"Elig_Qtime",
"waste_Mem",
"ReqWalltime",
"Walltime",
"Walltime_max",
"Time_Eff",
"waste_total",
"jobname",
]
PRESET_COLUMNS = {
"default": [
"username",
"JobID",
"state",
"Count",
"NTasks",
"CPUs",
"Nodes",
"CPU_Eff",
"waste_CPU",
"AllocMem",
"UsedMem",
"MaxRSS_max",
"MemPerCPU",
"Mem_Eff",
"waste_Mem",
"ReqWalltime",
"Walltime",
"Walltime_max",
"Time_Eff",
"jobname",],
"all": ALL_COLUMNS,
"eff": [
"username",
"JobID",
"Count",
"CPU_Eff",
"Mem_Eff",
"Time_Eff",
"jobname",],
"time": [
"username",
"JobID",
"state",
"Count",
"Partition",
"NTasks",
"CPUs",
"Nodes",
"ReqWalltime",
"Walltime",
"Walltime_max",
"Time_Eff",
# "Planned_Time",
"Elig_Qtime",
"jobname",],
"cpu": [
"username",
"JobID",
"state",
"Count",
"Partition",
"NTasks",
"CPUs",
"Nodes",
"CPU_Eff",
"waste_CPU",
"jobname",],
"mem": [
"username",
"JobID",
"state",
"Count",
"Partition",
"NTasks",
"CPUs",
"Nodes",
"AllocMem",
"UsedMem",
"MaxRSS_max",
"MemPerCPU",
"Mem_Eff",
"waste_Mem",
"jobname",],
"waste": [
"username",
"JobID",
"state",
"Count",
"NTasks",
"CPUs",
"Nodes",
"AllocMem",
"MemPerCPU",
"CPU_Eff",
"Mem_Eff",
"Time_Eff",
"waste_CPU",
"waste_Mem",
"waste_total",
"jobname",],
}
# One-character aliases for sorting and output format specifications.
ALIASES = {
"u": "username",
"i": "JobID",
"c": "CPUs",
"N": "Nodes",
"m": "ReqMem",
"n": "UsedMem",
"p": "MemPerCPU",
"l": "ReqWalltime",
"C": "Count",
"e": "CPU_Eff",
"M": "Mem_Eff",
"t": "Time_Eff",
"j": "jobname",
"X": "waste_CPU",
"Y": "waste_Mem",
"T": "waste_total",
}
NUMERIC_COLUMNS = {
"CPUs",
"NTasks",
"Nodes",
"ReqMem",
"AllocMem",
"UsedMem",
"MemPerCPU",
"Elig_Qtime",
"ReqWalltime",
"Count",
"CPU_Eff",
"MaxRSS_max",
"Mem_Eff",
"Time_Eff",
"Walltime_max",
"waste_CPU",
"waste_Mem",
"waste_total",
}
HISTO_COLUMNS = {
"CPU_Eff",
"Elig_Qtime",
"Mem_Eff",
"Planned_Time",
"Time_Eff",
"UsedMem",
"Walltime",
}
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 #
##########################################
_ALLOWED_BINOPS = {
ast.Add: op.add,
ast.Sub: op.sub,
ast.Mult: op.mul,
ast.Div: op.truediv,
ast.FloorDiv: op.floordiv,
ast.Mod: op.mod,
ast.Pow: op.pow,
}
_ALLOWED_UNARYOPS = {
ast.UAdd: op.pos,
ast.USub: op.neg,
ast.Not: op.not_,
}
_ALLOWED_CMPOPS = {
ast.Eq: op.eq,
ast.NotEq: op.ne,
ast.Lt: op.lt,
ast.LtE: op.le,
ast.Gt: op.gt,
ast.GtE: op.ge,
ast.In: lambda a, b: a in b,
ast.NotIn: lambda a, b: a not in b,
}
##########################################
####################################
# Class for expression evaluations #
####################################
# TODO: Note that we currently cannot filter on state strings, because
# we map state to numeric ID values
SEValue = Union[int, float, str, bool, None]
SEEvalValue = Union[SEValue, List[SEValue], Tuple[SEValue, ...]]
class SafeExpression:
def __init__(self, expression: str):
self.expression = expression.lower()
self.tree = ast.parse(self.expression, mode="eval")
def evaluate(self, variables: Mapping[str, SEValue]) -> bool:
# normalize to downcased variable names
variables = {k.lower(): v for k,v in variables.items()}
return bool(self._eval(self.tree.body, variables))
def _eval(self, node: ast.AST, variables: Mapping[str, SEValue]) -> SEEvalValue:
if isinstance(node, ast.Constant):
if isinstance(node.value, (int, float, str, bool)):
return node.value
raise ValueError(f"Unsupported constant: {node.value!r}")
if isinstance(node, ast.List):
return [cast(SEValue, self._eval(element, variables))
for element in node.elts]
if isinstance(node, ast.Tuple):
return tuple(cast(SEValue, self._eval(element, variables))
for element in node.elts)
# Variable names
if isinstance(node, ast.Name):
name = node.id.lower()
try:
return variables[name]
except KeyError:
raise ValueError(f"Unknown variable: {node.id}") from None
if isinstance(node, ast.BinOp):
op_type = type(node.op)
if op_type not in _ALLOWED_BINOPS:
raise ValueError(f"Unsupported operator: {op_type.__name__}")
left = self._eval(node.left, variables)
right = self._eval(node.right, variables)
# deal with columns containing None value
if left is None or right is None:
return False
return _ALLOWED_BINOPS[op_type](left, right)
if isinstance(node, ast.UnaryOp):
op_type = type(node.op)
if op_type not in _ALLOWED_UNARYOPS:
raise ValueError(f"Unsupported unary operator: {op_type.__name__}")
return _ALLOWED_UNARYOPS[op_type](self._eval(node.operand, variables))
if isinstance(node, ast.BoolOp):
if isinstance(node.op, ast.And):
for value in node.values:
if not self._eval(value, variables):
return False
return True
if isinstance(node.op, ast.Or):
for value in node.values:
if self._eval(value, variables):
return True
return False
raise ValueError(f"Unsupported boolean operator: {type(node.op).__name__}")
if isinstance(node, ast.Compare):
left = self._eval(node.left, variables)
for operator_node, comparator in zip(node.ops, node.comparators):
op_type = type(operator_node)
if op_type not in _ALLOWED_CMPOPS:
raise ValueError(f"Unsupported comparison: {op_type.__name__}")
right = self._eval(comparator, variables)
# deal with columns containing None value
if left is None or right is None:
return False
if not _ALLOWED_CMPOPS[op_type](left, right):
return False
left = right
return True
raise ValueError(f"Unsupported expression element: {type(node).__name__}")
@dataclass
class JobRecord:
jobid: str
jobidraw: str
username: str
partition: str
jobname: str
reqtasks: int
cpus: int
nodes: int
state: int
reqmem_gb: float | None
mem_per_cpu_gb: float | None
mem_alloc_tres: float | None
mem_used_tres: float | None
planned_sec: float
reqwall_hours: float | None
reqmem_bytes_total: float | None
elapsed_sec: int
elig_qtime_sec: float
timelimit_sec: int
totalcpu_sec: float
maxrss_bytes: float | None
cpu_eff: float | None
mem_eff: float | None
time_eff: float | None
@dataclass
class AllocTresStruct:
mem: int | None = None
cpu: int = 0
def __post_init__(self):
if isinstance(self.mem, str):
self.mem = parse_size_to_bytes(self.mem)
if isinstance(self.cpu, str):
self.cpu = int(self.cpu)
@classmethod
def from_tres_string(cls,tres_str: str) -> AllocTresStruct:
"""Parse comma separated fields of a TRES string."""
if tres_str == "":
return AllocTresStruct()
supported_fields = {f.name for f in fields(AllocTresStruct)}
result = {}
for f in tres_str.split(","):
key,val = f.split("=", maxsplit=1)
result[key] = val
values = {k: v for k, v in result.items() if k in supported_fields}
return cls(**values)
@dataclass
class UsedTresStruct:
mem: int | None = None
cpu: str = ""
def __post_init__(self):
if isinstance(self.mem, str):
self.mem = parse_size_to_bytes(self.mem)
@classmethod
def from_tres_string(cls,tres_str: str) -> UsedTresStruct:
"""Parse comma separated fields of a TRESUsageInTot string."""
if tres_str == "":
return UsedTresStruct()
supported_fields = {f.name for f in fields(UsedTresStruct)}
result = {}
for f in tres_str.split(","):
key,val = f.split("=", maxsplit=1)
result[key] = val
values = {k: v for k, v in result.items() if k in supported_fields}
return cls(**values)
@dataclass
class OutputRow:
username: str
JobID: str
state: str
Partition: str
NTasks: int
CPUs: int
Nodes: int
ReqMem: float | None
AllocMem: float | None
UsedMem: float | None
MemPerCPU: float | None
ReqWalltime: float | None
Planned_Time: float | None
Elig_Qtime: float | None
Count: int
CPU_Eff: float | None
Mem_Eff: float | None
Time_Eff: float | None
jobname: str
maxrss_max: float | None
walltime: float | None # in h
walltime_max: float | None # in h
waste_CPU: float | None # CPU h
waste_Mem: float | None
waste_total: float | None
vectors: dict[str, list[int | float]] # here we can store full vectors of the aggreg.
def as_dict(self, sdev: bool = False) -> dict[str, Any]:
d: dict[str, Any] = {
"username": self.username,
"JobID": self.JobID,
"state": self.state,
"Partition": self.Partition,
"NTasks": self.NTasks,
"CPUs": self.CPUs,
"Nodes": self.Nodes,
"ReqMem": self.ReqMem,
"AllocMem": self.AllocMem,
"UsedMem": self.UsedMem,
"MaxRSS_max": self.maxrss_max,
"MemPerCPU": self.MemPerCPU,
"ReqWalltime": self.ReqWalltime,
"Walltime": self.walltime,
"Walltime_max": self.walltime_max,
"Elig_Qtime": self.Elig_Qtime,
"Planned_Time": self.Planned_Time,
"Count": self.Count,
"CPU_Eff": self.CPU_Eff,
"Mem_Eff": self.Mem_Eff,
"Time_Eff": self.Time_Eff,
"jobname": self.jobname,
"waste_CPU": self.waste_CPU,
"waste_Mem": self.waste_Mem,
"waste_total": self.waste_total,
"vectors": self.vectors
}
if sdev:
for col, vals in [
("CPU_Eff", self.vectors['cpu_eff']),
("Mem_Eff", self.vectors['mem_eff']),
("Time_Eff", self.vectors['time_eff']),
]:
d[f"{col}_sdev"] = stdev_or_none(vals)
d[f"{col}_max"] = max(vals) if vals else None
d[f"{col}_min"] = min(vals) if vals else None
return d
def die(msg: str, code: int = 2) -> None:
print(f"error: {msg}", file=sys.stderr)
raise SystemExit(code)
def slurm_duration_to_seconds(value: str) -> float:
"""Parse Slurm-ish CPU time strings such as 01:02:03, 2-01:02:03, 5:12.345."""
value = (value or "").strip()
if not value or value in {"Unknown", "UNLIMITED", "Partition_Limit"}:
return 0.0
days = 0
if "-" in value:
d, value = value.split("-", 1)
days = int(d)
parts = value.split(":")
try:
if len(parts) == 3:
h, m, s = parts
sec = float(s)
return days * 86400 + int(h) * 3600 + int(m) * 60 + sec
if len(parts) == 2:
m, s = parts
return days * 86400 + int(m) * 60 + float(s)
if len(parts) == 1:
return days * 86400 + float(parts[0])
except ValueError:
return 0.0
return 0.0
def format_seconds(seconds: int | float | None) -> str:
if seconds is None or seconds <= 0:
return "Unknown"
seconds = int(round(seconds))
days, rem = divmod(seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes, sec = divmod(rem, 60)
if days:
return f"{days}-{hours:02d}:{minutes:02d}:{sec:02d}"
return f"{hours:02d}:{minutes:02d}:{sec:02d}"
def parse_size_to_bytes(value: str) -> int | None:
"""Parse Slurm memory size fields such as 1024K, 2000M, 8Gn, 4Gc, 1.5T."""
s = (value or "").strip()
if not s or s in {"Unknown", "0", "0K", "0M", "0G", "0T"}:
return None
# Slurm ReqMem may end in c/n for per-CPU or per-node. Strip that here.
if s[-1].lower() in {"c", "n"}:
s = s[:-1]
m = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)([KMGTP]?)", s, flags=re.I)
if not m:
return None
num = int(m.group(1))
unit = m.group(2).upper() or "K" # Slurm memory fields are normally KiB when unitless.
mult = {
"K": 1024,
"M": 1024**2,
"G": 1024**3,
"T": 1024**4,
"P": 1024**5,
}[unit]
return num * mult
def reqmem_total_bytes(reqmem: str, cpus: int, nodes: int) -> float | None:
"""Convert ReqMem into total requested bytes for the whole job allocation."""
raw = (reqmem or "").strip()
base = parse_size_to_bytes(raw)
if base is None:
return None
if raw.lower().endswith("c"):
return base * max(cpus, 1)
if raw.lower().endswith("n"):
return base * max(nodes, 1)
# Default Slurm ReqMem suffix is usually n: memory per node.
return base
def common_prefix(names: list[str]) -> str:
if not names:
return ""
prefix = os.path.commonprefix(names)
# Avoid ugly partial-token prefixes when possible.
prefix = re.sub(r"[^A-Za-z0-9_.-]+$", "", prefix)
if len(names) == 1:
return names[0]
namepattern = "*"
if prefix:
namepattern = f"{prefix}*"
return namepattern
def stdev_or_none(values: list[float]) -> float | None:
if len(values) < 2:
return 0.0 if len(values) == 1 else None
return statistics.stdev(values)
def mean_or_none(values: list[float]) -> float | None:
return statistics.mean(values) if values else None
def median_or_none(values: list[float]) -> float | None:
return statistics.median(values) if values else None
def pct(numerator: float | None, denominator: float | None) -> float | None:
if numerator is None or denominator is None or denominator <= 0:
return None
return 100.0 * numerator / denominator
def run_sacct(args: argparse.Namespace) -> list[dict[str, str]]:
cmd = [
"sacct",
"-P",
"-n",
"--units=K",
"--format=" + ",".join(SACCT_FIELDS),
]
if args.start:
cmd += ["-S", args.start]
if args.end:
cmd += ["-E", args.end]
if args.user:
cmd += ["-u", args.user]
if args.state:
cmd += ["--state", args.state]
# Include job steps because MaxRSS often lives on batch/extern/step rows.
# We later collapse rows back to the base job ID.
try:
proc = subprocess.run(cmd, text=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except FileNotFoundError:
die("sacct not found in PATH")
except subprocess.CalledProcessError as exc:
die(f"sacct failed with exit code {exc.returncode}:\n{exc.stderr.strip()}")
rows = parse_pipe_rows(proc.stdout.splitlines())
return rows
def parse_pipe_rows(lines: Iterable[str]) -> list[dict[str, str]]:
rows: list[dict[str, str]] = []
reader = csv.reader(lines, delimiter="|")
for parts in reader:
if not parts:
continue
parts = [p.strip() for p in parts]
if len(parts) < len(SACCT_FIELDS):
parts += [""] * (len(SACCT_FIELDS) - len(parts))
rows.append(dict(zip(SACCT_FIELDS, parts[: len(SACCT_FIELDS)])))
return rows
def write_cache_raw(path: str, rows: list[dict[str, str]]) -> None:
"""Write a raw sacct output file"""
if os.path.exists(path):
sys.stderr.write(f"WARNING, file exists: {path}. Balking out...\n")
sys.exit(1)
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=SACCT_FIELDS, delimiter="|", lineterminator="\n")
writer.writeheader()
writer.writerows(rows)
def read_cache_raw(path: str) -> list[dict[str, str]]:
"""Read a raw sacct output file."""
with open(path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f, delimiter="|")
missing = [x for x in SACCT_FIELDS if x not in (reader.fieldnames or [])]
# Older cache files from versions before CPUTimeRAW can still be read.
# CPU efficiency will be NA for those rows unless CPUTimeRAW is present.
missing_required = [x for x in missing if x != "CPUTimeRAW"]
if missing_required:
die(f"cache file is missing expected fields: {', '.join(missing_required)}")
return [{k: (row.get(k) or "").strip() for k in SACCT_FIELDS} for row in reader]
def write_binary_cache(records: list[JobRecord], filename):
if os.path.exists(filename):
sys.stderr.write(f"WARNING, file exists: {filename}. Balking out...\n")
sys.exit(1)
cache = [asdict(r) for r in records]
cache_blob = {
"slurm_eff_version": VERSION,
"cmdline": " ".join(sys.argv),
"created": time.time(),
"records": cache
}
with gzip.open(filename, "wb") as f:
msgpack.pack(cache_blob, f)
def read_binary_cache(filename: str) -> dict[str, str or list[JobRecord]]:
result = {}
with gzip.open(filename, "rb") as f:
blob = msgpack.unpack(f, raw=False)
for mdfield in ['slurm_eff_version', 'cmdline', 'created']:
result[mdfield] = blob.get(mdfield,None)
if result['slurm_eff_version'] != VERSION:
sys.stderr.write(f"WARNING: cache was written by version {result['slurm_eff_version']}, but we are running version {VERSION}\n")
result['records'] = [JobRecord(**d) for d in blob["records"]]
return result
def show_cache_info(filename: str):
cache = read_binary_cache(filename)
for mdfield in ['slurm_eff_version', 'cmdline', 'created']:
if cache[mdfield] is None:
print(f'{mdfield}: [undefined]')
elif mdfield == 'created':
print(f"{mdfield}: {time.ctime(float(cache[mdfield]))}")
else:
print(f'{mdfield}: {cache[mdfield]}')
def base_job_id(jobid: str) -> str:
return re.split(r"[._]", jobid, maxsplit=1)[0]
def is_top_level_row(row: dict[str, str]) -> bool:
jid = row.get("JobID", "")
return "." not in jid and "_" not in jid
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:
# Do not filter by User here. On many Slurm installations, job step rows
# are where MaxRSS is populated, but their User field may be empty or may
# not match the parent job's user. Filtering before grouping drops those
# rows and makes Mem_Eff become NA. Filter after identifying
# the top-level job row instead.
grouped[base_job_id(row.get("JobIDRaw") or row.get("JobID", ""))].append(row)
records: list[JobRecord] = []
# 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])
state = top.get("State") or "UNK"
if state.startswith("CANCELLED by"):
state = "CANCELLED"
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 = [
r for r in group
if "." in (r.get("JobID") or r.get("JobIDRaw") or "")
and ".extern" not in (r.get("JobID") or r.get("JobIDRaw") or "")
]
rss_source_rows = step_rows or [
r for r in group
if ".extern" not in (r.get("JobID") or r.get("JobIDRaw") or "")
]
maxrss_values = [parse_size_to_bytes(r.get("MaxRSS", "")) for r in rss_source_rows]
# stores the largest maxrss value over all job steps
maxrss_cleaned = [x for x in maxrss_values if x is not None]
maxrss = max(maxrss_cleaned, default=None)
# find mem_used_tres over reported job steps. It may be in the
# *.interactive or *.0 steps
used_tres = [UsedTresStruct.from_tres_string(r.get("TRESUsageInTot","")) \
for r in rss_source_rows]
tmp_used_tres_mem = [r.mem for r in used_tres if r.mem is not None]
if len(tmp_used_tres_mem) == 0:
continue
mem_used_tres = max(tmp_used_tres_mem)
mem_used_tres_gb = None
if mem_used_tres is not None:
mem_used_tres_gb = mem_used_tres / (1024**3)
req_tres = AllocTresStruct.from_tres_string(top.get("ReqTRes") or "")
# better than getting NTasks and then filtering out .extern and
# getting max over job steps
req_ntasks = req_tres.cpu
alloc_cpus = int(float(top.get("AllocCPUS") or 0))
# If no CPU was ever allocated, this is a job that never was
# scheduled to run. We exclude it
if alloc_cpus == 0:
continue
nodes = int(float(top.get("NNodes") or 0))
elapsed = int(float(top.get("ElapsedRaw") or 0))
planned_sec = float(top.get("PlannedCPURAW") or 0)
elig_qtime_sec = planned_sec / alloc_cpus
# sacct TimelimitRaw is minutes, while ElapsedRaw and CPUTimeRAW are seconds.
timelimit_raw_minutes = float(top.get("TimelimitRaw") or 0)
timelimit_seconds = int(round(timelimit_raw_minutes * 60))
cputime_raw = float(top.get("CPUTimeRAW") or 0)
totalcpu = slurm_duration_to_seconds(top.get("TotalCPU", ""))
# Memory requested in total by the user
# It is safer to rely on ReqTres.mem, instead constructing from ReqMem
# reqmem = top.get("ReqMem", "")
# reqmem_total = reqmem_total_bytes(reqmem, alloc_cpus, nodes)
reqmem_total = req_tres.mem
alloc_tres = AllocTresStruct.from_tres_string(top.get("AllocTRES") or "")
# Memory allocated by the scheduler, recorded in AllocTRES string
# mem_alloc_tres = parse_tres_mem_bytes(top.get("AllocTRES") or "")
mem_alloc_tres_gb = None
if alloc_tres.mem is not None:
mem_alloc_tres_gb = alloc_tres.mem / (1024**3)
reqmem_gb = None
if reqmem_total is not None:
reqmem_gb = reqmem_total / (1024**3)
reqwall_hours = timelimit_seconds / 3600.0 if timelimit_seconds > 0 else None
mem_per_cpu_gb = mem_alloc_tres_gb / alloc_cpus if reqmem_gb is not None \
and mem_alloc_tres_gb is not None and alloc_cpus > 0 else None
# EFFFICIENCY CALCULATIONS
#
# TODO: find out why cpu_eff for some jobs is > 100%. Probably this is
# related to 2 hyperthreads, and cputime_raw should have been doubled
# for these cases
cpu_eff = pct(totalcpu, cputime_raw)
time_eff = pct(elapsed, timelimit_seconds)
# Old metric:
# mem_eff = (Peak RSS of all job steps) / (user requested memory)
# mem_eff = pct(maxrss, reqmem_total)
# Better metric:
# mem_eff = (sum of peaks over all job steps) / (mem allocated by system)
mem_eff = pct(mem_used_tres, alloc_tres.mem)
records.append(
JobRecord(
jobid=top.get("JobID", ""),
jobidraw=top.get("JobIDRaw", ""),
username=top.get("User", ""),
partition=top.get("Partition", ""),
jobname=top.get("JobName", ""),
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,
mem_alloc_tres=mem_alloc_tres_gb,
planned_sec=planned_sec,
reqwall_hours=reqwall_hours,
reqmem_bytes_total=reqmem_total,
elapsed_sec=elapsed,
elig_qtime_sec=elig_qtime_sec,
timelimit_sec=timelimit_seconds,
totalcpu_sec=totalcpu,
maxrss_bytes=maxrss,
cpu_eff=cpu_eff,
mem_eff=mem_eff,
time_eff=time_eff,
)
)
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 = [state_mappings[state] for state in filter_state.split(",")]
records = [r for r in records if r.state in allowed_states]
if filter_user:
userlist = filter_user.split(',')
records = [r for r in records if r.username in userlist]
return records
def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> list[OutputRow]:
"""Aggregate records according to given grouping instructions."""
global default_mempercpu_gb
if args.dflt_mpcpu:
default_mempercpu_gb = float(args.dflt_mpcpu)
avg_function = median_or_none
if args.avg_function:
avgfns = {"mean": mean_or_none,
"median" : median_or_none}
try:
avg_function = avgfns[args.avg_function]
except KeyError:
sys.stderr.write(f"ERROR: No such averaging function: {args.avg_function}\n")
sys.exit(1)
storvecs = []
if args.histo:
if not (args.aggr_regexp or args.aggr_user):
sys.stderr.write("ERROR: Cannot use histogramming mode without an aggregation mode\n")
sys.exit(1)
storvecs=args.histo.lower().split(",")
if args.aggr_regexp:
compiled = [(pat, re.compile(pat)) for pat in args.aggr_regexp]
buckets: dict[tuple[Any, ...], list[JobRecord]] = defaultdict(list)
unmatched: list[JobRecord] = []
for rec in records:
matched = False
for pat, rx in compiled:
if rx.search(rec.jobname):
key = (pat,
rec.username,
rec.partition,
rec.state,
rec.reqtasks,
rec.cpus,
rec.nodes,
rec.reqmem_gb,
rec.reqwall_hours)
buckets[key].append(rec)
matched = True
break
if not matched:
unmatched.append(rec)
out = [make_aggregate_row(v, username=k[1], jobname=k[0], \
partition=k[2],
dflt_mpcpu=default_mempercpu_gb,
storvecs=storvecs,
avg_function=avg_function) \
for k, v in buckets.items()]
out.extend(make_single_row(r, dflt_mpcpu=default_mempercpu_gb,
storvecs=storvecs) \
for r in unmatched)
return out
if args.aggr_user:
buckets = defaultdict(list)
for rec in records:
key = (rec.username,
rec.partition,
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],partition=k[1], \
jobname=f"{common_prefix([r.jobname for r in v])}",
dflt_mpcpu=default_mempercpu_gb, storvecs=storvecs,
avg_function=avg_function)
for k, v in buckets.items()]
return [make_single_row(r, dflt_mpcpu=default_mempercpu_gb,
storvecs=storvecs) for r in records]
def make_single_row(rec: JobRecord, dflt_mpcpu: float,
storvecs: list[str] | None = None) -> OutputRow:
"""Returns an OutputRow based on a single slurm job record."""
walltime = rec.elapsed_sec / 3600
# TODO elig_qtime_sec = rec.
waste_mem = 0
if rec.mem_eff is not None and rec.reqmem_gb is not None:
waste_mem = max(0,walltime * (100-rec.mem_eff)/100 \
* (rec.reqmem_gb - rec.cpus * dflt_mpcpu))
waste_cpu = 0
if rec.cpu_eff is not None:
# We count failed jobs as wasted CPU!
if rec.state != state_mappings['COMPLETED']:
waste_cpu = walltime * rec.cpus
else:
waste_cpu = walltime * (100-rec.cpu_eff)/100 * rec.cpus
waste_total = waste_cpu + waste_mem/dflt_mpcpu
jobid = rec.jobidraw
if jobid == "":
jobid = rec.jobid
vectors: dict[str, list[int | float]] = {
"cpu_eff": [rec.cpu_eff] if rec.cpu_eff is not None else [],
"mem_eff": [rec.mem_eff] if rec.mem_eff is not None else [],
"time_eff": [rec.time_eff] if rec.time_eff is not None else []
}
if storvecs:
for colname in storvecs:
vectors[colname] = []
if colname == "walltime":
vectors[colname] = [rec.elapsed_sec/3600]
elif colname == "usedmem" and rec.mem_used_tres is not None:
vectors[colname] = [rec.mem_used_tres]
elif colname == "elig_qtime":
vectors[colname] = [rec.elig_qtime_sec/3600]
elif colname == "planned_time":
vectors[colname] = [rec.planned_sec/3600]
elif colname in ["cpu_eff", "mem_eff", "time_eff"]:
pass
else:
sys.stderr.write(f"ERROR: Cannot gather vector information for metric {colname}\n")
sys.exit(1)
return OutputRow(
username=rec.username,
JobID=jobid,
state=state_mappings_inv[rec.state],
Partition=rec.partition,
NTasks=rec.reqtasks,
CPUs=rec.cpus,
Nodes=rec.nodes,
ReqMem=rec.reqmem_gb,
AllocMem=rec.mem_alloc_tres,
UsedMem=rec.mem_used_tres,
MemPerCPU=rec.mem_per_cpu_gb,
ReqWalltime=rec.reqwall_hours,
Planned_Time=rec.planned_sec/3600,
Elig_Qtime=rec.elig_qtime_sec/3600,
Count=1,
CPU_Eff=rec.cpu_eff,
Mem_Eff=rec.mem_eff,
Time_Eff=rec.time_eff,
jobname=rec.jobname,
maxrss_max=rec.maxrss_bytes / (1024**3) if rec.maxrss_bytes is not None else None,
walltime=walltime,
walltime_max= walltime,
waste_Mem=waste_mem,
waste_CPU=waste_cpu,
waste_total=waste_total,
vectors = vectors
)
def make_aggregate_row(records: list[JobRecord], username: str, partition: str,
jobname: str, dflt_mpcpu: float,
storvecs: list[str] | None = None,
avg_function=median_or_none) -> OutputRow:
"""Returns an OutputRow based on the given list of job records."""
first = records[0]
cpu_eff_vals = [r.cpu_eff for r in records if r.cpu_eff is not None]
mem_eff_vals = [r.mem_eff for r in records if r.mem_eff is not None]
time_eff_vals = [r.time_eff for r in records if r.time_eff is not None]
walltime=avg_function([r.elapsed_sec for r in records if r.elapsed_sec is not None])
if walltime is not None:
walltime /= 3600
walltime_max=max([r.elapsed_sec for r in records
if r.elapsed_sec is not None]) / 3600
planned_sec = avg_function([r.planned_sec for r in records
if r.planned_sec is not None])
planned_time_h = None
if planned_sec is not None:
planned_time_h = planned_sec/3600
elig_qtime_sec = avg_function([r.elig_qtime_sec for r in records
if r.elig_qtime_sec is not None])
elig_qtime_h = None
if elig_qtime_sec is not None:
elig_qtime_h = elig_qtime_sec/3600
# We average over all jobs' allocated memory. Some jobs could have received
# different allocations, even though all of them had the same user required
# Memory. Maybe should generate a warning. Most of the time, what the user
# requested should match what the scheduler gave
alloc_mem = avg_function([r.mem_alloc_tres for r in records if r.mem_alloc_tres is not None])
used_mem = avg_function([r.mem_used_tres for r in records if r.mem_used_tres is not None])
maxrss_max = max([r.maxrss_bytes for r in records
if r.maxrss_bytes is not None]) / (1024**3) or None
memory_efficiency=avg_function(mem_eff_vals)
count=len(records)
waste_mem = 0
if (memory_efficiency is not None) and (walltime is not None) \
and first.reqmem_gb is not None:
waste_mem = max(0,count * walltime * (100-memory_efficiency)/100 \
* (first.reqmem_gb - first.cpus * dflt_mpcpu))
cpu_efficiency = avg_function(cpu_eff_vals)
waste_cpu=0
if cpu_efficiency is not None and walltime is not None:
# We count failed jobs as wasted CPU!
if first.state != state_mappings['COMPLETED']:
waste_cpu = count * walltime * first.cpus
else:
waste_cpu = count * walltime * (100-cpu_efficiency)/100 * first.cpus
waste_total = waste_cpu + waste_mem/dflt_mpcpu
vectors: dict[str, list[int | float]] = {
"cpu_eff": [r.cpu_eff for r in records if r.cpu_eff is not None],
"mem_eff": [r.mem_eff for r in records if r.mem_eff is not None],
"time_eff": [r.time_eff for r in records if r.time_eff is not None],
}
if storvecs:
for colname in storvecs:
if colname == "walltime":
vectors[colname] = [r.elapsed_sec/3600 for r in records
if r.elapsed_sec is not None]
elif colname == "usedmem":
vectors[colname] = [r.mem_used_tres for r in records
if r.mem_used_tres is not None]
elif colname == "elig_qtime":
vectors[colname] = [r.elig_qtime_sec/3600 for r in records
if r.elig_qtime_sec is not None]
elif colname == "planned_time":
vectors[colname] = [r.planned_sec/3600 for r in records
if r.planned_sec is not None]
elif colname in ["cpu_eff", "mem_eff", "time_eff"]:
pass
else:
sys.stderr.write(f"ERROR: Cannot gather vector information for metric {colname}\n")
sys.exit(1)
return OutputRow(
username=username,
JobID="",
state=state_mappings_inv[first.state],
Partition=partition,
NTasks=first.reqtasks,
CPUs=first.cpus,
Nodes=first.nodes,
ReqMem=first.reqmem_gb,
AllocMem=alloc_mem,
UsedMem=used_mem,
MemPerCPU=first.mem_per_cpu_gb,
ReqWalltime=first.reqwall_hours,
Planned_Time=planned_time_h,
Elig_Qtime=elig_qtime_h,
Count=count,
CPU_Eff=cpu_efficiency,
Mem_Eff=memory_efficiency,
Time_Eff=avg_function(time_eff_vals),
jobname=jobname,
maxrss_max=maxrss_max,
walltime=walltime,
walltime_max=walltime_max,
waste_Mem=waste_mem,
waste_CPU=waste_cpu,
waste_total=waste_total,
vectors = vectors,
)
def resolve_column_name(name: str) -> str:
"""Returns canonicalized full column name, accepts one letter column codes"""
n = name.strip()
reverse = {v.lower(): v for v in ALL_COLUMNS}
reverse.update({v.lower(): v for v in NUMERIC_COLUMNS})
if n in ALIASES:
return ALIASES[n]
if n.lower() in reverse:
return reverse[n.lower()]
die(f"unknown column/alias: {name}")
return ""
def sort_rows(rows: list[OutputRow], spec: str | None) -> list[OutputRow]:
if not spec:
return rows
terms = [x.strip() for x in spec.split(",") if x.strip()]
if len(terms) > 3:
die("--sort supports at most three columns")
parsed: list[tuple[str, bool]] = []
for t in terms:
desc = t.startswith("-")
asc = t.startswith("+")
name = t[1:] if (desc or asc) else t
col = resolve_column_name(name)
parsed.append((col, desc))
sorted_rows = rows
# Stable-sort from least significant to most significant.
for col, desc in reversed(parsed):
if col in NUMERIC_COLUMNS:
sorted_rows = sorted(
sorted_rows,
key=lambda r: float("-inf") if getattr(r, col) is None else getattr(r, col),
reverse=desc,
)
else:
sorted_rows = sorted(
sorted_rows,
key=lambda r: "" if getattr(r, col) is None else str(getattr(r, col)),
reverse=desc,
)
return sorted_rows
# adds sdev columns behind their respective avg column
def columns_for_sdev(base_cols: list[str]) -> list[str]:
out: list[str] = []
for col in base_cols:
out.append(col)
if col in {"CPU_Eff", "Mem_Eff", "Time_Eff"}:
out += [f"{col}_sdev", f"{col}_max", f"{col}_min"]
return out
def format_value(value: Any, column: str | None = None) -> str:
if value is None:
return "NA"
if column == "Nr.":
return f"{value})"
if column in ["ReqMem", "UsedMem","MemPerCPU", "MaxRSS_max", "AllocMem"]:
return f"{value:.2f}G"
if column in ["ReqWalltime", "Walltime", "Walltime_max",
"Elig_Qtime", "Planned_Time"]:
return f"{value:.2f}h"
if isinstance(value, float):
return f"{value:.2f}"
return str(value)
def print_table(rows: list[OutputRow], columns: list[str], sdev: bool) -> None:
dicts = [r.as_dict(sdev=sdev) for r in rows]
columns = columns_for_sdev(columns) if sdev else columns
# Add row numbers
counter = 0
for d in dicts:
d['Nr.'] = counter
counter += 1
columns = ['Nr.'] + columns
widths = {col: len(col) for col in columns}
for d in dicts:
for col in columns:
widths[col] = max(widths[col], len(format_value(d.get(col), col)))
print(" ".join(col.ljust(widths[col]) for col in columns))
print(" ".join("-" * widths[col] for col in columns))
for d in dicts:
print(" ".join(format_value(d.get(col), col).ljust(widths[col]) for col in columns))
def columns_from_fmtstr(fmt: str) -> list[str]:
"""Convert output format string to list of expanded column names."""
candidates = fmt.split(",")
return [resolve_column_name(col) for col in candidates]
# HISTOGRAMMING CODE START
from bisect import bisect_right
from collections.abc import Iterable, Sequence
from math import ceil, exp, floor, log, log10
Metricnumber = Union[int, float]
def percentile(sorted_values: Sequence[float], p: float) -> float:
"""Return percentile p in [0, 100] using linear interpolation."""
if not sorted_values:
raise ValueError("empty data")
if not 0 <= p <= 100:
raise ValueError("percentile must be in [0, 100]")
k = (len(sorted_values) - 1) * (p / 100.0)
f = floor(k)
c = ceil(k)
if f == c:
return float(sorted_values[int(k)])
return float(sorted_values[f] * (c - k) + sorted_values[c] * (k - f))
def freedman_diaconis_bin_width(values: Sequence[float]) -> float | None:
"""Return Freedman-Diaconis bin width, or None if not usable."""
if len(values) < 2:
return None
sorted_values = sorted(values)
q25 = percentile(sorted_values, 25)
q75 = percentile(sorted_values, 75)
iqr = q75 - q25
if iqr <= 0:
return None
return 2.0 * iqr / (len(values) ** (1.0 / 3.0))
def shared_bins(
distributions: Sequence[Sequence[Metricnumber]],
*,
min_bins: int = 8,
max_bins: int = 80,
log_bins: bool = False,
) -> list[float]:
"""
Compute shared histogram bin edges for multiple distributions.
If log_bins=True, all values must be strictly positive.
"""
all_values: list[float] = [
float(x)
for distribution in distributions
for x in distribution
]
if not all_values:
raise ValueError("no values")
if min_bins < 1:
raise ValueError("min_bins must be >= 1")
if max_bins < min_bins:
raise ValueError("max_bins must be >= min_bins")
lo = min(all_values)
hi = max(all_values)
if log_bins and lo <= 0:
raise ValueError("logarithmic bins require all values > 0")
if lo == hi:
if log_bins:
return [lo / 2.0, lo * 2.0]
return [lo - 0.5, hi + 0.5]
width = freedman_diaconis_bin_width(all_values)
if width is None or width <= 0:
nbins = ceil(log10(len(all_values)) * 3.322 + 1.0)
else:
nbins = ceil((hi - lo) / width)
nbins = max(min_bins, min(max_bins, nbins))
if log_bins:
log_lo = log(lo)
log_hi = log(hi)
step = (log_hi - log_lo) / nbins
return [
exp(log_lo + i * step)
for i in range(nbins + 1)
]
step = (hi - lo) / nbins
return [
lo + i * step
for i in range(nbins + 1)
]
def histogram_counts(
values: Iterable[Metricnumber],
edges: Sequence[float],
) -> list[int]:
"""Compute histogram counts for the given bin edges."""
if len(edges) < 2:
raise ValueError("at least two bin edges are required")
if any(edges[i] >= edges[i + 1] for i in range(len(edges) - 1)):
raise ValueError("bin edges must be strictly increasing")
counts = [0] * (len(edges) - 1)
for value in values:
x = float(value)
if x < edges[0] or x > edges[-1]:
continue
if x == edges[-1]:
counts[-1] += 1
continue
bin_index = bisect_right(edges, x) - 1
counts[bin_index] += 1
return counts
def histogram_table(
distributions: Sequence[Sequence[Metricnumber]],
*,
min_bins: int = 8,
max_bins: int = 80,
log_bins: bool = False,
) -> tuple[list[float], list[list[int]]]:
"""
Return shared bin edges and one histogram count list per distribution.
"""
edges = shared_bins(
distributions,
min_bins=min_bins,
max_bins=max_bins,
log_bins=log_bins,
)
counts = [
histogram_counts(distribution, edges)
for distribution in distributions
]
return edges, counts
def histo_graphs(out_rows: list[OutputRow], colnames: list[str]):
for metric in colnames:
tmpfile = tempfile.NamedTemporaryFile(mode='w+t', delete=False,
dir='.',prefix=f"tmp-seff-{metric}",
suffix=".data")
# create bins
edges, counts = histogram_table([getattr(row,"vectors")[metric] for row in out_rows])
# currently we name the aggregations with numbers
names = [str(x) for x in range (0, len(out_rows))]
tmpfile.write(f"# bin_center {' '.join(names)}\n")
for i in range(len(edges) - 1):
left = edges[i]
right = edges[i + 1]
center = (left + right) / 2.0
# TODO: for log-bins: center = (left * right) ** 0.5
tmpfile.write(f"{center:.6g}")
for colnum in range(0,len(names)):
tmpfile.write(f" {counts[colnum][i]}")
tmpfile.write("\n")
tmpfile.close()
gnuplot_cmd = f"""set title "{metric} distribution" noenhanced
set xlabel "{metric}" noenhanced
set ylabel "Jobs"
set style fill transparent solid 0.35 noborder
set encoding utf8
# set terminal dumb size 120, 30
set terminal dumb enhanced size 120, 30
set colorsequence classic
set terminal dumb ansi size 120, 30
plot_cmd = 'plot "{tmpfile.name}" \\
"""
# gnuplot_cmd += f' using 1:2 with boxes title "{names[0]}" \\\n'
# for col in range(1,len(names)):
# gnuplot_cmd += \
# f', "" using 1:{col+2} with boxes title "{names[col]}" \\\n'
gnuplot_cmd += f' using 1:2 with steps lw 2 title "{names[0]}" \\\n'
for col in range(1,len(names)):
gnuplot_cmd += \
f', "" using 1:{col+2} with steps lw 2 title "{names[col]}" \\\n'
gnuplot_cmd += "'\neval plot_cmd\n"
# print(f"DEBUG: GNUPLOT_SCRIPT:\n{gnuplot_cmd}")
try:
run(["gnuplot"],
input=gnuplot_cmd,
text=True,
check=True,)
except FileNotFoundError:
sys.stderr.write("ERROR: Cannot execute gnuplot for histogram creation. Is it installed?\n")
sys.exit(1)
os.remove(tmpfile.name)
# HISTOGRAMMING CODE END
def parse_args(argv: list[str]) -> argparse.Namespace:
p = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description="Display seff-style CPU, memory and walltime efficiency values from sacct data.",
epilog="""-S/-E are ignored when reading data from a cached file.
Examples:
# first get an overview (-U/--aggr-user) and write a cachefile
slurm-eff-tool -B sacct.cache -U
slurm-eff-tool.py -B sacct.cache --start 2026-05-01 --end 2026-05-22 -U
slurm-eff-tool.py -B sacct.cache --start 2026-05-01 --end now -U
# now you can read the cachefile for later runs and e.g. sort based on waste_Mem
slurm-eff-tool -L sacct.cache -U -s=-waste_mem
# only list a specific user's summary lines
slurm-eff-tool.py -L sacct.cache -U -u dfeich
# list that user's single jobs
slurm-eff-tool.py -L sacct.cache -u dfeich
# supports sorting, also by multiple keys
slurm-eff-tool.py -L sacct.cache --aggr-user -s-waste_total
slurm-eff-tool.py -L sacct.cache --aggr-user -s cpu,-mem,time
# you can cluster jobs by multiple regexps applying to the job names
slurm-eff-tool.py -L sacct.cache -u dfeich -R '^vasp','^gromacs'
# supports flexibel output formatting by defining columns to print
slurm-eff-tool.py -L sacct.cache -o username,Y
# only print rows that evaluate to true based on arithmetic expressions
slurm-eff-tool.py -L sacct.cache -U --expr "(waste_Mem > 2000 and Mem_Eff < 20) and MaxRSS_max/AllocMem < 0.5"
slurm-eff-tool.py -L sacct.cache -U --expr 'username=="feichtinger"'
slurm-eff-tool.py -L sacct.cache -U --expr 'partition in ["standard","short"]'
"""
)
p.add_argument("--avg-function", choices=["median","mean"], default="median",
help="averaging function to use for reporting aggregate rows",)
p.add_argument("-S", "--start", help="sacct start time, passed to sacct -S",
default="now - 24 hours")
p.add_argument("-E", "--end", help="sacct end time, passed to sacct -E",
default="now")
p.add_argument("-u", "--user", help="comma separated list of users (affects sacct DB search or filters later from cache file)")
p.add_argument("-U", "--aggr-user", action="store_true", help="aggregate jobs by user, CPUs, nodes, ReqMem, and timelimit")
p.add_argument(
"-R",
"--aggr-regexp",
action="append",
help="aggregate jobs based on a regexp applied to the jobnames. Option may be repeated multiple times.",
)
p.add_argument("--state", "--job-state", dest="state",
default=None,
help="state filter passed to sacct, e.g. COMPLETED,FAILED,TIMEOUT")
p.add_argument("-B", "--write-binary-cache", help="write a binary cache file in gzipped msgpack format")
p.add_argument("-L", "--load-binary-cache", help="load a binary cache file in gzipped msgpack format")
p.add_argument("-O", "--output-raw", help="write raw sacct output cache to this file (text format, large).")
p.add_argument("-F", "--from-raw", help="read raw sacct output cache from this file.")
p.add_argument("-i", "--info", help="show metadata information for the given binary cache file")
p.add_argument("-H", "--histo", help=f"print histogram of the given metric (Supported columns: {', '.join(HISTO_COLUMNS)}).", default=None)
p.add_argument("--dflt-mpcpu", help=f"Default memory/CPU ratio of cluster [{default_mempercpu_gb} GB/cpu]. Used in memory waste calculation.")
p.add_argument("--sdev", action="store_true", help="after each efficiency average, add sdev, max, and min columns")
p.add_argument("--json", action="store_true", help="emit JSON instead of an ASCII table")
p.add_argument("-s", "--sort", help="comma-separated numeric sort columns or short aliases; prefix with - for descending.")
fmthelp = ", ".join([f"{k}:{ALIASES[k]}" for k in ALIASES])
p.add_argument(
"-o",
"--format",
help= f"String of comma separated column names or short aliases defining the output format [{fmthelp}]",
)
p.add_argument("--expr", help="filter using an arithmetic expression using column names as variables (case insensitive). Don't use for state (use --state flag instead).",
default=None)
p.add_argument("-p", "--preset", help=f"use one of several preset output column formats [{','.join(PRESET_COLUMNS.keys())}]",
default=None)
args = p.parse_args(argv)
if args.aggr_user and args.aggr_regexp:
die("choose only one aggregation mode: --aggr-user or --aggr-regexp")
return args
def main(argv: list[str] | None = None) -> int:
# die silently if pipe process dies before us
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
args = parse_args(argv or sys.argv[1:])
if args.info:
show_cache_info(args.info)
sys.exit(0)
if args.write_binary_cache and os.path.exists(args.write_binary_cache):
sys.stderr.write(f"WARNING, file exists: {args.write_binary_cache}. Balking out...\n")
sys.exit(1)
if args.output_raw and os.path.exists(args.output_raw):
sys.stderr.write(f"WARNING, file exists: {args.output_raw}. Balking out...\n")
sys.exit(1)
output_columns = PRESET_COLUMNS["default"]
if args.preset:
try:
output_columns = PRESET_COLUMNS[args.preset]
except KeyError:
sys.stderr.write(f"ERROR: no such preset format ({args.preset})\n")
sys.exit(1)
if args.format:
output_columns = columns_from_fmtstr(args.format)
if args.aggr_user or args.aggr_regexp:
output_columns = [c for c in output_columns if c != "JobID"]
else:
output_columns = [c for c in output_columns if c != "Count"]
# CREATE RECORDS FROM SACCT QUERY OR FILE, OR LOAD PROCESSED RECORDS FROM BINARY CACHE
if args.load_binary_cache:
cache = read_binary_cache(args.load_binary_cache)
records = cache['records']
else:
if args.from_raw:
rows = read_cache_raw(args.from_raw)
else:
rows = run_sacct(args)
if args.output_raw:
write_cache_raw(args.output_raw, rows)
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)
# Aggregate
out_rows = aggregate_records(records, args)
# filter rows
if args.expr:
expr = SafeExpression(args.expr)
out_rows = [row for row in out_rows \
if expr.evaluate(row.as_dict())]
out_rows = sort_rows(out_rows, args.sort)
# Output table
if args.json:
print(json.dumps([r.as_dict(sdev=args.sdev) for r in out_rows], indent=2))
else:
print_table(out_rows, output_columns, args.sdev)
# Histogramming
if args.histo:
histo_graphs(out_rows, args.histo.lower().split(','))
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink