diff --git a/queue_job/README.rst b/queue_job/README.rst index 5c17681dba..f8ab6aa63f 100644 --- a/queue_job/README.rst +++ b/queue_job/README.rst @@ -68,10 +68,9 @@ Features: - Views for jobs, jobs are stored in PostgreSQL - Jobrunner: execute the jobs, highly efficient thanks to PostgreSQL's NOTIFY -- Channels: give a capacity for the root channel and its sub-channels - and segregate jobs in them. Allow for instance to restrict heavy jobs - to be executed one at a time while little ones are executed 4 at a - times. +- Channels: give a capacity for the root channel and its subchannels and + segregate jobs in them. Allow for instance to restrict heavy jobs to + be executed one at a time while little ones are executed 4 at a times. - Retries: Ability to retry jobs by raising a type of exception - Retry Pattern: the 3 first tries, retry after 10 seconds, the 5 next tries, retry after 1 minutes, ... @@ -123,22 +122,100 @@ Be sure to have the ``requests`` library. Configuration ============= -- Using environment variables and command line: +There are two ways to configure the job runner: + +Set ``channels`` (or ``ODOO_QUEUE_JOB_CHANNELS``) and every database +shares the same channel tree (we will call it server-wide channels): + +.. code:: ini + + [queue_job] + channels = root:10,root.priority:3,root.slow:1 + +Leave ``channels`` unset and set ``max_capacity`` instead. The job +runner then builds a separate channel tree for each database, based on +the *Job Channels* configured on each database (we will call it +per-database channels). + +.. code:: ini + + [queue_job] + max_capacity = 10 + +``channels`` always has precedence over ``max_capacity``. If +``channels`` is set, the per-database configuration is not used. If +neither ``channels`` nor ``max_capacity`` are set, the default execution +mode is per-database channels with a ``max_capacity`` of 1. + +In the per-database mode, channels are configured from the *Job +Channels* menu (or by XML data, see the Usage) instead of a global +configuration string. + +Alongside ``max_capacity``, a global configuration ``db_max_capacity`` +can be set. It represents the max number of jobs executed at the same +time for a single database (capped by the ``max_capacity`` anyway): + +.. code:: ini + + [queue_job] + max_capacity = 10 + db_max_capacity = 3 # no more than 3 simultaneous jobs per database + +``db_max_capacity`` may be an integer or a pattern such as +``prod_*:20,staging:2,*:5``, where the first match wins. When using a +pattern, unmatched databases can be configured by a global pattern at +the end (``*:n``), otherwise they will use the ``max_capacity``. + +The root channel capacity of a database can still be set independently, +however, will in any case be capped by the global ``max_capacity`` and +``db_max_capacity`` parameters. + +When set to 0, ``max_capacity`` or ``db_max_capacity`` means there is no +jobs executed. + +Editing a channel's capacity, sequential flag, throttle or set it to +pause from the *Job Channels* menu **is applied live on the job +runner**. + +.. note:: + + A new database still needs the jobrunner to be restarted. + +When using the server-wide channels, the configuration is static and +loaded at startup of the jobrunner. + +The execution of channels by the job runner is defined by: + +- ``capacity``: max number of jobs running at once in the channel (``0`` + means no limit of its own, the parent channel and ``max_capacity`` or + ``db_max_capacity`` still apply) +- ``sequential``: jobs run one after the other, and a failed job blocks + the channel (requires a capacity of 1) +- ``throttle``: minimum delay, in seconds, between the start of two jobs +- ``paused``: stop running jobs in this channel and its subchannels + +**Job Runner Configuration Parameters** + +- Using environment variables: - Adjust environment variables (optional): - - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels - configuration. The default is ``root:1`` + - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels for + server-wide channels + - ``ODOO_QUEUE_JOB_MAX_CAPACITY=4``, max number of concurrent jobs + (not used if ``ODOO_QUEUE_JOB_CHANNELS`` is set) for per-database + channels + - ``ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2``, max number of concurrent + jobs per DB (not used if ``ODOO_QUEUE_JOB_CHANNELS`` is set) - ``ODOO_QUEUE_JOB_PORT=8069``, default ``--http-port`` - ``ODOO_QUEUE_JOB_SCHEME=https``, default ``http`` - ``ODOO_QUEUE_JOB_HOST=load-balancer``, default ``--http-interface`` or ``localhost`` if unset - ``ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner``, default empty - ``ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t``, default empty - - Start Odoo with ``--load=web,queue_job`` and ``--workers`` greater - than 1. [1]_ -- Using the Odoo configuration file: +- Using the Odoo configuration file (set either ``channels``, either + ``max_capacity`` and/or ``db_max_capacity``) .. code:: ini @@ -150,12 +227,26 @@ Configuration (...) [queue_job] channels = root:2 + max_capacity = 8 + db_max_capacity = 3 scheme = https host = load-balancer port = 443 http_auth_user = jobrunner http_auth_password = s3cr3t +- Odoo has to be started with ``queue_job`` as server-wide module, + either using the command line option ``--load=web,queue_job``, either + by setting it in the Odoo configuration file, and ``--workers`` + greater than 1. [1]_ + +.. code:: ini + + [options] + (...) + workers = 6 + server_wide_modules = web,queue_job + - Confirm the runner is starting correctly by checking the odoo log file: @@ -166,14 +257,25 @@ Configuration ...INFO...queue_job.jobrunner.runner: queue job runner ready for db ...INFO...queue_job.jobrunner.runner: database connections ready -- Create jobs (eg using ``base_import_async``) and observe they start - immediately and in parallel. +- Create jobs (you can create test jobs by opening + ``https://yourodoourl/queue_job/create_test_job``) and observe they + start immediately and in parallel. - Tip: to enable debug logging for the queue job, use ``--log-handler=odoo.addons.queue_job:DEBUG`` -- Jobs that remain in ``enqueued`` or ``started`` state (because, for - instance, their worker has been killed) will be automatically - re-queued. +**Migrating from server-wide channels to per-database channels** + +As long as ``channels`` (or ``ODOO_QUEUE_JOB_CHANNELS``) is set, the job +runner keeps using the server-wide channels. + +To move to channels per database: + +1. Configure the channels you need on each database, from the *Job + Channels* menu: capacity, sequential, throttle, pause +2. Once this configuration is done, remove the ``channels`` options and + set ``max_capacity`` (and optionally ``db_max_capacity``) instead (or + their corresponding environment variables) +3. Restart the job runner .. [1] It works with the threaded Odoo server too, although this way of diff --git a/queue_job/__manifest__.py b/queue_job/__manifest__.py index 224cf047a9..76a3e8b523 100644 --- a/queue_job/__manifest__.py +++ b/queue_job/__manifest__.py @@ -25,6 +25,7 @@ "assets": { "web.assets_backend": [ "queue_job/static/src/views/**/*", + "queue_job/static/src/channel_tree/**/*", ], }, "installable": True, diff --git a/queue_job/delay.py b/queue_job/delay.py index 60e98dc9d6..7fcef9ff63 100644 --- a/queue_job/delay.py +++ b/queue_job/delay.py @@ -443,6 +443,7 @@ class Delayable: "recordset", "_graph", "_job_method", + "_job_method_name", "_job_args", "_job_kwargs", "_generated_job", @@ -471,6 +472,7 @@ def __init__( self.identity_key = identity_key self._job_method = None + self._job_method_name = "" self._job_args = () self._job_kwargs = {} @@ -484,8 +486,7 @@ def _tail(self): def __repr__(self): return ( - f"Delayable({self.recordset}." - f"{self._job_method.__name__ if self._job_method else ''}" + f"Delayable({self.recordset}.{self._job_method_name}" f"({self._job_args}, {self._job_kwargs}))" ) @@ -549,7 +550,8 @@ def split(self, size, chain=False): identity_key=self.identity_key, ) # Update the __self__ - delayable._job_method = getattr(recordset, self._job_method.__name__) + delayable._job_method = getattr(recordset, self._job_method_name) + delayable._job_method_name = self._job_method_name delayable._job_args = self._job_args delayable._job_kwargs = self._job_kwargs @@ -558,7 +560,7 @@ def split(self, size, chain=False): description = self.description or ( self._job_method.__doc__.splitlines()[0].strip() if self._job_method.__doc__ - else f"{self.recordset._name}.{self._job_method.__name__}" + else f"{self.recordset._name}.{self._job_method_name}" ) for index, delayable in enumerate(delayables): delayable.set( @@ -575,6 +577,7 @@ def _build_job(self): return self._generated_job self._generated_job = Job( self._job_method, + method_name=self._job_method_name, args=self._job_args, kwargs=self._job_kwargs, priority=self.priority, @@ -600,6 +603,10 @@ def __getattr__(self, name): ) recordset_method = getattr(self.recordset, name) self._job_method = recordset_method + # Keep the name that was looked up: the method found under it may + # carry another __name__ (e.g. replaced by auditlog), and the job must + # resolve the same attribute again when it is loaded and run. + self._job_method_name = name return self._store_args def _execute_direct(self): diff --git a/queue_job/job.py b/queue_job/job.py index 254f5ce853..ef8bc9113b 100644 --- a/queue_job/job.py +++ b/queue_job/job.py @@ -311,6 +311,7 @@ def _load_from_db_record(cls, job_db_record): job_ = cls( method, + method_name=method_name, args=args, kwargs=kwargs, priority=stored.priority, @@ -386,6 +387,7 @@ def __init__( description=None, channel=None, identity_key=None, + method_name=None, ): """Create a Job @@ -410,6 +412,9 @@ def __init__( :param identity_key: A hash to uniquely identify a job, or a function that returns this hash (the function takes the job as argument) + :param method_name: name of the method on the recordset. Defaults to + ``func.__name__``; pass it when the method was + looked up under a different name than its own. """ if args is None: args = () @@ -425,7 +430,7 @@ def __init__( raise TypeError("Job accepts only methods of Models") recordset = func.__self__ - self.method_name = func.__name__ + self.method_name = method_name or func.__name__ self.recordset = recordset self.job_config = ( @@ -779,7 +784,7 @@ def description(self): elif self.func.__doc__: return self.func.__doc__.splitlines()[0].strip() else: - return f"{self.model_name}.{self.func.__name__}" + return f"{self.model_name}.{self.method_name}" @property def uuid(self): diff --git a/queue_job/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py index 984ab46e26..5c4b512c4d 100644 --- a/queue_job/jobrunner/__init__.py +++ b/queue_job/jobrunner/__init__.py @@ -29,7 +29,7 @@ queue_job_config = dict(cp["queue_job"]) -from .runner import QueueJobRunner, _channels # noqa: E402 +from .runner import QueueJobRunner, _channels, _max_capacity # noqa: E402 _logger = logging.getLogger(__name__) @@ -96,7 +96,12 @@ def signal_time_expired_handler(self, n, stack): def _is_runner_enabled(): - return not _channels().strip().startswith("root:0") + channel_config = _channels() + if channel_config and channel_config.strip().startswith("root:0"): + return False + elif channel_config: + return True + return _max_capacity() != 0 def _start_runner_thread(server_type): @@ -109,7 +114,8 @@ def _start_runner_thread(server_type): else: _logger.info( "jobrunner thread (in %s) NOT started, " - "because the root channel's capacity is set to 0", + "because the root channel's capacity or the max capacity " + "is set to 0", server_type, ) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index b273b9cf47..a5f6c12ede 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -3,6 +3,7 @@ # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) import logging from collections import namedtuple +from dataclasses import asdict, dataclass from functools import total_ordering from heapq import heappop, heappush from weakref import WeakValueDictionary @@ -10,12 +11,26 @@ from ..exception import ChannelNotFound from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES +RELOAD_PAYLOAD = "reload" NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED) JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq") _logger = logging.getLogger(__name__) +@dataclass +class ChannelConfig: + """Configuration of a channel""" + + name: str + capacity: int = 0 + sequential: bool = False + throttle: int = 0 + paused: bool = False + capacity_default: int = 0 + sequential_default: bool = False + + class PriorityQueue: """A priority queue that supports removing arbitrary objects. @@ -408,7 +423,15 @@ class Channel: """ def __init__( - self, name, parent, capacity=None, sequential=False, throttle=0, paused=False + self, + name, + parent, + capacity=None, + sequential=False, + throttle=0, + paused=False, + capacity_default=None, + sequential_default=False, ): self.name = name self.parent = parent @@ -419,9 +442,13 @@ def __init__( self._running = set() self._failed = set() self._pause_until = 0 # utc seconds since the epoch - self.capacity = capacity + self.capacity = ( + capacity if (capacity is not None) else (parent and parent.capacity_default) + ) + self.capacity_default = capacity_default self.throttle = throttle # seconds - self.sequential = sequential + self.sequential = sequential or (parent and parent.sequential_default) + self.sequential_default = sequential_default self.paused = paused @property @@ -438,13 +465,17 @@ def configure(self, config): Supported keys are: * capacity + * capacity_default (default for sub channels) * sequential + * sequential_default (default for sub channels) * throttle * paused """ assert self.fullname.endswith(config["name"]) self.capacity = config.get("capacity", None) + self.capacity_default = config.get("capacity_default", None) self.sequential = bool(config.get("sequential", False)) + self.sequential_default = config.get("sequential_default", False) self.throttle = int(config.get("throttle", 0)) self.paused = bool(config.get("paused", False)) if self.sequential and self.capacity != 1: @@ -899,22 +930,23 @@ def parse_simple_config(cls, config_string): continue config = {} config_items = split_strip(channel_config_string, ":") - name = config_items[0] - if not name: + if not (name := config_items.pop(0)): raise ValueError( f"Invalid channel config {config_string}: missing channel name" ) config["name"] = name - if len(config_items) > 1: - capacity = config_items[1] + if len(config_items) > 0: try: - config["capacity"] = int(capacity) + config["capacity"] = int(config_items[0]) + config_items.pop(0) except Exception as ex: - raise ValueError( - f"Invalid channel config {config_string}: " - f"invalid capacity {capacity}" - ) from ex - for config_item in config_items[2:]: + if name == "root": + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity {config_items[0]}" + ) from ex + + for config_item in config_items: kv = split_strip(config_item, "=") if len(kv) == 1: k, v = kv[0], True @@ -929,7 +961,16 @@ def parse_simple_config(cls, config_string): raise ValueError( f"Invalid channel config {config_string}: duplicate key {k}" ) - config[k] = v + if k == "capacity_default": + try: + config[k] = int(v) + except Exception as ex: + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity_default {v}" + ) from ex + else: + config[k] = v else: config["capacity"] = 1 res.append(config) @@ -942,6 +983,17 @@ def simple_configure(self, config_string): >>> c = cm.get_channel_by_name('root') >>> c.capacity 1 + + >>> cm.simple_configure('root:bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:bogus: invalid capacity bogus + + >>> cm.simple_configure('root:4,:2') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4,:2: missing channel name + >>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential') >>> cm.get_channel_by_name('root').capacity 4 @@ -958,11 +1010,41 @@ def simple_configure(self, config_string): 1 >>> cm.get_channel_by_name('seq').sequential True - """ + + >>> cm.simple_configure('root:4:capacity_default=bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4:capacity_default=bogus: invalid capacity_default bogus + + >>> cm.simple_configure('root:4,sub:3:capacity_default=2') + >>> cm.get_channel_by_name('root.sub').capacity + 3 + >>> cm.get_channel_by_name('root.sub.auto', autocreate=True).capacity + 2 + + >>> cm.simple_configure('root:4,seq:2:sequential') + Traceback (most recent call last): + ... + ValueError: A sequential channel must have a capacity of 1 + + >>> cm.simple_configure('root:4,seq:sequential_default') + >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential + True + + """ # noqa: E501 for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) - def get_channel_from_config(self, config): + def configure(self, configs, db_name=None): + """Configure the channel manager from list of :class:`ChannelConfig` + + :param db_name: used to show the database name in the + logs when using per-database channels + """ + for config in configs: + self.get_channel_from_config(asdict(config), db_name=db_name) + + def get_channel_from_config(self, config, db_name=None): """Return a Channel object from a parsed configuration. If the channel does not exist it is created. @@ -974,7 +1056,10 @@ def get_channel_from_config(self, config): """ channel = self.get_channel_by_name(config["name"], autocreate=True) channel.configure(config) - _logger.info("Configured channel: %s", channel) + if db_name: + _logger.info("Configured channel: %s (db: %s)", channel, db_name) + else: + _logger.info("Configured channel: %s", channel) return channel def get_channel_by_name( @@ -1113,3 +1198,8 @@ def get_jobs_to_run(self, now): def get_wakeup_time(self): return self._root_channel.get_wakeup_time() + + @property + def running_count(self) -> int: + """Number of jobs currently running""" + return len(self._root_channel._running) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 443ef27b26..3136527647 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -19,11 +19,13 @@ anonymous ``/queue_job/runjob`` HTTP request. """ +import fnmatch import logging import os import selectors import threading import time +from collections import deque from contextlib import closing, contextmanager import psycopg2 @@ -34,7 +36,7 @@ from odoo.tools import config from . import queue_job_config -from .channels import ENQUEUED, NOT_DONE, ChannelManager +from .channels import ENQUEUED, NOT_DONE, RELOAD_PAYLOAD, ChannelConfig, ChannelManager SELECT_TIMEOUT = 60 ERROR_RECOVERY_DELAY = 5 @@ -57,14 +59,93 @@ class MasterElectionLost(Exception): # so we check it in addition to the environment variables. -def _channels(): +def _max_capacity() -> int: + """Maximum number of jobs running at the same time across all databases + + It comes from the ``ODOO_QUEUE_JOB_MAX_CAPACITY`` environment + variable, then ``max_capacity`` in the ``[queue_job]`` section of the + configuration file. + + If none is configured, the max capacity defaults to 1. + An explicit value of 0 means no job is dispatched at all. + """ + value = os.environ.get("ODOO_QUEUE_JOB_MAX_CAPACITY") or queue_job_config.get( + "max_capacity" + ) + if value: + return int(value) + return 1 + + +def _db_max_capacity() -> str: return ( - os.environ.get("ODOO_QUEUE_JOB_CHANNELS") - or queue_job_config.get("channels") - or "root:1" + os.environ.get("ODOO_QUEUE_JOB_DB_MAX_CAPACITY") + or queue_job_config.get("db_max_capacity") + or "" ) +def parse_db_max_capacity(spec): + """Parse a per-database max capacity configuration string + + The string is a comma-separated list of ``pattern:capacity`` items, where + ``pattern`` matches database names with fnmatch wildcards. + + The first matching pattern wins, so specific patterns must be first in the + string. + + A single integer is applied to all databases, as a shorthand for + ``*:capacity``. + + >>> parse_db_max_capacity('prod_*:20,staging:2,*:5') + [('prod_*', 20), ('staging', 2), ('*', 5)] + >>> parse_db_max_capacity('8') + [('*', 8)] + >>> parse_db_max_capacity('') + [] + >>> parse_db_max_capacity(None) + [] + """ + rules = [] + if not spec: + return rules + for item in spec.replace("\n", ",").split(","): + item = item.strip() + if not item: + continue + pattern, sep, capacity = item.rpartition(":") + if not sep: + pattern = "*" + try: + rules.append((pattern.strip(), int(capacity))) + except ValueError as ex: + raise ValueError(f"Invalid db max capacity {spec}: {capacity}") from ex + return rules + + +def db_max_capacity_for(db_name, rules, default=None): + """Max capacity of a database, first match wins + + >>> rules = parse_db_max_capacity('prod_*:20,staging:2,*:5') + >>> db_max_capacity_for('prod_foo', rules) + 20 + >>> db_max_capacity_for('staging', rules) + 2 + >>> db_max_capacity_for('dev', rules) + 5 + >>> db_max_capacity_for('dev', [], default=7) + 7 + """ + for pattern, capacity in rules: + if fnmatch.fnmatch(db_name, pattern): + return capacity + return default + + +def _channels(): + return os.environ.get("ODOO_QUEUE_JOB_CHANNELS") or queue_job_config.get("channels") + + def _odoo_now(): # important: this must return the same as postgresql # EXTRACT(EPOCH FROM TIMESTAMP dt) @@ -125,9 +206,11 @@ def __init__(self, db_name): try: self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) self.has_queue_job = self._has_queue_job() + self.has_channel_config_columns = False if self.has_queue_job: self._acquire_master_lock() self._initialize() + self.has_channel_config_columns = self._has_channel_config_columns() except BaseException: self.close() raise @@ -184,6 +267,64 @@ def _initialize(self): with closing(self.conn.cursor()) as cr: cr.execute("LISTEN queue_job") + def _has_channel_config_columns(self): + with closing(self.conn.cursor()) as cr: + cr.execute( + """ + SELECT count(*) + FROM information_schema.columns + WHERE table_name = %s + AND column_name IN ( + 'capacity', + 'sequential', + 'throttle', + 'paused', + 'capacity_default', + 'sequential_default' + ) + """, + ("queue_job_channel",), + ) + return cr.fetchone()[0] == 6 + + def load_channels_config(self): + """Return the channels configuration stored in the database""" + if not self.has_channel_config_columns: + return None + with closing(self.conn.cursor()) as cr: + cr.execute( + "SELECT complete_name, " + "COALESCE(capacity, 0), " + "COALESCE(sequential, false), " + "COALESCE(throttle, 0), " + "COALESCE(paused, false), " + "COALESCE(capacity_default, 0), " + "COALESCE(sequential_default, false) " + "FROM queue_job_channel " + ) + rows = cr.fetchall() + configs = [ + ChannelConfig( + name=name, + capacity=capacity, + sequential=sequential, + throttle=throttle, + paused=paused, + capacity_default=capacity_default, + sequential_default=sequential_default, + ) + for ( + name, + capacity, + sequential, + throttle, + paused, + capacity_default, + sequential_default, + ) in rows + ] + return configs + @contextmanager def select_jobs(self, where, args): # pylint: disable=sql-injection @@ -323,16 +464,40 @@ def __init__( user=None, password=None, channel_config_string=None, + max_capacity=None, + db_max_capacity=None, ): self.scheme = scheme self.host = host self.port = port self.user = user self.password = password - self.channel_manager = ChannelManager() + if channel_config_string is None: channel_config_string = _channels() - self.channel_manager.simple_configure(channel_config_string) + + self._server_wide_channel_manager = None + if channel_config_string: + channel_manager = ChannelManager() + channel_manager.simple_configure(channel_config_string) + self._server_wide_channel_manager = channel_manager + # max_capacity is always equal to the root channel in server-wide + # configuration + max_capacity = channel_manager.get_channel_by_name("root").capacity + + if max_capacity is None: + max_capacity = _max_capacity() + self.max_capacity = max(0, max_capacity) + + if db_max_capacity is None: + db_max_capacity = _db_max_capacity() + self.db_max_capacity_rules = parse_db_max_capacity(db_max_capacity) + + self._channel_manager_by_db = {} + self._channel_managers = [] + + self._round_robin_offset = 0 + self.db_by_name = {} self._stop = False self._stop_event = threading.Event() @@ -389,11 +554,98 @@ def close_databases(self, remove_jobs=True): for db_name, db in self.db_by_name.items(): try: if remove_jobs: - self.channel_manager.remove_db(db_name) + self._channel_manager_by_db[db_name].remove_db(db_name) db.close() except Exception: _logger.warning("error closing database %s", db_name, exc_info=True) self.db_by_name = {} + self._channel_manager_by_db = {} + self._channel_managers = [] + + @staticmethod + def _create_paused_root_channel_manager(db_name=None): + """Create a channel manager with a single root and paused channel + + It is used when the channels configuration cannot be used because + of a misconfiguration or because a database schema is outdated. + """ + channel_manager = ChannelManager() + channel_manager.configure( + [ChannelConfig(name="root", capacity=0, paused=True)], db_name=db_name + ) + return channel_manager + + def _build_channel_manager(self, db): + """Build and configure the channel manager of a database""" + db_max = max( + 0, + db_max_capacity_for( + db.db_name, self.db_max_capacity_rules, default=self.max_capacity + ), + ) + + channels_config = db.load_channels_config() + if channels_config is None: + # database not updated to the proper schema, + # no job execution until it is properly upgraded + _logger.error( + "database %s schema is outdated, -u queue_job required", db.db_name + ) + return self._create_paused_root_channel_manager(db_name=db.db_name) + + root_config = next( + (config for config in channels_config if config.name == "root"), None + ) + if root_config is None: + root_config = ChannelConfig("root") + channels_config.insert(0, root_config) + + if not db_max: + # if a database is set at 0, it does not run any jobs, pause it + root_config.paused = True + elif not root_config.capacity: + root_config.capacity = db_max + else: + root_config.capacity = min(root_config.capacity, db_max) + + channel_manager = ChannelManager() + try: + channel_manager.configure(channels_config, db_name=db.db_name) + except ValueError: + # A bad channel configuration on a single database + # (e.g. sequential with a capacity != 1) should not + # break other databases, use a paused root channel + # instead. If the configuration is corrected on the DB, + # the next reload should rebuild the channel manager. + _logger.warning( + "invalid channel configuration for db %s, pausing until resolution", + db.db_name, + exc_info=True, + ) + return self._create_paused_root_channel_manager(db_name=db.db_name) + return channel_manager + + def _register_db_server_wide(self, db): + """Register the server-wide channel manager for a DB and reload its jobs""" + db_name = db.db_name + channel_manager = self._server_wide_channel_manager + self._channel_manager_by_db[db_name] = channel_manager + self._channel_managers = [self._server_wide_channel_manager] + self._notify_existing_jobs(db, channel_manager) + + def _configure_db(self, db): + """Build the channel manager for a database and reload its jobs""" + db_name = db.db_name + channel_manager = self._build_channel_manager(db) + self._channel_manager_by_db[db_name] = channel_manager + self._channel_managers = list(self._channel_manager_by_db.values()) + self._notify_existing_jobs(db, channel_manager) + _logger.info("channels configuration loaded for db %s", db_name) + + def _notify_existing_jobs(self, db, channel_manager): + with db.select_jobs("state in %s", (NOT_DONE,)) as cr: + for job_data in cr: + channel_manager.notify(db.db_name, *job_data) def initialize_databases(self): for db_name in sorted(self.get_db_names()): @@ -401,9 +653,10 @@ def initialize_databases(self): db = Database(db_name) if db.has_queue_job: self.db_by_name[db_name] = db - with db.select_jobs("state in %s", (NOT_DONE,)) as cr: - for job_data in cr: - self.channel_manager.notify(db_name, *job_data) + if self._server_wide_channel_manager: + self._register_db_server_wide(db) + else: + self._configure_db(db) _logger.info("queue job runner ready for db %s", db_name) else: db.close() @@ -413,24 +666,71 @@ def requeue_dead_jobs(self): if db.has_queue_job: db.requeue_dead_jobs() - def run_jobs(self): - now = _odoo_now() - for job in self.channel_manager.get_jobs_to_run(now): - if self._stop: - 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) - _async_http_get( - self.scheme, - self.host, - self.port, - self.user, - self.password, - job.db_name, - job.uuid, + def _all_running_count(self) -> int: + return sum( + channel_manager.running_count for channel_manager in self._channel_managers + ) + + def _dispatch_job(self, job): + _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) + _async_http_get( + self.scheme, + self.host, + self.port, + self.user, + self.password, + job.db_name, + job.uuid, + ) + + def _round_robin_jobs(self, channel_managers, now): + managers_count = len(channel_managers) + # Ensure the channel managers are ordered in way that all have + # equal chances to enqueue jobs. The manager that was last last + # time will be first the next time and so on. + job_generators = deque( + ( + # store the actual position of the channel manager in the channel + # managers list + index % managers_count, + channel_managers[index % managers_count].get_jobs_to_run(now), + ) + for index in range( + self._round_robin_offset, self._round_robin_offset + managers_count ) + ) + while job_generators: + index, job_generator = job_generators.popleft() + job = next(job_generator, None) + if job is None: + # generator exhausted for this tick, remove from the deque + # it will come back next time + continue + job_generators.append((index, job_generator)) + self._round_robin_offset = index + 1 + yield job + + def run_jobs(self): + channel_managers = self._channel_managers + if not channel_managers: + return + + jobs = self._round_robin_jobs(channel_managers, _odoo_now()) + while not self._stop: + if self._all_running_count() >= self.max_capacity: + _logger.debug( + "max capacity of %s reached, waiting for capacity", + self.max_capacity, + ) + return + job = next(jobs, None) + if job is None: + return + self._dispatch_job(job) def process_notifications(self): + reload_db_names = set() for db in self.db_by_name.values(): if not db.conn.notifies: # If there are no activity in the queue_job table it seems that @@ -442,13 +742,39 @@ def process_notifications(self): if self._stop: break notification = db.conn.notifies.pop() - uuid = notification.payload + payload = notification.payload + if payload == RELOAD_PAYLOAD and not self._server_wide_channel_manager: + reload_db_names.add(db.db_name) + continue + + uuid = payload + channel_manager = self._channel_manager_by_db[db.db_name] with db.select_jobs("uuid = %s", (uuid,)) as cr: job_datas = cr.fetchone() if job_datas: - self.channel_manager.notify(db.db_name, *job_datas) + channel_manager.notify(db.db_name, *job_datas) else: - self.channel_manager.remove_job(uuid) + channel_manager.remove_job(uuid) + + for db_name in reload_db_names: + self._configure_db(self.db_by_name[db_name]) + + def next_wakeup_time(self): + # A wake-up time of 0 does not mean to wake immediately, but to stop + # until we get new notifications (practically, set a SELECT_TIMEOUT, + # and wait for notifications). Wake-ups of more than 0 are used to + # dispatch jobs with ETA, as these will not notify anything when they + # actually have to start. + # So when getting the next wake-up time, we find the first channel + # manager with ETA jobs (a wake-up time more than 0), and if none, + # a wake-up time of zero is returned to go back to the select waiting + # for notifications. + wakeup_times = [ + wakeup_time + for channel_manager in self._channel_managers + if (wakeup_time := channel_manager.get_wakeup_time()) + ] + return min(wakeup_times, default=0) def wait_notification(self): for db in self.db_by_name.values(): @@ -460,7 +786,7 @@ def wait_notification(self): conns = [db.conn for db in self.db_by_name.values()] conns.append(self._stop_pipe[0]) # look if the channels specify a wakeup time - wakeup_time = self.channel_manager.get_wakeup_time() + wakeup_time = self.next_wakeup_time() if not wakeup_time: # this could very well be no timeout at all, because # any activity in the job queue will wake us up, but diff --git a/queue_job/models/queue_job_channel.py b/queue_job/models/queue_job_channel.py index 0a5c1c917b..b6671e4577 100644 --- a/queue_job/models/queue_job_channel.py +++ b/queue_job/models/queue_job_channel.py @@ -4,12 +4,29 @@ from odoo import api, exceptions, fields, models +from ..jobrunner import runner +from ..jobrunner.channels import RELOAD_PAYLOAD + class QueueJobChannel(models.Model): _name = "queue.job.channel" _description = "Job Channels" _rec_name = "complete_name" + # fields that trigger a reload of the jobrunner for this database when changed + _JOBRUNNER_CONFIG_FIELDS = frozenset( + ( + "capacity", + "sequential", + "throttle", + "paused", + "name", + "parent_id", + "capacity_default", + "sequential_default", + ) + ) + name = fields.Char() complete_name = fields.Char( compute="_compute_complete_name", store=True, readonly=True, recursive=True @@ -23,7 +40,47 @@ class QueueJobChannel(models.Model): string="Job Functions", ) removal_interval = fields.Integer( - default=lambda self: self.env["queue.job"]._removal_interval, required=True + default=lambda self: self.env["queue.job"]._removal_interval, + required=True, + help="Number of days after which done jobs are deleted.", + ) + capacity = fields.Integer( + help="Maximum number of jobs running at the same time in this channel. " + "0 means no limit, but they are still limited by the capacity of the parent " + "channel. On the root channel, 0 is limited by the global server-wide " + "configuration." + ) + sequential = fields.Boolean( + help="Jobs are executed one after the other and failed jobs block the channel. " + "Requires a capacity of 1." + ) + throttle = fields.Integer( + help="Minimum delay in seconds between the start of two jobs in this channel." + ) + paused = fields.Boolean( + help="A paused channel (an its subchannels) do not execute any jobs until " + "resumed." + ) + capacity_default = fields.Integer( + string="Subchannels Default Capacity", + help="Default capacity for unconfigured subchannels. " + "0 means they would have the same capacity as the current channel.", + ) + sequential_default = fields.Boolean( + string="Subchannels Default Sequential", + help="If sequential is enabled for unconfigured subchannels.", + ) + effective_paused = fields.Boolean( + compute="_compute_effective_paused", + recursive=True, + help="If this channel is actually paused, depending on the parents.", + ) + effective_capacity = fields.Integer( + compute="_compute_effective_capacity", + recursive=True, + help="Actual capacity of this channel. Its own capacity if set, " + "or the capacity of the closest parent. The root channel " + "is limited by the server-wide max_capacity/db_max_capacity configuration.", ) _name_uniq = models.Constraint( @@ -31,6 +88,35 @@ class QueueJobChannel(models.Model): "Channel complete name must be unique", ) + @api.constrains( + "capacity", "sequential", "throttle", "capacity_default", "sequential_default" + ) + def _check_jobrunner_configuration(self): + for record in self: + if record.capacity < 0: + raise exceptions.ValidationError( + self.env._("The capacity of a channel cannot be negative.") + ) + if record.throttle < 0: + raise exceptions.ValidationError( + self.env._("The throttle of a channel cannot be negative.") + ) + if record.sequential and record.capacity != 1: + raise exceptions.ValidationError( + self.env._("A sequential channel must have a capacity of 1.") + ) + if record.capacity_default < 0: + raise exceptions.ValidationError( + self.env._("The default capacity of a channel cannot be negative.") + ) + if record.sequential_default and record.capacity_default != 1: + raise exceptions.ValidationError( + self.env._( + "A channel with a sequential default must have a " + "default capacity of 1." + ) + ) + @api.depends("name", "parent_id.complete_name") def _compute_complete_name(self): for record in self: @@ -42,6 +128,32 @@ def _compute_complete_name(self): complete_name = record.name record.complete_name = complete_name + @api.depends("paused", "parent_id.effective_paused") + def _compute_effective_paused(self): + for record in self: + record.effective_paused = record.paused or bool( + record.parent_id and record.parent_id.effective_paused + ) + + @api.depends("capacity", "parent_id.effective_capacity") + def _compute_effective_capacity(self): + for record in self: + if record.parent_id: + max_capacity = record.parent_id.effective_capacity + else: + max_capacity = record._root_max_capacity() + if record.capacity: + record.effective_capacity = min(record.capacity, max_capacity) + else: + record.effective_capacity = max_capacity + + def _root_max_capacity(self): + """Server-wide capacity of the root channel for current database.""" + rules = runner.parse_db_max_capacity(runner._db_max_capacity()) + return runner.db_max_capacity_for( + self.env.cr.dbname, rules, default=runner._max_capacity() + ) + @api.constrains("parent_id", "name") def parent_required(self): for record in self: @@ -72,8 +184,17 @@ def create(self, vals_list): new_vals_list.append(vals) vals_list = new_vals_list records |= super().create(vals_list) + records._notify_channel_config_changed() return records + @api.onchange("capacity") + def _onchange_capacity(self): + self.capacity_default = self.capacity + + @api.onchange("sequential") + def _onchange_sequential(self): + self.sequential_default = self.sequential + def write(self, values): for channel in self: if ( @@ -83,7 +204,10 @@ def write(self, values): ): msg = self.env._("Cannot change the root channel") raise exceptions.UserError(msg) - return super().write(values) + res = super().write(values) + if self._JOBRUNNER_CONFIG_FIELDS.intersection(values): + self._notify_channel_config_changed() + return res @api.ondelete(at_uninstall=False) def _check_not_root_ondelete(self): @@ -91,3 +215,18 @@ def _check_not_root_ondelete(self): if channel.name == "root": msg = self.env._("Cannot remove the root channel") raise exceptions.UserError(msg) + + def unlink(self): + res = super().unlink() + self._notify_channel_config_changed() + return res + + def action_pause(self): + self.write({"paused": True}) + + def action_resume(self): + self.write({"paused": False}) + + def _notify_channel_config_changed(self): + """Notify the jobrunner to reload its configuration""" + self.env.cr.execute("SELECT pg_notify('queue_job', %s)", (RELOAD_PAYLOAD,)) diff --git a/queue_job/readme/CONFIGURE.md b/queue_job/readme/CONFIGURE.md index 7239106218..d2233d54b8 100644 --- a/queue_job/readme/CONFIGURE.md +++ b/queue_job/readme/CONFIGURE.md @@ -1,16 +1,89 @@ -- Using environment variables and command line: +There are two ways to configure the job runner: + +Set `channels` (or `ODOO_QUEUE_JOB_CHANNELS`) and every database shares the +same channel tree (we will call it server-wide channels): + +``` ini +[queue_job] +channels = root:10,root.priority:3,root.slow:1 +``` + +Leave `channels` unset and set `max_capacity` instead. The job runner then +builds a separate channel tree for each database, based on the *Job Channels* +configured on each database (we will call it per-database channels). + +``` ini +[queue_job] +max_capacity = 10 +``` + +`channels` always has precedence over `max_capacity`. If `channels` is set, the +per-database configuration is not used. If neither `channels` nor +`max_capacity` are set, the default execution mode is per-database channels +with a `max_capacity` of 1. + +In the per-database mode, channels are configured from the *Job Channels* menu +(or by XML data, see the Usage) instead of a global configuration string. + +Alongside `max_capacity`, a global configuration `db_max_capacity` can be set. +It represents the max number of jobs executed at the same time for a single +database (capped by the `max_capacity` anyway): + +``` ini +[queue_job] +max_capacity = 10 +db_max_capacity = 3 # no more than 3 simultaneous jobs per database +``` + +`db_max_capacity` may be an integer or a pattern such as +`prod_*:20,staging:2,*:5`, where the first match wins. When using a pattern, +unmatched databases can be configured by a global pattern at the end (`*:n`), +otherwise they will use the `max_capacity`. + +The root channel capacity of a database can still be set independently, +however, will in any case be capped by the global `max_capacity` and +`db_max_capacity` parameters. + +When set to 0, `max_capacity` or `db_max_capacity` means there is no jobs executed. + +Editing a channel's capacity, sequential flag, throttle or set it to pause from +the *Job Channels* menu **is applied live on the job runner**. + +> [!NOTE] +> A new database still needs the jobrunner to be restarted. + +When using the server-wide channels, the configuration is static and loaded at +startup of the jobrunner. + +The execution of channels by the job runner is defined by: + +- `capacity`: max number of jobs running at once in the channel (`0` means no + limit of its own, the parent channel and `max_capacity` or `db_max_capacity` + still apply) +- `sequential`: jobs run one after the other, and a failed job blocks the + channel (requires a capacity of 1) +- `throttle`: minimum delay, in seconds, between the start of two jobs +- `paused`: stop running jobs in this channel and its subchannels + + +**Job Runner Configuration Parameters** + +- Using environment variables: - Adjust environment variables (optional): - - `ODOO_QUEUE_JOB_CHANNELS=root:4` or any other channels - configuration. The default is `root:1` + - `ODOO_QUEUE_JOB_CHANNELS=root:4` or any other channels for server-wide + channels + - `ODOO_QUEUE_JOB_MAX_CAPACITY=4`, max number of concurrent jobs (not used + if `ODOO_QUEUE_JOB_CHANNELS` is set) for per-database channels + - `ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2`, max number of concurrent jobs per DB + (not used if `ODOO_QUEUE_JOB_CHANNELS` is set) - `ODOO_QUEUE_JOB_PORT=8069`, default `--http-port` - `ODOO_QUEUE_JOB_SCHEME=https`, default `http` - `ODOO_QUEUE_JOB_HOST=load-balancer`, default `--http-interface` or `localhost` if unset - `ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner`, default empty - `ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t`, default empty - - Start Odoo with `--load=web,queue_job` and `--workers` greater than - 1.[^1] -- Using the Odoo configuration file: +- Using the Odoo configuration file (set either `channels`, either + `max_capacity` and/or `db_max_capacity`) ``` ini [options] @@ -21,6 +94,8 @@ server_wide_modules = web,queue_job (...) [queue_job] channels = root:2 +max_capacity = 8 +db_max_capacity = 3 scheme = https host = load-balancer port = 443 @@ -28,17 +103,29 @@ http_auth_user = jobrunner http_auth_password = s3cr3t ``` +- Odoo has to be started with `queue_job` as server-wide module, either using + the command line option `--load=web,queue_job`, either by setting it in the + Odoo configuration file, and `--workers` greater than 1.[^1] + +``` ini +[options] +(...) +workers = 6 +server_wide_modules = web,queue_job +``` + - Confirm the runner is starting correctly by checking the odoo log file: -``` +``` ...INFO...queue_job.jobrunner.runner: starting ...INFO...queue_job.jobrunner.runner: initializing database connections ...INFO...queue_job.jobrunner.runner: queue job runner ready for db ...INFO...queue_job.jobrunner.runner: database connections ready ``` -- Create jobs (eg using `base_import_async`) and observe they start +- Create jobs (you can create test jobs by opening + `https://yourodoourl/queue_job/create_test_job`) and observe they start immediately and in parallel. - Tip: to enable debug logging for the queue job, use `--log-handler=odoo.addons.queue_job:DEBUG` @@ -46,5 +133,17 @@ http_auth_password = s3cr3t [^1]: It works with the threaded Odoo server too, although this way of running Odoo is obviously not for production purposes. -* Jobs that remain in `enqueued` or `started` state (because, for instance, - their worker has been killed) will be automatically re-queued. + +**Migrating from server-wide channels to per-database channels** + +As long as `channels` (or `ODOO_QUEUE_JOB_CHANNELS`) is set, the job +runner keeps using the server-wide channels. + +To move to channels per database: + +1. Configure the channels you need on each database, from the *Job + Channels* menu: capacity, sequential, throttle, pause +2. Once this configuration is done, remove the `channels` options and set + `max_capacity` (and optionally `db_max_capacity`) instead (or their + corresponding environment variables) +3. Restart the job runner diff --git a/queue_job/readme/DESCRIPTION.md b/queue_job/readme/DESCRIPTION.md index 8617673ff3..dd33de83dc 100644 --- a/queue_job/readme/DESCRIPTION.md +++ b/queue_job/readme/DESCRIPTION.md @@ -34,7 +34,7 @@ Features: - Views for jobs, jobs are stored in PostgreSQL - Jobrunner: execute the jobs, highly efficient thanks to PostgreSQL's NOTIFY -- Channels: give a capacity for the root channel and its sub-channels +- Channels: give a capacity for the root channel and its subchannels and segregate jobs in them. Allow for instance to restrict heavy jobs to be executed one at a time while little ones are executed 4 at a times. diff --git a/queue_job/static/description/index.html b/queue_job/static/description/index.html index ae17d9b53b..d7cc39a366 100644 --- a/queue_job/static/description/index.html +++ b/queue_job/static/description/index.html @@ -405,10 +405,9 @@

Job Queue

  • Views for jobs, jobs are stored in PostgreSQL
  • Jobrunner: execute the jobs, highly efficient thanks to PostgreSQL’s NOTIFY
    • -
  • Channels: give a capacity for the root channel and its sub-channels -and segregate jobs in them. Allow for instance to restrict heavy jobs -to be executed one at a time while little ones are executed 4 at a -times.
    • +
  • Channels: give a capacity for the root channel and its subchannels and +segregate jobs in them. Allow for instance to restrict heavy jobs to +be executed one at a time while little ones are executed 4 at a times.
  • Retries: Ability to retry jobs by raising a type of exception
  • Retry Pattern: the 3 first tries, retry after 10 seconds, the 5 next tries, retry after 1 minutes, …
    • @@ -484,24 +483,87 @@

    Installation

    Configuration

    +

    There are two ways to configure the job runner:

    +

    Set channels (or ODOO_QUEUE_JOB_CHANNELS) and every database +shares the same channel tree (we will call it server-wide channels):

    +
    +[queue_job]
+channels = root:10,root.priority:3,root.slow:1
+
    +

    Leave channels unset and set max_capacity instead. The job +runner then builds a separate channel tree for each database, based on +the Job Channels configured on each database (we will call it +per-database channels).

    +
    +[queue_job]
+max_capacity = 10
+
    +

    channels always has precedence over max_capacity. If +channels is set, the per-database configuration is not used. If +neither channels nor max_capacity are set, the default execution +mode is per-database channels with a max_capacity of 1.

    +

    In the per-database mode, channels are configured from the Job +Channels menu (or by XML data, see the Usage) instead of a global +configuration string.

    +

    Alongside max_capacity, a global configuration db_max_capacity +can be set. It represents the max number of jobs executed at the same +time for a single database (capped by the max_capacity anyway):

    +
    +[queue_job]
+max_capacity = 10
+db_max_capacity = 3  # no more than 3 simultaneous jobs per database
+
    +

    db_max_capacity may be an integer or a pattern such as +prod_*:20,staging:2,*:5, where the first match wins. When using a +pattern, unmatched databases can be configured by a global pattern at +the end (*:n), otherwise they will use the max_capacity.

    +

    The root channel capacity of a database can still be set independently, +however, will in any case be capped by the global max_capacity and +db_max_capacity parameters.

    +

    When set to 0, max_capacity or db_max_capacity means there is no +jobs executed.

    +

    Editing a channel’s capacity, sequential flag, throttle or set it to +pause from the Job Channels menu is applied live on the job +runner.

    +
    +

    Note

    +

    A new database still needs the jobrunner to be restarted.

    +
    +

    When using the server-wide channels, the configuration is static and +loaded at startup of the jobrunner.

    +

    The execution of channels by the job runner is defined by:

    +
      +
    • capacity: max number of jobs running at once in the channel (0 +means no limit of its own, the parent channel and max_capacity or +db_max_capacity still apply)
      • +
    • sequential: jobs run one after the other, and a failed job blocks +the channel (requires a capacity of 1)
      • +
    • throttle: minimum delay, in seconds, between the start of two jobs
      • +
    • paused: stop running jobs in this channel and its subchannels
      • +
    +

    Job Runner Configuration Parameters

      -
    • Using environment variables and command line:
        +
      • Using environment variables:
        • Adjust environment variables (optional):
            -
          • ODOO_QUEUE_JOB_CHANNELS=root:4 or any other channels -configuration. The default is root:1
            • +
          • ODOO_QUEUE_JOB_CHANNELS=root:4 or any other channels for +server-wide channels
            • +
          • ODOO_QUEUE_JOB_MAX_CAPACITY=4, max number of concurrent jobs +(not used if ODOO_QUEUE_JOB_CHANNELS is set) for per-database +channels
            • +
          • ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2, max number of concurrent +jobs per DB (not used if ODOO_QUEUE_JOB_CHANNELS is set)
          • ODOO_QUEUE_JOB_PORT=8069, default --http-port
          • ODOO_QUEUE_JOB_SCHEME=https, default http
          • ODOO_QUEUE_JOB_HOST=load-balancer, default --http-interface or localhost if unset
          • ODOO_QUEUE_JOB_HTTP_AUTH_USER=jobrunner, default empty
          • ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t, default empty
            • -
          • Start Odoo with --load=web,queue_job and --workers greater -than 1. [1]
        • -
      • Using the Odoo configuration file:
        • +
      • Using the Odoo configuration file (set either channels, either +max_capacity and/or db_max_capacity)
       [options]
@@ -512,6 +574,8 @@ 
      Configuration
      
       (...)
 [queue_job]
 channels = root:2
+max_capacity = 8
+db_max_capacity = 3
 scheme = https
 host = load-balancer
 port = 443
@@ -519,6 +583,18 @@ 
      Configuration
      
       http_auth_password = s3cr3t
        +
      • Odoo has to be started with queue_job as server-wide module, +either using the command line option --load=web,queue_job, either +by setting it in the Odoo configuration file, and --workers +greater than 1. [1]
        • +
      +
      +[options]
+(...)
+workers = 6
+server_wide_modules = web,queue_job
+
      +
      • Confirm the runner is starting correctly by checking the odoo log file:
      @@ -529,14 +605,24 @@

      Configuration

      ...INFO...queue_job.jobrunner.runner: database connections ready
        -
      • Create jobs (eg using base_import_async) and observe they start -immediately and in parallel.
        • +
      • Create jobs (you can create test jobs by opening +https://yourodoourl/queue_job/create_test_job) and observe they +start immediately and in parallel.
      • Tip: to enable debug logging for the queue job, use --log-handler=odoo.addons.queue_job:DEBUG
        • -
      • Jobs that remain in enqueued or started state (because, for -instance, their worker has been killed) will be automatically -re-queued.
      +

      Migrating from server-wide channels to per-database channels

      +

      As long as channels (or ODOO_QUEUE_JOB_CHANNELS) is set, the job +runner keeps using the server-wide channels.

      +

      To move to channels per database:

      +
        +
      1. Configure the channels you need on each database, from the Job +Channels menu: capacity, sequential, throttle, pause
        2. +
      2. Once this configuration is done, remove the channels options and +set max_capacity (and optionally db_max_capacity) instead (or +their corresponding environment variables)
        3. +
      3. Restart the job runner
        4. +
      diff --git a/queue_job/static/src/channel_tree/channel_tree.esm.js b/queue_job/static/src/channel_tree/channel_tree.esm.js new file mode 100644 index 0000000000..7adc3fde84 --- /dev/null +++ b/queue_job/static/src/channel_tree/channel_tree.esm.js @@ -0,0 +1,230 @@ +/* @odoo-module */ +/* global vis */ + +import {loadCSS, loadJS} from "@web/core/assets"; + +import {_t} from "@web/core/l10n/translation"; +import {registry} from "@web/core/registry"; +import {useService} from "@web/core/utils/hooks"; + +const {Component, onWillStart, useRef, onMounted} = owl; + +const {document} = globalThis; + +const SERVER_NODE = "server_node"; + +function getCSSVariableValue(key) { + return getComputedStyle(document.documentElement) + .getPropertyValue(`--${key}`) + .trim(); +} + +const CHANNEL_FIELDS = [ + "name", + "complete_name", + "parent_id", + "capacity", + "effective_capacity", + "sequential", + "throttle", + "paused", + "effective_paused", + "capacity_default", + "sequential_default", +]; + +class ChannelTree extends Component { + static template = "queue_job.ChannelTree"; + static props = ["*"]; + + setup() { + this.orm = useService("orm"); + this.action = useService("action"); + this.rootRef = useRef("root_vis"); + this.network = null; + + onWillStart(async () => { + await Promise.all([ + loadJS("/queue_job/static/lib/vis/vis-network.min.js"), + loadCSS("/queue_job/static/lib/vis/vis-network.min.css"), + ]); + await this.loadChannels(); + }); + + onMounted(() => this.renderNetwork()); + } + + get $el() { + return this.rootRef.el; + } + + async loadChannels() { + this.channels = await this.orm.searchRead( + "queue.job.channel", + [], + CHANNEL_FIELDS + ); + } + + htmlTitle(html) { + const container = document.createElement("div"); + container.innerHTML = html; + return container; + } + + edgeWidth(capacity) { + if (!capacity) { + return 1; + } + return capacity; + } + + nodePopup(channel) { + const rows = [ + [_t("Capacity"), channel.capacity || _t("inherited")], + [_t("Effective capacity"), channel.effective_capacity], + [_t("Sequential"), channel.sequential ? _t("yes") : _t("no")], + [ + _t("Throttle"), + channel.throttle ? `${channel.throttle}${_t("s")}` : _t("none"), + ], + [_t("Default capacity"), channel.capacity_default || _t("inherited")], + [ + _t("Default sequential"), + channel.sequential_default ? _t("yes") : _t("no"), + ], + ]; + + let pausedStatus = ""; + if (channel.paused) { + pausedStatus = ` (${_t("Paused")})`; + } else if (channel.effective_paused) { + pausedStatus = ` (${_t("Paused by parent")})`; + } + const headerTitle = `${channel.complete_name}${pausedStatus}`; + const header = `
      ${headerTitle}
      `; + + const lines = rows + .map(([label, value]) => `
      ${label}: ${value}
      `) + .join(""); + + const hints = + `
      ` + + `
      ${_t("Double-click to open the channel")}
      ` + + `
      ${_t("Right-click to create a subchannel")}
      ` + + `
      `; + return `
      ${header}${lines}${hints}
      `; + } + + renderNetwork() { + if (this.network) { + this.$el.innerHTML = ""; + } + + const activeColor = getCSSVariableValue("teal"); + const pausedColor = getCSSVariableValue("orange"); + + const nodes = this.channels.map((channel) => ({ + id: channel.id, + label: channel.name, + title: this.htmlTitle(this.nodePopup(channel)), + color: channel.effective_paused ? pausedColor : activeColor, + // Show root channel larger + font: channel.parent_id ? undefined : {size: 20}, + })); + + const edges = this.channels + .filter((channel) => channel.parent_id) + .map((channel) => ({ + from: channel.parent_id[0], + to: channel.id, + width: this.edgeWidth(channel.effective_capacity), + label: String(channel.effective_capacity), + title: _t("Effective capacity: %s", channel.effective_capacity), + })); + + const rootChannel = this.channels.find((channel) => !channel.parent_id); + if (rootChannel) { + // Build an invisible node on the left of the root channel to + // show an outgoing line towards "an exit" (execution by the server) + nodes.push({ + id: SERVER_NODE, + label: "", + color: {background: "transparent", border: "transparent"}, + chosen: false, + }); + edges.push({ + from: SERVER_NODE, + to: rootChannel.id, + width: this.edgeWidth(rootChannel.effective_capacity), + label: String(rootChannel.effective_capacity), + title: _t("Server capacity: %s", rootChannel.effective_capacity), + // The line would be transparent (as the node itself) otherwise + color: {color: activeColor, inherit: false}, + }); + } + + const data = { + nodes: new vis.DataSet(nodes), + edges: new vis.DataSet(edges), + }; + + const options = { + layout: { + hierarchical: { + direction: "LR", + sortMethod: "directed", + // Align each channel "level" with the same level (default + // is to align leaves on the right (e.g. with root.p1 and + // root.p2.foo, shakeTowards: roots aligns p1 with p2, + // whereas the defaults aligns p1 with foo) + shakeTowards: "roots", + }, + }, + physics: false, + }; + + const network = new vis.Network(this.$el, data, options); + network.on("doubleClick", (params) => { + if (params.nodes.length > 0 && params.nodes[0] !== SERVER_NODE) { + this.actionOpenChannel(params.nodes[0]); + } + }); + network.on("oncontext", (params) => { + params.event.preventDefault(); + const nodeId = network.getNodeAt(params.pointer.DOM); + if (nodeId !== undefined && nodeId !== SERVER_NODE) { + this.actionNewChildChannel(nodeId); + } + }); + + this.network = network; + } + + async actionOpenChannel(resId) { + const action = await this.orm.call("queue.job.channel", "get_formview_action", [ + [resId], + ]); + await this.action.doAction(action); + } + + async actionNewChildChannel(parentId) { + await this.action.doAction( + { + type: "ir.actions.act_window", + res_model: "queue.job.channel", + views: [[false, "form"]], + target: "new", + context: {default_parent_id: parentId}, + }, + { + onClose: async () => { + await this.loadChannels(); + this.renderNetwork(); + }, + } + ); + } +} + +registry.category("actions").add("queue_job_channel_tree", ChannelTree); diff --git a/queue_job/static/src/channel_tree/channel_tree.scss b/queue_job/static/src/channel_tree/channel_tree.scss new file mode 100644 index 0000000000..d12beb1e41 --- /dev/null +++ b/queue_job/static/src/channel_tree/channel_tree.scss @@ -0,0 +1,28 @@ +.o_queue_job_channel_tree { + display: flex; + flex-direction: column; + height: 100%; + width: 100%; + + .o_queue_job_channel_tree_header { + padding: 8px 16px; + border-bottom: 1px solid $border-color; + } + + .o_queue_job_channel_tree_canvas { + flex: 1 1 auto; + min-height: 0; + + // style vis-network tooltip matching odoo styling + // as the default style does not really match... + .vis-tooltip { + background-color: $white; + border: 1px solid $border-color; + border-radius: $border-radius; + box-shadow: $box-shadow-sm; + color: $body-color; + font-family: $font-family-base; + font-size: $font-size-sm; + } + } +} diff --git a/queue_job/static/src/channel_tree/channel_tree.xml b/queue_job/static/src/channel_tree/channel_tree.xml new file mode 100644 index 0000000000..4774370f0c --- /dev/null +++ b/queue_job/static/src/channel_tree/channel_tree.xml @@ -0,0 +1,15 @@ + + + + +
      +
      + The properties shown here are only relevant when the + per-database channels are used, not when server-wide channels + configuration is used. +
      +
      +
      + + + diff --git a/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.esm.js b/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.esm.js index c9b69acebb..327d7ce26c 100644 --- a/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.esm.js +++ b/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.esm.js @@ -21,8 +21,10 @@ class JobDirectGraph extends Component { this.network = null; this.forceRender = false; onWillStart(async () => { - await loadJS("/queue_job/static/lib/vis/vis-network.min.js"); - loadCSS("/queue_job/static/lib/vis/vis-network.min.css"); + await Promise.all([ + loadJS("/queue_job/static/lib/vis/vis-network.min.js"), + loadCSS("/queue_job/static/lib/vis/vis-network.min.css"), + ]); }); useEffect( () => { diff --git a/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.scss b/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.scss index acdf88442b..bfe473d9cd 100644 --- a/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.scss +++ b/queue_job/static/src/views/fields/job_direct_graph/job_direct_graph.scss @@ -6,5 +6,17 @@ div.root_vis { height: 100%; width: 100%; + + // style vis-network tooltip matching odoo styling + // as the default style does not really match... + .vis-tooltip { + background-color: $white; + border: 1px solid $border-color; + border-radius: $border-radius; + box-shadow: $box-shadow-sm; + color: $body-color; + font-family: $font-family-base; + font-size: $font-size-sm; + } } } diff --git a/queue_job/tests/__init__.py b/queue_job/tests/__init__.py index 009f183567..a31cc36f8e 100644 --- a/queue_job/tests/__init__.py +++ b/queue_job/tests/__init__.py @@ -1,5 +1,5 @@ from . import test_run_job_controller_http -from . import test_run_rob_controller +from . import test_run_job_controller from . import test_runner_channels from . import test_runner_runner from . import test_delayable diff --git a/queue_job/tests/common.py b/queue_job/tests/common.py index f74e4ad651..14d7cabba3 100644 --- a/queue_job/tests/common.py +++ b/queue_job/tests/common.py @@ -1,13 +1,10 @@ # Copyright 2019 Camptocamp # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -import doctest -import logging import typing from contextlib import contextmanager from operator import attrgetter from unittest import TestCase, mock -from odoo.tests.case import TestCase as _TestCase from odoo.tools import groupby from odoo.addons.queue_job.delay import Graph @@ -417,51 +414,3 @@ def test_export(self): delayable = mock.MagicMock(name="DelayableBinding") delayable_cls.return_value = delayable yield delayable_cls, delayable - - -class OdooDocTestCase(doctest.DocTestCase, _TestCase): - """ - We need a custom DocTestCase class in order to: - - define test_tags to run as part of standard tests - - output a more meaningful test name than default "DocTestCase.runTest" - """ - - def __init__( - self, doctest, optionflags=0, setUp=None, tearDown=None, checker=None, seq=0 - ): - super().__init__( - doctest._dt_test, - optionflags=optionflags, - setUp=setUp, - tearDown=tearDown, - checker=checker, - ) - self.test_sequence = seq - - def setUp(self): - """Log an extra statement which test is started.""" - super().setUp() - logging.getLogger(__name__).info("Running tests for %s", self._dt_test.name) - - -def load_doctests(module): - """ - Generates a tests loading method for the doctests of the given module - https://docs.python.org/3/library/unittest.html#load-tests-protocol - """ - - def load_tests(loader, tests, ignore): - """ - Apply the 'test_tags' attribute to each DocTestCase found by the DocTestSuite. - Also extend the DocTestCase class trivially to fit the class teardown - that Odoo backported for its own test classes from Python 3.8. - """ - - for idx, test in enumerate(doctest.DocTestSuite(module)): - odoo_test = OdooDocTestCase(test, seq=idx) - odoo_test.test_tags = {"standard", "at_install", "queue_job", "doctest"} - tests.addTest(odoo_test) - - return tests - - return load_tests diff --git a/queue_job/tests/test_model_job_channel.py b/queue_job/tests/test_model_job_channel.py index 20ebbc0bfe..19eaf320be 100644 --- a/queue_job/tests/test_model_job_channel.py +++ b/queue_job/tests/test_model_job_channel.py @@ -1,11 +1,16 @@ # copyright 2018 Camptocamp # license lgpl-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +from unittest import mock + from psycopg2 import IntegrityError import odoo +from odoo import exceptions from odoo.tests import common +from odoo.addons.queue_job.jobrunner import runner + class TestJobChannel(common.TransactionCase): def setUp(self): @@ -57,3 +62,177 @@ def test_channel_display_name(self): {"name": "test", "parent_id": self.root_channel.id} ) self.assertEqual(channel.display_name, channel.complete_name) + + def test_capacity_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The capacity of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_capacity", + "parent_id": self.root_channel.id, + "capacity": -1, + } + ) + + def test_capacity_default_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The default capacity of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_capacity_default", + "parent_id": self.root_channel.id, + "capacity_default": -1, + } + ) + + def test_throttle_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The throttle of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_throttle", + "parent_id": self.root_channel.id, + "throttle": -1, + } + ) + + def test_sequential_should_have_capacity_one(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "A sequential channel must have a capacity of 1.", + ): + self.Channel.create( + { + "name": "test_sequential", + "parent_id": self.root_channel.id, + "sequential": True, + "capacity": 2, + } + ) + + def test_sequential_default_should_have_capacity_default_one(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "A channel with a sequential default must have a default capacity of 1.", + ): + self.Channel.create( + { + "name": "test_sequential_default", + "parent_id": self.root_channel.id, + "sequential_default": True, + "capacity_default": 2, + } + ) + + def test_effective_capacity_root(self): + with ( + mock.patch.object(runner, "_max_capacity", lambda: 42), + mock.patch.object(runner, "_db_max_capacity", lambda: ""), + ): + self.root_channel.capacity = 5 + self.assertEqual(self.root_channel.effective_capacity, 5) + + self.root_channel.capacity = 0 + self.assertEqual(self.root_channel.effective_capacity, 42) + + def test_effective_capacity_subchannel(self): + with ( + # the root channel is restrained to _max_capacity, ensure + # it is high enough for the test + mock.patch.object(runner, "_max_capacity", lambda: 100), + mock.patch.object(runner, "_db_max_capacity", lambda: ""), + ): + self.root_channel.capacity = 33 + + channel = self.Channel.create( + {"name": "test", "parent_id": self.root_channel.id, "capacity": 0} + ) + + self.assertEqual(channel.effective_capacity, 33) + + subchannel = self.Channel.create( + {"name": "test", "parent_id": channel.id, "capacity": 0} + ) + self.assertEqual(subchannel.effective_capacity, 33) + + channel.capacity = 18 + + self.assertEqual(channel.effective_capacity, 18) + self.assertEqual(subchannel.effective_capacity, 18) + + def test_effective_paused(self): + channel = self.Channel.create( + {"name": "test", "parent_id": self.root_channel.id} + ) + subchannel = self.Channel.create({"name": "test", "parent_id": channel.id}) + + self.assertFalse(channel.effective_paused) + self.assertFalse(subchannel.effective_paused) + + channel.paused = True + + self.assertTrue(channel.effective_paused) + self.assertTrue(subchannel.effective_paused) + + def test_action_pause(self): + self.root_channel.action_pause() + self.assertTrue(self.root_channel.paused) + + def test_action_resume(self): + self.root_channel.paused = True + self.root_channel.action_resume() + self.assertFalse(self.root_channel.paused) + + def _patch_notify(self): + return mock.patch.object( + type(self.Channel), "_notify_channel_config_changed", autospec=True + ) + + def test_notify_create_channel(self): + with self._patch_notify() as notify: + self.Channel.create( + { + "name": "create_notify", + "parent_id": self.root_channel.id, + "capacity": 2, + } + ) + notify.assert_called_once() + + def test_notify_write_jobrunner_config(self): + channel = self.Channel.create( + {"name": "write_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.capacity = 3 + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.paused = True + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.removal_interval = 60 + notify.assert_not_called() + + with self._patch_notify() as notify: + channel.capacity_default = 1 + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.sequential_default = True + notify.assert_called_once() + + def test_notify_unlink_channel(self): + channel = self.Channel.create( + {"name": "unlink_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.unlink() + notify.assert_called_once() diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_job_controller.py similarity index 97% rename from queue_job/tests/test_run_rob_controller.py rename to queue_job/tests/test_run_job_controller.py index 4090c811ea..2c60c5b0a3 100644 --- a/queue_job/tests/test_run_rob_controller.py +++ b/queue_job/tests/test_run_job_controller.py @@ -39,17 +39,17 @@ def test_runjob_on_fail(self): function.on_fail_method = "_test_on_fail" job = self.env["queue.job"].with_delay()._test_job(failure_rate=1) with ( - self.assertRaises(JobError), patch( "odoo.addons.queue_job.models.queue_job.QueueJob._test_on_fail" ) as mocked_hook, patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env, + self.assertRaises(JobError), mute_logger("odoo.addons.queue_job.controllers.main"), ): mocked_temp_env.return_value.__enter__.return_value = self.env RunJobController._runjob(self.env, job) - self.assertEqual(job.state, "failed") - self.assertEqual(mocked_hook.call_count, 1) + self.assertEqual(job.state, "failed") + self.assertEqual(mocked_hook.call_count, 1) @mute_logger("odoo.addons.queue_job.controllers.main") def test_runjob_on_fail_no_deadlock(self): @@ -100,8 +100,8 @@ def on_fail(**kwargs): def test_runjob_on_fail_not_configured(self): job = self.env["queue.job"].with_delay()._test_job(failure_rate=1) with ( - self.assertRaises(JobError), patch("odoo.addons.queue_job.job.Job.in_temporary_env") as mocked_temp_env, + self.assertRaises(JobError), mute_logger("odoo.addons.queue_job.controllers.main"), ): mocked_temp_env.return_value.__enter__.return_value = self.env diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index d323d00683..313e350bd3 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -1,10 +1,18 @@ # Copyright 2015-2016 Camptocamp SA # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +import doctest + +from odoo.tests import BaseCase, tagged # pylint: disable=odoo-addons-relative-import # we are testing, we want to test as we were an external consumer of the API from odoo.addons.queue_job.jobrunner import channels -from .common import load_doctests -load_tests = load_doctests(channels) +@tagged("doctest") +class TestDoctest(BaseCase): + def test_doctest(self): + results = doctest.testmod( + channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE + ) + self.assertEqual(results.failed, 0, "doctest failed") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 2adbf610db..7967429877 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -1,23 +1,76 @@ # Copyright 2015-2016 Camptocamp SA # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) - -# pylint: disable=odoo-addons-relative-import -# we are testing, we want to test as we were an external consumer of the API +import doctest import os import threading +from unittest import mock from unittest.mock import patch from odoo.tests import BaseCase, tagged +from odoo.tools import mute_logger +# pylint: disable=odoo-addons-relative-import +# we are testing, we want to test as we were an external consumer of the API from odoo.addons.queue_job.jobrunner import START_DELAY, QueueJobRunnerThread, runner +from odoo.addons.queue_job.jobrunner.channels import ChannelConfig, ChannelManager -from .common import load_doctests -load_tests = load_doctests(runner) +@tagged("doctest") +class TestDoctest(BaseCase): + def test_doctest(self): + results = doctest.testmod( + runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE + ) + self.assertEqual(results.failed, 0, "doctest failed") @tagged("-at_install", "post_install") class TestRunner(BaseCase): + def setUp(self): + super().setUp() + # ensure there is no collision with actual environment variables/config + env_patcher = mock.patch.dict( + "os.environ", + { + "ODOO_QUEUE_JOB_MAX_CAPACITY": "", + "ODOO_QUEUE_JOB_DB_MAX_CAPACITY": "", + "ODOO_QUEUE_JOB_CHANNELS": "", + }, + ) + env_patcher.start() + self.addCleanup(env_patcher.stop) + + conf_patcher = mock.patch.object(runner, "queue_job_config", {}) + conf_patcher.start() + self.addCleanup(conf_patcher.stop) + + @staticmethod + def _mock_db(db_name, channels_config): + """Mock a Database + + :param channels_config: list of ChannelConfig as configured in the DB + """ + return mock.MagicMock( + db_name=db_name, + has_queue_job=True, + load_channels_config=mock.MagicMock(return_value=channels_config), + ) + + def _register_channel_manager(self, jobrunner, db_name, channel_manager): + jobrunner._channel_manager_by_db[db_name] = channel_manager + jobrunner._channel_managers.append(channel_manager) + + def _new_channel_manager(self, jobrunner, db_name, channel_config, pending_jobs=0): + channel_manager = ChannelManager() + channel_manager.configure(channel_config) + self._register_channel_manager(jobrunner, db_name, channel_manager) + jobrunner.db_by_name[db_name] = self._mock_db(db_name, channel_config) + for number in range(pending_jobs): + channel_manager.notify( + db_name, "root", f"{db_name}-{number}", number, 0, 10, None, "pending" + ) + return channel_manager + @classmethod def _is_open_file_descriptor(cls, fd): try: @@ -91,3 +144,378 @@ def test_runner_thread_stop_interrupts_the_start_delay(self): thread.stop() thread.join(timeout=START_DELAY / 2) self.assertFalse(thread.is_alive()) + + def test_max_capacity_from_env(self): + with ( + mock.patch.dict( + os.environ, {"ODOO_QUEUE_JOB_MAX_CAPACITY": "5"}, clear=True + ), + mock.patch.object(runner, "queue_job_config", {}), + ): + self.assertEqual(runner._max_capacity(), 5) + + def test_max_capacity_from_odoo_config(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object(runner, "queue_job_config", {"max_capacity": "3"}), + ): + self.assertEqual(runner._max_capacity(), 3) + + def test_max_capacity_env_priority_over_odoo_config(self): + with ( + mock.patch.dict( + os.environ, {"ODOO_QUEUE_JOB_MAX_CAPACITY": "5"}, clear=True + ), + mock.patch.object(runner, "queue_job_config", {"max_capacity": "3"}), + ): + self.assertEqual(runner._max_capacity(), 5) + + def test_max_capacity_default(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object(runner, "queue_job_config", {}), + ): + self.assertEqual(runner._max_capacity(), 1) + + def test_max_capacity_zero(self): + with ( + mock.patch.dict( + os.environ, {"ODOO_QUEUE_JOB_MAX_CAPACITY": "0"}, clear=True + ), + mock.patch.object(runner, "queue_job_config", {}), + ): + self.assertEqual(runner._max_capacity(), 0) + + def test_db_max_capacity_from_env(self): + with ( + mock.patch.dict( + os.environ, {"ODOO_QUEUE_JOB_DB_MAX_CAPACITY": "prod_*:20,*:5"} + ), + mock.patch.object(runner, "queue_job_config", {}), + ): + self.assertEqual(runner._db_max_capacity(), "prod_*:20,*:5") + + def test_db_max_capacity_from_odoo_config(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object( + runner, "queue_job_config", {"db_max_capacity": "prod_*:20,*:5"} + ), + ): + self.assertEqual(runner._db_max_capacity(), "prod_*:20,*:5") + + def test_db_max_capacity_env_priority_over_odoo_config(self): + with ( + mock.patch.dict(os.environ, {"ODOO_QUEUE_JOB_DB_MAX_CAPACITY": "*:5"}), + mock.patch.object(runner, "queue_job_config", {"db_max_capacity": "*:3"}), + ): + self.assertEqual(runner._db_max_capacity(), "*:5") + + def test_db_max_capacity_default(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object(runner, "queue_job_config", {}), + ): + self.assertEqual(runner._db_max_capacity(), "") + + def test_max_capacity_negative(self): + jobrunner = runner.QueueJobRunner(max_capacity=-5) + self.assertEqual(jobrunner.max_capacity, 0) + + def test_max_capacity_equals_to_channel_config_string(self): + jobrunner = runner.QueueJobRunner(channel_config_string="root:3") + self.assertEqual(jobrunner.max_capacity, 3) + + def test_max_capacity_server_wide_has_priority(self): + jobrunner = runner.QueueJobRunner( + channel_config_string="root:3", max_capacity=1 + ) + self.assertEqual(jobrunner.max_capacity, 3) + + def test_max_capacity_channel_config_no_root(self): + jobrunner = runner.QueueJobRunner(channel_config_string="sub:3") + self.assertEqual(jobrunner.max_capacity, 1) + + def test_no_job_dispatched_when_no_database_channel_manager(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + with mock.patch.object(jobrunner, "_dispatch_job") as dispatch: + jobrunner.run_jobs() + dispatch.assert_not_called() + + def test_global_capacity_across_databases(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + manager_a = self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 5)], pending_jobs=4 + ) + manager_b = self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 5)], pending_jobs=4 + ) + with mock.patch.object(jobrunner, "_dispatch_job") as dispatch: + jobrunner.run_jobs() + # number of running jobs must be limited by max_capacity + self.assertEqual(dispatch.call_count, 3) + self.assertEqual(manager_a.running_count + manager_b.running_count, 3) + + def test_round_robin(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)], pending_jobs=2 + ) + self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 3)], pending_jobs=2 + ) + dispatched = [] + with mock.patch.object( + jobrunner, "_dispatch_job", side_effect=lambda job: dispatched.append(job) + ): + jobrunner.run_jobs() + # with max_capacity=3 and 2 pending jobs per db, the first + # run_jobs() will dispatch 3 jobs in alternating order + self.assertEqual( + [job.db_name for job in dispatched], ["db_a", "db_b", "db_a"] + ) + # notify the first dispatched job as done so a slot is freed up + # so we can check that the next call to run_jobs() continues + # to round-robin + jobrunner._channel_manager_by_db[dispatched[0].db_name].notify( + dispatched[0].db_name, + "root", + dispatched[0].uuid, + dispatched[0].seq, + 0, + 10, + None, + "done", + ) + jobrunner.run_jobs() + # round-robin continues on the next dispatch "ticks" + self.assertEqual( + [job.db_name for job in dispatched], ["db_a", "db_b", "db_a", "db_b"] + ) + + def test_round_robin_empty_channel_manager(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)], pending_jobs=1 + ) + self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 3)], pending_jobs=3 + ) + dispatched = [] + with mock.patch.object( + jobrunner, "_dispatch_job", side_effect=lambda job: dispatched.append(job) + ): + jobrunner.run_jobs() + # round-robin continues with the only channel manager still having pending jobs + self.assertEqual([job.db_name for job in dispatched], ["db_a", "db_b", "db_b"]) + + def test_build_channel_manager_root_db_max_capacity(self): + jobrunner = runner.QueueJobRunner(max_capacity=10) + db = self._mock_db("db_a", []) + channel_manager = jobrunner._build_channel_manager(db) + root = channel_manager.get_channel_by_name("root") + # the root channels inherits the global max capacity + self.assertEqual(root.capacity, 10) + + def test_build_channel_manager_db_max_zero(self): + jobrunner = runner.QueueJobRunner(max_capacity=3, db_max_capacity="db_a:0,*:5") + db = self._mock_db("db_a", [ChannelConfig("root", 5)]) + channel_manager = jobrunner._build_channel_manager(db) + root = channel_manager.get_channel_by_name("root") + # when db max capacity is at 0, the root channel of the channel manager + # is considered paused + self.assertTrue(root.paused) + + def test_build_channel_manager_db_max_negative(self): + jobrunner = runner.QueueJobRunner(max_capacity=3, db_max_capacity="db_a:-1,*:5") + db = self._mock_db("db_a", [ChannelConfig("root", 5)]) + channel_manager = jobrunner._build_channel_manager(db) + root = channel_manager.get_channel_by_name("root") + # negative value is handled as 0, which is paused + self.assertTrue(root.paused) + + def test_build_channel_manager_root_capacity_db(self): + jobrunner = runner.QueueJobRunner(max_capacity=3, db_max_capacity="*:2") + db = self._mock_db("db_a", [ChannelConfig("root", 20)]) + channel_manager = jobrunner._build_channel_manager(db) + root = channel_manager.get_channel_by_name("root") + # the root channel is set to the db max capacity + self.assertEqual(root.capacity, 2) + + @mute_logger("odoo.addons.queue_job.jobrunner.runner") + def test_build_channel_manager_schema_error(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + # when the database schema has no columns, None is returned as channels config + db = self._mock_db("db_a", None) + channel_manager = jobrunner._build_channel_manager(db) + root = channel_manager.get_channel_by_name("root") + self.assertEqual(root.capacity, 0) + self.assertTrue(root.paused) + + def test_build_channel_manager_handle_missing_root(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + # the configuration does not have root + db = self._mock_db("db_a", [ChannelConfig("sub", 2)]) + channel_manager = jobrunner._build_channel_manager(db) + # does not raise, "root" is automatically added + root = channel_manager.get_channel_by_name("root", autocreate=False) + self.assertEqual(root.capacity, 3) + + def test_register_db_server_wide(self): + jobrunner = runner.QueueJobRunner(channel_config_string="root:3") + global_manager = jobrunner._server_wide_channel_manager + db = self._mock_db("db_a", None) + jobrunner._register_db_server_wide(db) + + self.assertEqual(len(jobrunner._channel_managers), 1) + self.assertEqual(jobrunner._channel_managers, [global_manager]) + self.assertEqual(jobrunner._channel_manager_by_db[db.db_name], global_manager) + + def test_max_capacity_zero_no_dispatch(self): + jobrunner = runner.QueueJobRunner(max_capacity=0) + self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 10)], pending_jobs=4 + ) + self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 10)], pending_jobs=4 + ) + with mock.patch.object(jobrunner, "_dispatch_job") as dispatch: + jobrunner.run_jobs() + dispatch.assert_not_called() + + def test_max_capacity_unconfigured_defaults_to_one(self): + jobrunner = runner.QueueJobRunner() + self.assertEqual(jobrunner.max_capacity, 1) + self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 10)], pending_jobs=4 + ) + self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 10)], pending_jobs=4 + ) + with mock.patch.object(jobrunner, "_dispatch_job") as dispatch: + jobrunner.run_jobs() + self.assertEqual(dispatch.call_count, 1) + + def test_notify_reload(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + db = self._mock_db("db_a", []) + db.conn.notifies = [mock.Mock(payload=runner.RELOAD_PAYLOAD)] + jobrunner.db_by_name = {"db_a": db} + + with mock.patch.object(jobrunner, "_configure_db") as reconfigure: + jobrunner.process_notifications() + reconfigure.assert_called_once_with(db) + + def test_notify_reload_ignored_with_server_wide_channels(self): + jobrunner = runner.QueueJobRunner(channel_config_string="root:3") + db = self._mock_db("db_a", []) + db.conn.notifies = [mock.Mock(payload=runner.RELOAD_PAYLOAD)] + jobrunner.db_by_name = {"db_a": db} + jobrunner._channel_manager_by_db = {"db_a": mock.MagicMock()} + with mock.patch.object(jobrunner, "_configure_db") as reconfigure: + jobrunner.process_notifications() + reconfigure.assert_not_called() + + def test_notify_job_started(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + db = self._mock_db("db_a", None) + channel_manager = ChannelManager() + channel_manager.configure([ChannelConfig("root", 3)]) + jobrunner.db_by_name = {"db_a": db} + self._register_channel_manager(jobrunner, "db_a", channel_manager) + + db.conn.notifies = [mock.Mock(payload="job-1")] + # channel, uuid, seq, date_created, priority, eta, state + job_row = ("root", "job-1", 0, 0, 10, None, "started") + db.select_jobs.return_value.__enter__.return_value.fetchone.return_value = ( + job_row + ) + + jobrunner.process_notifications() + + db.select_jobs.assert_called_once_with("uuid = %s", ("job-1",)) + # the channel manager now counts the job as a running job + self.assertEqual(channel_manager.running_count, 1) + + @mute_logger("odoo.addons.queue_job.jobrunner.runner") + def test_build_channel_manager_invalid_configuration(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + db = self._mock_db("db_a", [ChannelConfig("root", 2, sequential=True)]) + + channel_manager = jobrunner._build_channel_manager(db) + + root = channel_manager.get_channel_by_name("root") + self.assertTrue(root.paused) + + def test_notify_existing_jobs(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + db = self._mock_db("db_a", [ChannelConfig("root", 3)]) + # channel, uuid, seq, date_created, priority, eta, state + job_row = ("root", "db_a-job1", 0, 0, 10, None, "pending") + db.select_jobs.return_value.__enter__.return_value = [job_row] + + jobrunner.db_by_name = {"db_a": db} + + channel_manager = jobrunner._build_channel_manager(db) + self._register_channel_manager(jobrunner, "db_a", channel_manager) + + jobrunner._notify_existing_jobs(db, channel_manager) + + db.select_jobs.assert_called_once_with("state in %s", (runner.NOT_DONE,)) + pending_jobs = list(channel_manager.get_jobs_to_run(0)) + self.assertEqual([job.uuid for job in pending_jobs], ["db_a-job1"]) + + def test_close_databases_reset_state(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + channel_manager = self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)], pending_jobs=1 + ) + db = jobrunner.db_by_name["db_a"] + + jobrunner.close_databases() + + self.assertEqual(list(channel_manager.get_jobs_to_run(0)), []) + db.close.assert_called_once() + self.assertEqual(jobrunner.db_by_name, {}) + self.assertEqual(jobrunner._channel_manager_by_db, {}) + self.assertEqual(jobrunner._channel_managers, []) + + def test_close_databases_remove_jobs_false(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + channel_manager = self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)], pending_jobs=1 + ) + + jobrunner.close_databases(remove_jobs=False) + + # job is kept when remove_jobs=False + self.assertEqual(len(list(channel_manager.get_jobs_to_run(0))), 1) + + def test_next_wakeup_time_no_channel_managers(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + self.assertEqual(jobrunner.next_wakeup_time(), 0) + + def test_next_wakeup_time_one_db_with_eta(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + channel_manager = self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)] + ) + channel_manager.notify("db_a", "root", "db_a-0", 0, 0, 10, 500, "pending") + self.assertEqual(jobrunner.next_wakeup_time(), 500) + + def test_next_wakeup_time_several_db(self): + jobrunner = runner.QueueJobRunner(max_capacity=3) + channel_manager_a = self._new_channel_manager( + jobrunner, "db_a", [ChannelConfig("root", 3)] + ) + channel_manager_b = self._new_channel_manager( + jobrunner, "db_b", [ChannelConfig("root", 3)] + ) + channel_manager_c = self._new_channel_manager( + jobrunner, "db_c", [ChannelConfig("root", 3)] + ) + # channel, uuid, seq, date_created, priority, eta, state + channel_manager_a.notify("db_a", "root", "db_a-0", 0, 0, 10, 500, "pending") + channel_manager_b.notify("db_b", "root", "db_b-0", 0, 0, 10, 200, "pending") + channel_manager_c.notify("db_c", "root", "db_b-0", 0, 0, 10, 0, "pending") + self.assertEqual(jobrunner.next_wakeup_time(), 200) diff --git a/queue_job/utils.py b/queue_job/utils.py index d037b8b530..871668b64f 100644 --- a/queue_job/utils.py +++ b/queue_job/utils.py @@ -10,7 +10,7 @@ def must_run_without_delay(env): - """Retrun true if jobs have to run immediately. + """Return True if jobs have to run immediately. :param env: `odoo.api.Environment` instance """ diff --git a/queue_job/views/queue_job_channel_views.xml b/queue_job/views/queue_job_channel_views.xml index 50c245716b..9d7813c8ca 100644 --- a/queue_job/views/queue_job_channel_views.xml +++ b/queue_job/views/queue_job_channel_views.xml @@ -5,23 +5,87 @@ queue.job.channel
      - - +