[12.0] Add graph of dependencies between jobs

queue_job/README.rst

@@ -23,7 +23,7 @@ Job Queue
     :target: https://runbot.odoo-community.org/runbot/230/12.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.
 
@@ -90,7 +90,7 @@ Configuration
 
   * Adjust environment variables (optional):
 
-    - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels configuration. 
+    - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels configuration.
       The default is ``root:1``
 
     - if ``xmlrpc_port`` is not set: ``ODOO_QUEUE_JOB_PORT=8069``
@@ -324,7 +324,7 @@ Changelog
     understand changes between version. The primary audience is
     end users and integrators. Purely technical changes such as
     code refactoring must not be mentioned here.
-    
+
     This file may contain ONE level of section titles, underlined
     with the ~ (tilde) character. Other section markers are
     forbidden and will likely break the structure of the README.rst
@@ -416,7 +416,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/12.0/queue_job>`_ project on GitHub.
 

queue_job/__manifest__.py

@@ -14,6 +14,7 @@
  'data': ['security/security.xml',
           'security/ir.model.access.csv',
           'views/queue_job_views.xml',
+          'views/queue_job_assets.xml',
           'data/queue_data.xml',
           "data/queue_job_function_data.xml"],
  'installable': True,

queue_job/controllers/main.py

@@ -2,24 +2,29 @@
 # Copyright 2013-2016 Camptocamp SA
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 
+import random
 import logging
+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 ..job import Job, ENQUEUED
 from ..exception import (RetryableJobError, FailedJobError, NothingToDoJob)
 
 _logger = logging.getLogger(__name__)
 
 PG_RETRY = 5  # seconds
 
+DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5
+
 
 class RunJobController(http.Controller):
 
@@ -36,6 +41,36 @@ 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/session', type='http', auth="none")
     def session(self):
         """Used by the jobrunner to spawn a session
@@ -130,6 +165,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):
@@ -146,11 +185,27 @@ 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
     ):
         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)
@@ -163,6 +218,31 @@ 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(
@@ -171,7 +251,53 @@ def create_test_job(
                 channel=channel,
                 description=description,
             )
-            ._test_job()
+            ._test_job(failure_rate=failure_rate)
         )
+        return 'job uuid: %s' % (delayed.db_record().uuid,)
 
-        return delayed.db_record().uuid
+    TEST_GRAPH_MAX_PER_GROUP = 5
+
+    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,
+        )