diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index bfc8984862..f468b772be 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -2,6 +2,8 @@ # 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 @@ -9,6 +11,7 @@ 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,7 +516,7 @@ 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): @@ -465,6 +524,7 @@ def remove(self, job): 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) diff --git a/queue_job/jobrunner/metrics.py b/queue_job/jobrunner/metrics.py new file mode 100644 index 0000000000..f52020699f --- /dev/null +++ b/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 diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 6ff2e2f260..0fb1e54450 100644 --- a/queue_job/jobrunner/runner.py +++ b/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")