From 58a34e18f91ed59787410e993789676fc324ff4c Mon Sep 17 00:00:00 2001 From: Aditi Gaur Date: Wed, 24 Jun 2026 14:14:10 -0700 Subject: [PATCH] Mathematically bind the cardinality over thr 12 h window. 18 month retention cannot be bounded. This commit is made under the assumption that this entire thing can be removed ASAP. --- README.md | 2 +- azure-slurm-exporter/exporter/sacct.py | 72 ++++++++++++++++++++----- azure-slurm-exporter/test/test_sacct.py | 60 +++++++++++++++++++++ 3 files changed, 119 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 0e1863e8..ee76c88a 100644 --- a/README.md +++ b/README.md @@ -478,7 +478,7 @@ Log can be found under `/var/log/azslurm-exporter.log` | Metric | Type | Labels | Description | |---|---|---|---| | `sacct_terminal_jobs_total` | Counter | `partition`, `exit_code`, `reason`, `state`| Cumulative count of completed/failed/cancelled jobs | -| `sacct_failed_jobs` | Gauge | `jobid`, `jobname`, `nodelist`, `partition`, `exit_code`, `reason`, `state`, `starttime`, `endtime` | Per-job indicator for all non-completed terminal jobs in the query interval (value is always `1`) | +| `sacct_failed_jobs` | Gauge | `jobid`, `jobname`, `nodelist`, `partition`, `exit_code`, `reason`, `state`, `starttime`, `endtime` | Per-job indicator (value is always `1`) for failed jobs in the query interval. Restricted to genuine failure states (`failed`, `node_fail`, `out_of_memory`, `boot_fail`) and capped at `cardinality_bound` (default 20) series per interval, keeping the highest node-count jobs, to bound metrics-backend cardinality. `starttime`/`endtime` are the query window. | Terminal states tracked: `completed`, `failed`, `cancelled`, `timeout`, `node_fail`, `preempted`, `out_of_memory`, `deadline`, `boot_fail`. Exit codes are mapped to human-readable reasons (e.g. `137:0` → `SIGKILL - Force killed`). diff --git a/azure-slurm-exporter/exporter/sacct.py b/azure-slurm-exporter/exporter/sacct.py index 445b928b..bbb269ae 100644 --- a/azure-slurm-exporter/exporter/sacct.py +++ b/azure-slurm-exporter/exporter/sacct.py @@ -46,6 +46,8 @@ def __init__(self, binary_path="/usr/bin/sacct", interval=300, timeout=120, star self.default_output_fmt = "jobid,jobname,nodelist,nnodes,partition,exitcode,derivedexitcode,state,user,start,submit,end,reason" self.sacct_output = namedtuple("sacct_output", self.default_output_fmt) self.terminal_states = "completed,failed,cancelled,timeout,node_fail,preempted,out_of_memory,deadline,boot_fail" + self.failed_states = "failed,node_fail,out_of_memory,boot_fail" + self.cardinality_bound = 20 self.starttime = starttime if starttime is not None else datetime.now().strftime("%Y-%m-%dT%H:%M:%S") self.endtime = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") self.default_options = ["-P" ,"-n", "-o", self.default_output_fmt, "-s", self.terminal_states, "--allocations" ] @@ -74,14 +76,51 @@ def export_metrics(self) -> List[Union[Counter,Gauge]]: """ return self.cached_output["sacct_metrics"] + def _node_count(self, row) -> int: + """Number of nodes allocated to a job, used to rank failed jobs. Defaults to 0 when unparseable.""" + try: + return int(row.nnodes) + except (ValueError, TypeError): + return 0 + def parse_output(self, stdout) -> List[Union[Counter,Gauge]]: """ - Parse sacct command output and increment terminal jobs metric for each job in the output to describe total number of finished jobs by partition, - exit_code, reason, and state as well as return failed jobs gauge describing each job that failed within the time interval of the query. + Parse sacct output into a terminal jobs counter covering every finished job and a per-job + gauge for failed jobs. + + TODO: ###### MUST REMOVE ASAP ######. + This is a hack to mathematically bound the failed jobs exported. Any job can finish in a + non-COMPLETED state, so the "failed" set is otherwise unbounded. To bound it: + 1) The failed jobs gauge is restricted to genuine failure states (self.failed_states). When more + failed jobs than the bound are present, the jobs allocated the most nodes are kept, since larger + failures consume the most resources and are the most useful to drill into. + 2) The failed jobs exported per interval is capped at self.cardinality_bound. Upper bounds below + assume cardinality_bound = 20 and interval = 5 minutes. + + Each job is one unique series keyed by 3 cardinality-exploding labels: jobid, jobname, nodelist. + The rest (partition, exit_code, reason, state) are low-cardinality. starttime/endtime are the + collector's query-window bounds (one value per interval, shared by all jobs). + NOTE: the original intent of the developer seems to be to emit the job's own start/end timestamps but instead they are emitting + Query window's starttime and endtime. + Doing the correct thing would actually add 2 more exploding labels. We intentionally are choosing to KEEP THIS BUG. + + Series index: <= 20 new series every 5 minutes. A series stays "active" only while ingested in + the last ~12 hours, bounding active series from this gauge to 20 * (12h / 5min) = 2,880 per + 12-hour window. + Inverted label index: <= 3 exploding labels * 20 jobs + 2 unique timestamps (query window timestamps) = 62 new index entries every 5 minutes, + i.e. 62 * 144 = 8928 within the 12-hour active window. + + Both series and their label-index entries are retained in the metrics backend for up to 18 months + with no way to purge. This bounds the 12-hour active-series window, but long-term storage still grows + unbounded, capped only by the rolling 18-month retention. Source: https://learn.microsoft.com/en-us/azure/azure-monitor/fundamentals/service-limits#prometheus-metrics + ## TODO: This entire section NEEDS TO BE GONE BY NEXT RELEASE. """ - sacct_failed_jobs = Gauge("sacct_failed_jobs","Number of failed slurm jobs in an interval", + sacct_failed_jobs = Gauge("sacct_failed_jobs", + f"Per-job indicator (value is always 1) for up to {self.cardinality_bound} failed jobs (ranked by node count) in the query interval", ["jobid","jobname","nodelist","partition", "exit_code","reason","state", "starttime","endtime"], registry=None) + failed_states = set(self.failed_states.split(",")) + failed_rows = [] lines = stdout.decode().strip().splitlines() log.debug(f"Number of jobs:{len(lines)}") lines_iter = (line.split("|") for line in lines) @@ -93,18 +132,23 @@ def parse_output(self, stdout) -> List[Union[Counter,Gauge]]: reason=reason, state=row.state.lower()).inc() - if row.state.lower() != "completed": - sacct_failed_jobs.labels( - jobid=row.jobid, - jobname=row.jobname, - nodelist=row.nodelist, - partition=row.partition, - exit_code=row.exitcode, - reason=reason, - state=row.state.lower(), - starttime=self.starttime, - endtime=self.endtime).set(1) + if row.state.lower() in failed_states: + failed_rows.append(row) + top_failed_rows = sorted(failed_rows, key=self._node_count, reverse=True)[:self.cardinality_bound] + log.debug(f"Number of failed jobs: {len(failed_rows)}, exporting top {len(top_failed_rows)} by node count") + for row in top_failed_rows: + reason = self.SLURM_EXIT_CODE_MAPPING.get(row.exitcode, "") + sacct_failed_jobs.labels( + jobid=row.jobid, + jobname=row.jobname, + nodelist=row.nodelist, + partition=row.partition, + exit_code=row.exitcode, + reason=reason, + state=row.state.lower(), + starttime=self.starttime, + endtime=self.endtime).set(1) return [self.sacct_terminal_jobs, sacct_failed_jobs] diff --git a/azure-slurm-exporter/test/test_sacct.py b/azure-slurm-exporter/test/test_sacct.py index 686a057c..58aeb1e2 100644 --- a/azure-slurm-exporter/test/test_sacct.py +++ b/azure-slurm-exporter/test/test_sacct.py @@ -82,6 +82,66 @@ def test_parse_output_empty_output(self, sacct): for sample in family.samples: assert sample.value == 0.0 + @staticmethod + def _failed_gauge_samples(metrics): + """Collect all samples belonging to the sacct_failed_jobs gauge.""" + return [ + s + for metric in metrics + for family in metric.collect() + if family.name == "sacct_failed_jobs" + for s in family.samples + ] + + def test_failed_gauge_only_includes_failure_states(self, sacct): + """Only genuine failure states are exported per-job; cancelled/timeout are counted but not in the gauge.""" + stdout = ( + b"1|job1|node1|1|batch|0:0|0:0|CANCELLED|u1|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + b"2|job2|node2|1|batch|1:0|0:0|TIMEOUT|u2|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + b"3|job3|node3|1|batch|1:0|0:0|FAILED|u3|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + b"4|job4|node4|1|batch|0:125|0:0|OUT_OF_MEMORY|u4|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + ) + sacct.cached_output["sacct_metrics"] = sacct.parse_output(stdout) + metrics = sacct.export_metrics() + exported_states = {s.labels["state"] for s in self._failed_gauge_samples(metrics)} + assert exported_states == {"failed", "out_of_memory"} + + def test_failed_gauge_capped_at_cardinality_bound(self, sacct): + """No more than cardinality_bound failed jobs are exported in a single interval.""" + sacct.cardinality_bound = 20 + lines = [ + b"%d|job%d|node%d|1|batch|1:0|0:0|FAILED|user|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + % (i, i, i) + for i in range(50) + ] + sacct.cached_output["sacct_metrics"] = sacct.parse_output(b"".join(lines)) + metrics = sacct.export_metrics() + assert len(self._failed_gauge_samples(metrics)) == 20 + + def test_failed_gauge_keeps_top_jobs_by_node_count(self, sacct): + """When over the bound, the failed jobs allocated the most nodes are kept.""" + sacct.cardinality_bound = 2 + stdout = ( + b"1|small|node1|1|batch|1:0|0:0|FAILED|u|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + b"2|big|node[1-10]|10|batch|1:0|0:0|FAILED|u|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + b"3|medium|node[1-5]|5|batch|1:0|0:0|FAILED|u|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + ) + sacct.cached_output["sacct_metrics"] = sacct.parse_output(stdout) + metrics = sacct.export_metrics() + exported_jobids = {s.labels["jobid"] for s in self._failed_gauge_samples(metrics)} + assert exported_jobids == {"2", "3"} + + def test_failed_gauge_uses_query_window_timestamps(self, sacct): + """starttime/endtime are the collector's query window, not the job's start/end (intentional).""" + sacct.starttime = "2024-01-01T00:00:00" + sacct.endtime = "2024-01-01T00:05:00" + stdout = b"1001|job1|node1|1|batch|1:0|0:0|FAILED|user1|2024-01-01T10:00:00|2024-01-01T09:00:00|2024-01-01T11:00:00|None\n" + sacct.cached_output["sacct_metrics"] = sacct.parse_output(stdout) + samples = self._failed_gauge_samples(sacct.export_metrics()) + assert len(samples) == 1 + assert samples[0].labels["starttime"] == "2024-01-01T00:00:00" + assert samples[0].labels["endtime"] == "2024-01-01T00:05:00" + def test_parse_output_mixed_partitions(self, sacct): """Test parse_output correctly counts jobs across partitions.""" stdout = (