diff --git a/queue_job/README.rst b/queue_job/README.rst index 555196bde1..a284086532 100644 --- a/queue_job/README.rst +++ b/queue_job/README.rst @@ -123,22 +123,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-side 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-side 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 sub-channels + +**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-side 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 +228,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 +258,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-side channels to per-database channels** + +As long as ``channels`` (or ``ODOO_QUEUE_JOB_CHANNELS``) is set, the job +runner keeps using the server-side 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/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py index e2561b0e74..2a1168707b 100644 --- a/queue_job/jobrunner/__init__.py +++ b/queue_job/jobrunner/__init__.py @@ -20,7 +20,7 @@ queue_job_config = config.misc.get("queue_job", {}) -from .runner import QueueJobRunner, _channels +from .runner import QueueJobRunner, _channels, _max_capacity _logger = logging.getLogger(__name__) @@ -87,7 +87,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): @@ -100,7 +105,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 7d6642dab8..636338d112 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. @@ -1023,6 +1038,11 @@ def simple_configure(self, config_string): for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) + def configure(self, configs): + """Configure the channel manager from list of :class:`ChannelConfig`""" + for config in configs: + self.get_channel_from_config(asdict(config)) + def get_channel_from_config(self, config): """Return a Channel object from a parsed configuration. @@ -1173,3 +1193,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 95e134ba44..62a713fff8 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) @@ -123,9 +204,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 @@ -182,6 +265,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 @@ -321,16 +462,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_side_channel_manager = None + if channel_config_string: + channel_manager = ChannelManager() + channel_manager.simple_configure(channel_config_string) + self._server_side_channel_manager = channel_manager + # max_capacity is always equal to the root channel in server-side + # 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_pipe = os.pipe() @@ -387,11 +552,88 @@ 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 _unique_channel_managers(channel_managers): + seen = set() + result = [] + for channel_manager in channel_managers: + if id(channel_manager) not in seen: + seen.add(id(channel_manager)) + result.append(channel_manager) + return result + + 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 + ), + ) + channel_manager = ChannelManager() + + 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 + ) + channel_manager.configure([ChannelConfig(name="root", capacity=0)]) + return channel_manager + + 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.configure(channels_config) + return channel_manager + + def _reconfigure_db(self, db_name): + """Rebuild the channel manager for a database and reload its jobs""" + db = self.db_by_name.get(db_name) + if db is None: + return + if self._server_side_channel_manager: + channel_manager = self._server_side_channel_manager + else: + try: + channel_manager = self._build_channel_manager(db) + except ValueError: + # a bad channel configuration on a single database + # (e.g. sequential with a capacity != 1) should not + # break other databases, skip it + _logger.warning("error configuring db %s", db_name, exc_info=True) + return + with db.select_jobs("state in %s", (NOT_DONE,)) as cr: + for job_data in cr: + channel_manager.notify(db_name, *job_data) + self._register_channel_manager(db_name, channel_manager) + _logger.info("channels configuration loaded for db %s", db_name) + + def _register_channel_manager(self, db_name, channel_manager): + self._channel_manager_by_db[db_name] = channel_manager + self._channel_managers = self._unique_channel_managers( + self._channel_manager_by_db.values() + ) def initialize_databases(self): for db_name in sorted(self.get_db_names()): @@ -399,9 +641,7 @@ 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) + self._reconfigure_db(db_name) _logger.info("queue job runner ready for db %s", db_name) else: db.close() @@ -411,24 +651,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 @@ -440,13 +727,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_side_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._reconfigure_db(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(): @@ -458,7 +771,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 4aabb0188c..96dc8e9d8e 100644 --- a/queue_job/models/queue_job_channel.py +++ b/queue_job/models/queue_job_channel.py @@ -4,12 +4,28 @@ from odoo import _, api, exceptions, fields, models +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,13 +39,68 @@ 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-side " + "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 sub-channels) do not execute any jobs until " + "resumed." + ) + capacity_default = fields.Integer( + help="Default capacity for unconfigured sub-channels. " + "0 means they would have the same capacity as the current channel." + ) + sequential_default = fields.Boolean( + help="If sequential is enabled for unconfigured sub-channels." ) _sql_constraints = [ ("name_uniq", "unique(complete_name)", "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: @@ -70,8 +141,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 ( @@ -80,10 +160,25 @@ def write(self, values): and ("name" in values or "parent_id" in values) ): raise exceptions.UserError(_("Cannot change the root channel")) - return super().write(values) + res = super().write(values) + if self._JOBRUNNER_CONFIG_FIELDS.intersection(values): + self._notify_channel_config_changed() + return res def unlink(self): for channel in self: if channel.name == "root": raise exceptions.UserError(_("Cannot remove the root channel")) - return super().unlink() + 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..95dd380748 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-side 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-side 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 sub-channels + + +**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-side + 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-side channels to per-database channels** + +As long as `channels` (or `ODOO_QUEUE_JOB_CHANNELS`) is set, the job +runner keeps using the server-side 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/static/description/index.html b/queue_job/static/description/index.html index 445a3c2eff..81f0c3f788 100644 --- a/queue_job/static/description/index.html +++ b/queue_job/static/description/index.html @@ -3,7 +3,7 @@ -README.rst +Job Queue