[18.0][ADD] queue_job: per DB channel managers with UI configuration and hot reload

queue_job/README.rst

@@ -128,7 +128,10 @@ Configuration
   - Adjust environment variables (optional):
 
     - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels
-      configuration. The default is ``root:1``
+    - ``ODOO_QUEUE_JOB_MAX_CAPACITY=4``, max number of concurrent jobs
+      (not used if ``ODOO_QUEUE_JOB_CHANNELS`` is set)
+    - ``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
@@ -138,7 +141,8 @@ Configuration
     - 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 ``db_max_capacity``)
 
 .. code:: ini
 
@@ -150,12 +154,17 @@ 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
 
+``db_max_capacity`` may be an integer or a pattern such as
+``prod_*:20,staging:2,*:5``
+
 - Confirm the runner is starting correctly by checking the odoo log
   file:
 

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,
             )
 

queue_job/jobrunner/channels.py

@@ -3,19 +3,32 @@
 # 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
 
 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
+
+
 class PriorityQueue:
     """A priority queue that supports removing arbitrary objects.
 
@@ -965,6 +978,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.
 
@@ -1115,3 +1133,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)