[IMP] monitoring_prometheus: new gauge registry on each /metrics call

Shared gauges mixed data between parallel requests and between databases.
This also avoids cleaning up stale series.

monitoring_prometheus/controllers/prometheus_metrics.py

@@ -1,52 +1,37 @@
 # Copyright 2016 Camptocamp SA
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
 
-import json
 import logging
 
-from prometheus_client import Gauge, generate_latest
+from prometheus_client import REGISTRY, Gauge, generate_latest
 
 from odoo.http import Controller, request, route
 
 _logger = logging.getLogger(__name__)
 
+
 # Gauges must be instantiated only once per process, registering the same
 # metric name twice raises a "Duplicated timeseries" error.
 _GAUGES = {}
 
 
-def _get_gauge(name, documentation, label_names):
+def _get_gauge(name, documentation, label_names, registry=REGISTRY):
     gauge = _GAUGES.get(name)
     if gauge is None:
-        gauge = Gauge(name, documentation, label_names)
+        gauge = Gauge(name, documentation, label_names, registry=registry)
         _GAUGES[name] = gauge
     return gauge
 
 
 class PrometheusController(Controller):
-    def _publish_gathered_metrics(self):
-        records = request.env["prometheus.metric"].sudo().search([])
-        if not records:
-            return
-
-        cleared = set()
-        for record in records:
-            name = record.name
-            labels = json.loads(record.labels or "{}")
-            gauge = _get_gauge(name, record.documentation or "", sorted(labels))
-            if name not in cleared:
-                # drop the series that disappeared since the last collection
-                gauge.clear()
-                cleared.add(name)
-            if labels:
-                gauge.labels(**labels).set(record.value)
-            else:
-                gauge.set(record.value)
-
     @route("/metrics", auth="public")
     def metrics(self):
+        # REGISTRY holds the in-process metrics (request latency, longpolling)
+        output = generate_latest(REGISTRY)
         try:
-            self._publish_gathered_metrics()
+            records = request.env["prometheus.metric"].sudo().search([])
+            registry = records._build_prometheus_registry()
+            output += generate_latest(registry)
         except Exception:
             _logger.exception("Could not publish the gathered Prometheus metrics")
-        return generate_latest()
+        return output

monitoring_prometheus/models/prometheus_gatherer.py

@@ -41,7 +41,7 @@ class PrometheusGatherer(models.AbstractModel):
         Metric = self.env["prometheus.metric"].sudo()
         touched = self.env["prometheus.metric"]
         for metric in self._gather_metrics():
-            touched |= Metric._update_metric(metric)
+            touched |= Metric._create_or_update_metric(metric)
         # drop the series that disappeared since the last collection
         (Metric.search([]) - touched).unlink()
         return True

monitoring_prometheus/models/prometheus_metric.py

@@ -3,6 +3,8 @@
 
 import json
 
+from prometheus_client import CollectorRegistry, Gauge
+
 from odoo import api, fields, models
 
 
@@ -21,29 +23,78 @@ class PrometheusMetric(models.Model):
     documentation = fields.Char()
     # JSON-serialized dict, used both to publish the metric and as part of
     # the uniqueness key alongside ``name``.
-    labels = fields.Char(default="{}")
+    _labels = fields.Char(default="{}")
     value = fields.Float(required=True)
 
+    _sql_constraints = [
+        (
+            "name_labels_uniq",
+            "unique(name, _labels)",
+            "A metric can only be stored once per name and labels.",
+        )
+    ]
+
     @staticmethod
-    def _labels_key(labels):
+    def _serialize_labels(labels):
         return json.dumps(labels or {}, sort_keys=True)
 
+    def labels(self):
+        return json.loads(self._labels or "{}")
+
+    def _check_upsert_context(self):
+        # forces every writer through ``_create_or_update_metric`` (the upsert)
+        assert self.env.context.get("prometheus_metric_upsert"), (
+            "prometheus.metric records must be written through "
+            "_create_or_update_metric()"
+        )
+
+    @api.model_create_multi
+    def create(self, vals_list):
+        self._check_upsert_context()
+        return super().create(vals_list)
+
+    def write(self, vals):
+        self._check_upsert_context()
+        return super().write(vals)
+
     @api.model
-    def _update_metric(self, metric):
+    def _create_or_update_metric(self, metric):
         """Create or update the record matching ``metric``, return it."""
-        labels_key = self._labels_key(metric.get("labels"))
+        labels_key = self._serialize_labels(metric.get("labels"))
         record = self.search(
-            [("name", "=", metric["name"]), ("labels", "=", labels_key)],
+            [("name", "=", metric["name"]), ("_labels", "=", labels_key)],
             limit=1,
         )
         values = {
             "name": metric["name"],
             "documentation": metric.get("documentation", ""),
-            "labels": labels_key,
+            "_labels": labels_key,
             "value": metric["value"],
         }
         if record:
-            record.write(values)
+            record.with_context(prometheus_metric_upsert=True).write(values)
         else:
-            record = self.create(values)
+            record = self.with_context(prometheus_metric_upsert=True).create(values)
         return record
+
+    def _group_metrics(self):
+        # group the records by metric definition so each gauge is created once
+        groups = {}
+        for record in self:
+            labels = record.labels()
+            label_names = tuple(sorted(labels))
+            key = (record.name, record.documentation or "", label_names)
+            groups.setdefault(key, []).append((labels, record.value))
+        return groups
+
+    def _build_prometheus_registry(self):
+        groups = self._group_metrics()
+        registry = CollectorRegistry()
+        for (name, documentation, label_names), values in groups.items():
+            gauge = Gauge(name, documentation, label_names, registry=registry)
+            for labels, value in values:
+                if labels:
+                    gauge.labels(**labels).set(value)
+                else:
+                    gauge.set(value)
+        return registry