Mathematically bind the cardinality over thr 12 h window.

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`).
 

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]
 

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 = (