Expose jobrunner metrics

queue_job/jobrunner/channels.py

@@ -2,13 +2,16 @@
 # Copyright 2015-2016 Camptocamp SA
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 import logging
+import math
+import weakref
 from collections import namedtuple
 from functools import total_ordering
 from heapq import heappop, heappush
 from weakref import WeakValueDictionary
 
 from ..exception import ChannelNotFound
 from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES
+from . import metrics
 
 NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED)
 JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq")
@@ -409,14 +412,70 @@ def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
         self.parent = parent
         if self.parent:
             self.parent.children[name] = self
+            self.parent._register_channel_gauges()
         self.children = {}
         self._queue = ChannelQueue()
         self._running = set()
         self._failed = set()
+        self._waiting_dependencies = set()
         self._pause_until = 0  # utc seconds since the epoch
         self.capacity = capacity
         self.throttle = throttle  # seconds
         self.sequential = sequential
+        self._metrics_labels = None
+        self._register_channel_gauges()
+
+    def __del__(self):
+        self._unregister_channel_gauges()
+
+    def _register_channel_gauges(self) -> None:
+        self._unregister_channel_gauges()
+        self._metrics_labels = {
+            "channel": self.fullname,
+            "root": not bool(self.parent),
+            "leaf": not bool(self.children),
+        }
+        metrics.channel_capacity.labels(**self._metrics_labels).set_function(
+            weakref.proxy(self)._capacity_gauge
+        )
+        metrics.channel_pending.labels(**self._metrics_labels).set_function(
+            weakref.proxy(self)._pending_gauge
+        )
+        metrics.channel_running.labels(**self._metrics_labels).set_function(
+            weakref.proxy(self)._running_gauge
+        )
+        metrics.channel_failed.labels(**self._metrics_labels).set_function(
+            weakref.proxy(self)._failed_gauge
+        )
+        metrics.channel_waiting_dependencies.labels(
+            **self._metrics_labels
+        ).set_function(weakref.proxy(self)._waiting_dependencies_gauge)
+
+    def _unregister_channel_gauges(self) -> None:
+        if not self._metrics_labels:
+            return
+        metrics.channel_capacity.remove_by_labels(self._metrics_labels)
+        metrics.channel_pending.remove_by_labels(self._metrics_labels)
+        metrics.channel_running.remove_by_labels(self._metrics_labels)
+        metrics.channel_failed.remove_by_labels(self._metrics_labels)
+        metrics.channel_waiting_dependencies.remove_by_labels(self._metrics_labels)
+
+    def _capacity_gauge(self) -> float:
+        if self.capacity is None:
+            return math.inf
+        return self.capacity
+
+    def _pending_gauge(self) -> float:
+        return len(self._queue)
+
+    def _running_gauge(self) -> float:
+        return len(self._running)
+
+    def _failed_gauge(self) -> float:
+        return len(self._failed)
+
+    def _waiting_dependencies_gauge(self) -> float:
+        return len(self._waiting_dependencies)
 
     @property
     def sequential(self):
@@ -457,14 +516,15 @@ def __str__(self):
         capacity = "∞" if self.capacity is None else str(self.capacity)
         return (
             f"{self.fullname}(C:{capacity},Q:{len(self._queue)},"
-            f"R:{len(self._running)},F:{len(self._failed)})"
+            f"R:{len(self._running)},F:{len(self._failed)},W:{len(self._waiting_dependencies)})"
         )
 
     def remove(self, job):
         """Remove a job from the channel."""
         self._queue.remove(job)
         self._running.discard(job)
         self._failed.discard(job)
+        self._waiting_dependencies.discard(job)
         if self.parent:
             self.parent.remove(job)
 
@@ -486,6 +546,7 @@ def set_pending(self, job):
             self._queue.add(job)
             self._running.discard(job)
             self._failed.discard(job)
+            self._waiting_dependencies.discard(job)
             if self.parent:
                 self.parent.remove(job)
             _logger.debug("job %s marked pending in channel %s", job.uuid, self)
