[19.0] queue_job: forward ports from 18.0

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 <dbname>
    ...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

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,

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

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

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