From 25653be2922e210c38085374554c1f7ffbb25e13 Mon Sep 17 00:00:00 2001
From: Romuald Brunet <romuald@gandi.net>
Date: Thu, 7 May 2026 18:10:29 +0200
Subject: [PATCH] [WIP] Implement "aggregator" collector

XXX need a better name

It will collect metrics from db files like the MultiProcessCollector,
but additionally it will

- collect metrics from counter / histograms from PIDs that are no longer
  alive
- store them into a dedicated .db file (PID 0)
- remove those .db files

This is done to avoid leaving thousands of unused files for old PIDs
when a long running parent process spawn thousands of processes over
multiple days / weeks
---
 src/gandi_pyramid_prometheus/aggregator.py | 109 +++++++++++++++++++++
 src/gandi_pyramid_prometheus/view.py       |   5 +-
 2 files changed, 111 insertions(+), 3 deletions(-)
 create mode 100644 src/gandi_pyramid_prometheus/aggregator.py

diff --git a/src/gandi_pyramid_prometheus/aggregator.py b/src/gandi_pyramid_prometheus/aggregator.py
new file mode 100644
index 0000000..3fe69ac
--- /dev/null
+++ b/src/gandi_pyramid_prometheus/aggregator.py
@@ -0,0 +1,109 @@
+import errno
+import glob
+import inspect
+import os
+import re
+from collections import defaultdict
+from pathlib import Path
+
+from prometheus_client.multiprocess import MmapedDict, MultiProcessCollector
+
+GET_PID = re.compile(r"_([0-9]+)\.db$")
+
+# Compat prometheus 0.17; 0.18+
+_WRITE_KW = {}
+if "timestamp" in inspect.signature(MmapedDict.write_value).parameters:
+    # 0.18+ needs timestamp argument
+    _WRITE_KW["timestamp"] = 0.0
+
+
+class AggregatorCollector(MultiProcessCollector):
+    def collect(self):
+        files = glob.glob(os.path.join(self._path, "*.db"))
+
+        try:
+            # recent version can use merge() (one less syscall)
+            if hasattr(self, "merge"):
+                return self.merge(files, accumulate=True)
+            return super().collect()
+        finally:
+            # XXX lockfile
+            collect_and_delete_inactive_files(self._path, files)
+
+
+def is_inactive(filename):
+    match = GET_PID.search(filename)
+    pid = int(match.group(1)) if match else None
+    if not pid:  # 0 (or no match)
+        return False
+
+    try:
+        os.kill(pid, 0)
+    except OSError as err:
+        if err.errno == errno.ESRCH:
+            return True
+    return False
+
+
+def collect_and_delete_inactive_files(dirname, files):
+    """
+    Collect metrics for inactive PID to store them in a "PID 0" specific file,
+    then remove them from disk
+
+    """
+    dest_root = Path(dirname)
+
+    inactive = [f for f in files if is_inactive(f)]
+
+    todo = [Path(x) for x in inactive]
+
+    histogram_dict = defaultdict(float)
+    counter_dict = defaultdict(float)
+
+    # Collect counter / histogram values from now dead processes metrics
+    for path in todo:
+        target = None
+
+        if path.name.startswith("counter_"):
+            target = counter_dict
+        elif path.name.startswith("histogram_"):
+            target = histogram_dict
+        elif path.name.startswith("gauge_"):
+            # drop gauges
+            target = None
+        else:
+            print(f"Unknown metric prefix: {path.name}")
+
+        if target is not None:
+            # print(f"{path} is inactive - collect and remove it")
+            mm_values = MmapedDict.read_all_values_from_file(path)
+            for key, value, _ts, *_pos in mm_values:
+                # _pos is the handle position (returned after version 0.18)
+                target[key] += value
+        else:
+            print(f"{path} is inactive - remove it")
+            pass
+
+        try:
+            path.unlink()
+        except Exception:
+            pass
+
+    # Increase our previously stored metrics with the values with collected just now
+    histogram_dump = dest_root / "histogram_0.db"
+    counter_dump = dest_root / "counter_0.db"
+
+    out_map = {
+        histogram_dump: histogram_dict,
+        counter_dump: counter_dict,
+    }
+
+    for destination, source in out_map.items():
+        out_file = MmapedDict(destination)
+        for key, value in source.items():
+            try:
+                ovalue, _ = out_file.read_value(key)
+            except Exception:
+                ovalue = 0
+
+            out_file.write_value(key, value + ovalue, **_WRITE_KW)
diff --git a/src/gandi_pyramid_prometheus/view.py b/src/gandi_pyramid_prometheus/view.py
index 19444f4..a8b3f0c 100644
--- a/src/gandi_pyramid_prometheus/view.py
+++ b/src/gandi_pyramid_prometheus/view.py
@@ -6,18 +6,17 @@
     CollectorRegistry,
     generate_latest,
 )
-from prometheus_client.multiprocess import MultiProcessCollector
 from pyramid.response import Response
 
 from . import prometheus as prom
+from .aggregator import AggregatorCollector
 
 
 def get_metrics(request):
     """Pyramid view that return the metrics"""
-
     if prom.IS_MULTIPROC:
         registry = CollectorRegistry()
-        MultiProcessCollector(registry)
+        AggregatorCollector(registry)
     else:
         registry = REGISTRY