@@ -499,6 +560,7 @@ def set_running(self, job):
             self._queue.remove(job)
             self._running.add(job)
             self._failed.discard(job)
+            self._waiting_dependencies.discard(job)
             if self.parent:
                 self.parent.set_running(job)
             _logger.debug("job %s marked running in channel %s", job.uuid, self)
@@ -509,10 +571,23 @@ def set_failed(self, job):
             self._queue.remove(job)
             self._running.discard(job)
             self._failed.add(job)
+            self._waiting_dependencies.discard(job)
             if self.parent:
                 self.parent.remove(job)
             _logger.debug("job %s marked failed in channel %s", job.uuid, self)
 
+    def set_waiting_dependencies(self, job):
+        if job not in self._waiting_dependencies:
+            self._queue.remove(job)
+            self._running.discard(job)
+            self._failed.discard(job)
+            self._waiting_dependencies.add(job)
+            if self.parent:
+                self.parent.remove(job)
+            _logger.debug(
+                "job %s marked waiting dependencies in channel %s", job.uuid, self
+            )
+
     def has_capacity(self):
         if self.sequential and self._failed:
             # a sequential queue blocks on failed jobs
@@ -1056,7 +1131,7 @@ def notify(
             job.channel.set_failed(job)
         elif state == WAIT_DEPENDENCIES:
             # wait until all parent jobs are done
-            pass
+            job.channel.set_waiting_dependencies(job)
         else:
             _logger.error("unexpected state %s for job %s", state, job)
 
@@ -1077,3 +1152,6 @@ def get_jobs_to_run(self, now):
 
     def get_wakeup_time(self):
         return self._root_channel.get_wakeup_time()
+
+    def _jobs_to_do_gauge(self) -> float:
+        return len(self._jobs_by_uuid)

queue_job/jobrunner/metrics.py

@@ -0,0 +1,73 @@
+import socket
+from http.server import HTTPServer
+
+from prometheus_client import CollectorRegistry, Counter, Gauge, MetricsHandler
+
+_registry = CollectorRegistry()
+
+_channel_label_names = ("channel", "root", "leaf")
+channel_capacity = Gauge(
+    "queue_job_channel_capacity",
+    documentation="Channel Capacity",
+    labelnames=_channel_label_names,
+    registry=_registry,
+)
+channel_pending = Gauge(
+    "queue_job_channel_pending",
+    documentation="Pending jobs in channel",
+    labelnames=_channel_label_names,
+    registry=_registry,
+)
+channel_running = Gauge(
+    "queue_job_channel_running",
+    documentation="Running jobs in channel",
+    labelnames=_channel_label_names,
+    registry=_registry,
+)
+channel_failed = Gauge(
+    "queue_job_channel_failed",
+    documentation="Failed jobs in channel",
+    labelnames=_channel_label_names,
+    registry=_registry,
+)
+channel_waiting_dependencies = Gauge(
+    "queue_job_channel_waiting_dependencies",
+    documentation="Jobs waiting for dependencies in channel",
+    labelnames=_channel_label_names,
+    registry=_registry,
+)
+
+jobs_to_do = Gauge(
+    "queue_job_jobs_to_do",
+    documentation=(
+        "Number of jobs waiting to be done (including running and failed jobs)"
+    ),
+    registry=_registry,
+)
+jobs_scheduled_total = Counter(
+    "queue_job_jobs_scheduled_total",
+    documentation=(
+        "Total number of jobs scheduled for execution (asked Odoo to run job)"
+    ),
+    labelnames=("db",),
+    registry=_registry,
+)
+dead_jobs_requeued_total = Counter(
+    "queue_job_dead_jobs_requeued_total",
+    documentation=("Total number of dead jobs requeued"),
+    labelnames=("db",),
+    registry=_registry,
+)
+
+
+def make_metrics_server(bind_addr, port) -> HTTPServer:
+    infos = socket.getaddrinfo(
+        bind_addr,
+        port,
+        type=socket.SOCK_STREAM,
+        flags=socket.AI_PASSIVE,
+    )
+    _, _, _, _, sockaddr = next(iter(infos))
+    server = HTTPServer(sockaddr, MetricsHandler.factory(_registry))
+    server.socket.setblocking(False)
+    return server

queue_job/jobrunner/runner.py

@@ -33,7 +33,7 @@
 import odoo
 from odoo.tools import config
 
-from . import queue_job_config
+from . import metrics, queue_job_config
 from .channels import ENQUEUED, NOT_DONE, ChannelManager
 
 SELECT_TIMEOUT = 60
@@ -85,6 +85,18 @@ def _connection_info_for(db_name):
     return connection_info
 
 
+def _metrics_config() -> tuple[str, int | None]:
+    listen_address = (
+        os.environ.get("ODOO_QUEUE_JOB_METRICS_LISTEN_ADDRESS")
+        or queue_job_config.get("metrics_listen_address")
+        or "127.0.0.1"
+    )
+    port = os.environ.get("ODOO_QUEUE_JOB_METRICS_PORT") or queue_job_config.get(
+        "metrics_port"
+    )
+    return listen_address, int(port) if port else None
+
+
 def _async_http_get(scheme, host, port, user, password, db_name, job_uuid):
     # TODO: better way to HTTP GET asynchronously (grequest, ...)?
     #       if this was python3 I would be doing this with
@@ -312,6 +324,7 @@ def requeue_dead_jobs(self):
 
             for (uuid,) in cr.fetchall():
                 _logger.warning("Re-queued dead job with uuid: %s", uuid)
+                metrics.dead_jobs_requeued_total.labels(db=self.db_name).inc()
 
 
 class QueueJobRunner:
@@ -336,6 +349,20 @@ def __init__(
         self.db_by_name = {}
         self._stop = False
         self._stop_pipe = os.pipe()
+        metrics_listen_address, metrics_port = _metrics_config()
+        if metrics_listen_address and metrics_port:
+            self._metrics_server = metrics.make_metrics_server(
+                metrics_listen_address, metrics_port
+            )
+            _logger.info(
+                "jobrunner metrics served on http://%s:%d/metrics",
+                metrics_listen_address,
+                metrics_port,
+            )
+        else:
+            self._metrics_server = None
+            _logger.info("jobrunner metrics server not configured")
+        metrics.jobs_to_do.set_function(self.channel_manager._jobs_to_do_gauge)
 
     def __del__(self):
         # pylint: disable=except-pass
@@ -419,6 +446,7 @@ def run_jobs(self):
                 break
             _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name)
             self.db_by_name[job.db_name].set_job_enqueued(job.uuid)
+            metrics.jobs_scheduled_total.labels(db=job.db_name).inc()
             _async_http_get(
                 self.scheme,
                 self.host,
@@ -458,6 +486,8 @@ def wait_notification(self):
         # we'll select() on database connections and the stop pipe
         conns = [db.conn for db in self.db_by_name.values()]
         conns.append(self._stop_pipe[0])
+        if self._metrics_server:
+            conns.append(self._metrics_server.fileno())
         # look if the channels specify a wakeup time
         wakeup_time = self.channel_manager.get_wakeup_time()
         if not wakeup_time:
@@ -484,7 +514,18 @@ def wait_notification(self):
                         if key.fileobj == self._stop_pipe[0]:
                             # stop-pipe is not a conn so doesn't need poll()
                             continue
-                        key.fileobj.poll()
+                        elif (
+                            self._metrics_server
+                            and key.fileobj == self._metrics_server.fileno()
+                        ):
+                            # TODO: Room for improvement? handle_request does a
+                            # select() which is redundant here because we know
+                            # the socket is ready. _handle_request_noblock()
+                            # seems better suited but is not public.
+                            self._metrics_server.handle_request()
+                        else:
+                            # db conn
+                            key.fileobj.poll()
 
     def stop(self):
         _logger.info("graceful stop requested")