[14.0] Add graph of dependencies

queue_job/README.rst

@@ -23,7 +23,7 @@ Job Queue
     :target: https://runbot.odoo-community.org/runbot/230/14.0
     :alt: Try me on Runbot
 
-|badge1| |badge2| |badge3| |badge4| |badge5| 
+|badge1| |badge2| |badge3| |badge4| |badge5|
 
 This addon adds an integrated Job Queue to Odoo.
 
@@ -409,7 +409,7 @@ promote its widespread use.
 
 Current `maintainer <https://odoo-community.org/page/maintainer-role>`__:
 
-|maintainer-guewen| 
+|maintainer-guewen|
 
 This module is part of the `OCA/queue <https://github.com/OCA/queue/tree/14.0/queue_job>`_ project on GitHub.
 

queue_job/__manifest__.py

@@ -1,6 +1,5 @@
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 
-
 {
     "name": "Job Queue",
     "version": "14.0.2.4.0",
@@ -13,6 +12,7 @@
     "data": [
         "security/security.xml",
         "security/ir.model.access.csv",
+        "views/queue_job_assets.xml",
         "views/queue_job_views.xml",
         "views/queue_job_channel_views.xml",
         "views/queue_job_function_views.xml",

queue_job/controllers/main.py

@@ -3,23 +3,28 @@
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 
 import logging
+import random
+import time
 import traceback
 from io import StringIO
 
-from psycopg2 import OperationalError
-from werkzeug.exceptions import Forbidden
+from psycopg2 import OperationalError, errorcodes
+from werkzeug.exceptions import BadRequest, Forbidden
 
 import odoo
 from odoo import _, http, tools
 from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY
 
+from ..delay import chain, group
 from ..exception import FailedJobError, NothingToDoJob, RetryableJobError
 from ..job import ENQUEUED, Job
 
 _logger = logging.getLogger(__name__)
 
 PG_RETRY = 5  # seconds
 
+DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5
+
 
 class RunJobController(http.Controller):
     def _try_perform_job(self, env, job):
@@ -36,6 +41,35 @@ def _try_perform_job(self, env, job):
         env.cr.commit()
         _logger.debug("%s done", job)
 
+    def _enqueue_dependent_jobs(self, env, job):
+        tries = 0
+        while True:
+            try:
+                job.enqueue_waiting()
+            except OperationalError as err:
+                # Automatically retry the typical transaction serialization
+                # errors
+                if err.pgcode not in PG_CONCURRENCY_ERRORS_TO_RETRY:
+                    raise
+                if tries >= DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE:
+                    _logger.info(
+                        "%s, maximum number of tries reached to update dependencies",
+                        errorcodes.lookup(err.pgcode),
+                    )
+                    raise
+                wait_time = random.uniform(0.0, 2**tries)
+                tries += 1
+                _logger.info(
+                    "%s, retry %d/%d in %.04f sec...",
+                    errorcodes.lookup(err.pgcode),
+                    tries,
+                    DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE,
+                    wait_time,
+                )
+                time.sleep(wait_time)
+            else:
+                break
+
     @http.route("/queue_job/runjob", type="http", auth="none", save_session=False)
     def runjob(self, db, job_uuid, **kw):
         http.request.session.db = db
@@ -116,6 +150,10 @@ def retry_postpone(job, message, seconds=None):
                     buff.close()
             raise
 
+        _logger.debug("%s enqueue depends started", job)
+        self._enqueue_dependent_jobs(env, job)
+        _logger.debug("%s enqueue depends done", job)
+
         return ""
 
     def _get_failure_values(self, job, traceback_txt, orig_exception):
@@ -132,11 +170,42 @@ def _get_failure_values(self, job, traceback_txt, orig_exception):
 
     @http.route("/queue_job/create_test_job", type="http", auth="user")
     def create_test_job(
-        self, priority=None, max_retries=None, channel=None, description="Test job"
+        self,
+        priority=None,
+        max_retries=None,
+        channel=None,
+        description="Test job",
+        size=1,
+        failure_rate=0,
     ):
+        """Create test jobs
+
+        Examples of urls:
+
+        * http://127.0.0.1:8069/queue_job/create_test_job: single job
+        * http://127.0.0.1:8069/queue_job/create_test_job?size=10: a graph of 10 jobs
+        * http://127.0.0.1:8069/queue_job/create_test_job?size=10&failure_rate=0.5:
+          a graph of 10 jobs, half will fail
+
+        """
         if not http.request.env.user.has_group("base.group_erp_manager"):
             raise Forbidden(_("Access Denied"))
 
+        if failure_rate is not None:
+            try:
+                failure_rate = float(failure_rate)
+            except (ValueError, TypeError):
+                failure_rate = 0
+
+        if not (0 <= failure_rate <= 1):
+            raise BadRequest("failure_rate must be between 0 and 1")
+
+        if size is not None:
+            try:
+                size = int(size)
+            except (ValueError, TypeError):
+                size = 1
+
         if priority is not None:
             try:
                 priority = int(priority)
@@ -149,6 +218,35 @@ def create_test_job(
             except ValueError:
                 max_retries = None
 
+        if size == 1:
+            return self._create_single_test_job(
+                priority=priority,
+                max_retries=max_retries,
+                channel=channel,
+                description=description,
+                failure_rate=failure_rate,
+            )
+
+        if size > 1:
+            return self._create_graph_test_jobs(
+                size,
+                priority=priority,
+                max_retries=max_retries,
+                channel=channel,
+                description=description,
+                failure_rate=failure_rate,
+            )
+        return ""
+
+    def _create_single_test_job(
+        self,
+        priority=None,
+        max_retries=None,
+        channel=None,
+        description="Test job",
+        size=1,
+        failure_rate=0,
+    ):
         delayed = (
             http.request.env["queue.job"]
             .with_delay(
@@ -157,7 +255,57 @@ def create_test_job(
                 channel=channel,
                 description=description,
             )
-            ._test_job()
+            ._test_job(failure_rate=failure_rate)
         )
+        return "job uuid: %s" % (delayed.db_record().uuid,)
+
+    TEST_GRAPH_MAX_PER_GROUP = 5
 
-        return delayed.db_record().uuid
+    # flake8: noqa: C901
+    def _create_graph_test_jobs(
+        self,
+        size,
+        priority=None,
+        max_retries=None,
+        channel=None,
+        description="Test job",
+        failure_rate=0,
+    ):
+        model = http.request.env["queue.job"]
+        current_count = 0
+
+        possible_grouping_methods = (chain, group)
+
+        tails = []  # we can connect new graph chains/groups to tails
+        root_delayable = None
+        while current_count < size:
+            jobs_count = min(
+                size - current_count, random.randint(1, self.TEST_GRAPH_MAX_PER_GROUP)
+            )
+
+            jobs = []
+            for __ in range(jobs_count):
+                current_count += 1
+                jobs.append(
+                    model.delayable(
+                        priority=priority,
+                        max_retries=max_retries,
+                        channel=channel,
+                        description="%s #%d" % (description, current_count),
+                    )._test_job(failure_rate=failure_rate)
+                )
+
+            grouping = random.choice(possible_grouping_methods)
+            delayable = grouping(*jobs)
+            if not root_delayable:
+                root_delayable = delayable
+            else:
+                tail_delayable = random.choice(tails)
+                tail_delayable.on_done(delayable)
+            tails.append(delayable)
+
+        root_delayable.delay()
+
+        return "graph uuid: %s" % (
+            list(root_delayable._head())[0]._generated_job.graph_uuid,
+        )