[IMP] queue_job: add error_handler

Author: trisdoan

queue_job/README.rst

@@ -74,6 +74,7 @@ Features:
   description, number of retries
 * Related Actions: link an action on the job view, such as open the record
   concerned by the job
+* Error Handler: trigger a method when job fails, such as calling a webhook
 
 **Table of contents**
 
@@ -429,6 +430,21 @@ Based on this configuration, we can tell that:
 * retries 10 to 15 postponed 30 seconds later
 * all subsequent retries postponed 5 minutes later
 
+**Job function: Error Handler**
+
+The *Error Handler* is a method executed whenever the job fails
+
+It's configured similarly to Related Action
+
+There is an OOTB handler: _call_webhook, which calls a webhook with configurable information.
+
+Example of using _call_webhook to call a webhook to Slack:
+
+.. code-block:: xml
+
+    <field name="error_handler" eval='{"func_name": "_call_webhook", "kwargs": {"webhook_url": "XXX", "only_if_max_retries_reached":True, "payload": {"text": "Hello World!"}}}' />
+
+
 **Job Context**
 
 The context of the recordset of the job, or any recordset passed in arguments of
@@ -687,6 +703,7 @@ Contributors
 * Souheil Bejaoui <souheil.bejaoui@acsone.eu>
 * Eric Antones <eantones@nuobit.com>
 * Simone Orsi <simone.orsi@camptocamp.com>
+* Tris Doan <tridm@trobz.com>
 
 Maintainers
 ~~~~~~~~~~~

queue_job/controllers/main.py

@@ -33,7 +33,14 @@ def _try_perform_job(self, env, job):
         env.cr.commit()
         _logger.debug("%s started", job)
 
-        job.perform()
+        try:
+            job.perform()
+        except Exception as exc:
+            with registry(job.env.cr.dbname).cursor() as new_cr:
+                job.env = job.env(cr=new_cr)
+                job.error_handler(exc)
+            raise
+
         # Triggers any stored computed fields before calling 'set_done'
         # so that will be part of the 'exec_time'
         env.flush_all()

queue_job/job.py

@@ -891,6 +891,26 @@ def related_action(self):
         action_kwargs = self.job_config.related_action_kwargs
         return action(**action_kwargs)
 
+    def error_handler(self, exc):
+        record = self.db_record()
+        funcname = self.job_config.error_handler_func_name
+        if not self.job_config.error_handler_enable or not funcname:
+            return None
+
+        if not isinstance(funcname, str):
+            raise ValueError(
+                "error_handler must be the name of the method on queue.job as string"
+            )
+        action = getattr(record, funcname)
+        _logger.info("Job %s fails due to %s, execute %s", self.uuid, exc, action)
+        action_kwargs = self.job_config.error_handler_kwargs
+        action_kwargs["job"] = self
+        action_kwargs["job"] = self
+        try:
+            return action(**action_kwargs)
+        except Exception as exc:
+            _logger.warning("Error handler failed: %s", exc)
+
 
 def _is_model_method(func):
     return inspect.ismethod(func) and isinstance(

queue_job/models/queue_job.py

@@ -1,10 +1,13 @@
 # Copyright 2013-2020 Camptocamp SA
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 
+import json
 import logging
 import random
 from datetime import datetime, timedelta
 
+import requests
+
 from odoo import _, api, exceptions, fields, models
 from odoo.osv import expression
 from odoo.tools import config, html_escape
@@ -506,3 +509,31 @@ def _test_job(self, failure_rate=0):
         _logger.info("Running test job.")
         if random.random() <= failure_rate:
             raise JobError("Job failed")
+
+    def _call_webhook(self, **kwargs):
+        only_if_max_retries_reached = kwargs.get("only_if_max_retries_reached", False)
+        job = kwargs.get("job")
+        if only_if_max_retries_reached and (job and job.retry < job.max_retries):
+            return
+
+        webhook_url = kwargs.get("webhook_url", None)
+        if not webhook_url:
+            return
+        payload = kwargs.get("payload", None)
+        json_values = json.dumps(payload, sort_keys=True, default=str)
+        headers = kwargs.get("headers", {"Content-Type": "application/json"})
+        # inspired by https://github.com/odoo/odoo/blob/18.0/odoo/addons/base
+        # /models/ir_actions.py#L867
+        try:
+            response = requests.post(
+                url=webhook_url, data=json_values, headers=headers, timeout=1
+            )
+            response.raise_for_status()
+        except requests.exceptions.ReadTimeout:
+            _logger.warning(
+                "Webhook call timed out after 1s - it may or may not have failed. "
+                "If this happens often, it may be a sign that the system you're "
+                "trying to reach is slow or non-functional."
+            )
+        except requests.exceptions.RequestException as exc:
+            _logger.warning("Webhook call failed: %s", exc)

queue_job/models/queue_job_function.py

@@ -28,6 +28,9 @@ class QueueJobFunction(models.Model):
         "related_action_enable "
         "related_action_func_name "
         "related_action_kwargs "
+        "error_handler_enable "
+        "error_handler_func_name "
+        "error_handler_kwargs "
         "job_function_id ",
     )
 
@@ -79,6 +82,33 @@ def _default_channel(self):
         "enable, func_name, kwargs.\n"
         "See the module description for details.",
     )
+    error_handler = JobSerialized(base_type=dict)
+    edit_error_handler = fields.Text(
+        string="Error Handler",
+        compute="_compute_edit_error_handler",
+        inverse="_inverse_edit_error_handler",
+        help="The handler is executed when the job fails. "
+        "Configured as a dictionary with optional keys: "
+        "enable, func_name, kwargs.\n"
+        "See the module description for details.",
+    )
+
+    @api.depends("error_handler")
+    def _compute_edit_error_handler(self):
+        for record in self:
+            record.edit_error_handler = str(record.error_handler)
+
+    def _inverse_edit_error_handler(self):
+        try:
+            edited = (self.edit_error_handler or "").strip()
+            if edited:
+                self.error_handler = ast.literal_eval(edited)
+            else:
+                self.error_handler = {}
+        except (ValueError, TypeError, SyntaxError) as ex:
+            raise exceptions.UserError(
+                self._error_handler_format_error_message()
+            ) from ex
 
     @api.depends("model_id.model", "method")
     def _compute_name(self):
@@ -149,6 +179,9 @@ def job_default_config(self):
             related_action_func_name=None,
             related_action_kwargs={},
             job_function_id=None,
+            error_handler_enable=True,
+            error_handler_func_name=None,
+            error_handler_kwargs={},
         )
 
     def _parse_retry_pattern(self):
@@ -182,6 +215,9 @@ def job_config(self, name):
             related_action_func_name=config.related_action.get("func_name"),
             related_action_kwargs=config.related_action.get("kwargs", {}),
             job_function_id=config.id,
+            error_handler_enable=config.error_handler.get("enable", True),
+            error_handler_func_name=config.error_handler.get("func_name"),
+            error_handler_kwargs=config.error_handler.get("kwargs", {}),
         )
 
     def _retry_pattern_format_error_message(self):
@@ -215,6 +251,14 @@ def _related_action_format_error_message(self):
             ' "kwargs" {{"limit": 10}}}}'
         ).format(self.name)
 
+    def _error_handler_format_error_message(self):
+        return _(
+            "Unexpected format of Error Handler for {}.\n"
+            "Example of valid format:\n"
+            '{{"enable": True, "func_name": "_call_webhook",'
+            ' "kwargs" {"webhook_url": "XXX","payload": {"text":"Hello World!"}}}}'
+        ).format(self.name)
+
     @api.constrains("related_action")
     def _check_related_action(self):
         valid_keys = ("enable", "func_name", "kwargs")

queue_job/readme/CONTRIBUTORS.rst

@@ -10,3 +10,4 @@
 * Souheil Bejaoui <[email protected]>
 * Eric Antones <[email protected]>
 * Simone Orsi <[email protected]>
+* Tris Doan <[email protected]>

queue_job/readme/DESCRIPTION.rst

@@ -44,3 +44,4 @@ Features:
   description, number of retries
 * Related Actions: link an action on the job view, such as open the record
   concerned by the job
+* Error Handler: trigger a method when job fails, such as calling a webhook

queue_job/readme/USAGE.rst

@@ -274,6 +274,21 @@ Based on this configuration, we can tell that:
 * retries 10 to 15 postponed 30 seconds later
 * all subsequent retries postponed 5 minutes later
 
+**Job function: Error Handler**
+
+The *Error Handler* is a method executed whenever the job fails
+
+It's configured similarly to Related Action
+
+There is an OOTB handler: _call_webhook, which calls a webhook with configurable information.
+
+Example of using _call_webhook to call a webhook to Slack:
+
+.. code-block:: xml
+
+    <field name="error_handler" eval='{"func_name": "_call_webhook", "kwargs": {"webhook_url": "XXX", "only_if_max_retries_reached":True, "payload": {"text": "Hello World!"}}}' />
+
+
 **Job Context**
 
 The context of the recordset of the job, or any recordset passed in arguments of

queue_job/static/description/index.html

@@ -408,6 +408,7 @@ <h1 class="title">Job Queue</h1>
 description, number of retries</li>
 <li>Related Actions: link an action on the job view, such as open the record
 concerned by the job</li>
+<li>Error Handler: trigger a method when job fails, such as calling a webhook</li>
 </ul>
 <p><strong>Table of contents</strong></p>
 <div class="contents local topic" id="contents">
@@ -744,6 +745,14 @@ <h3><a class="toc-backref" href="#toc-entry-7">Configure default options for job
 <li>retries 10 to 15 postponed 30 seconds later</li>
 <li>all subsequent retries postponed 5 minutes later</li>
 </ul>
+<p><strong>Job function: Error Handler</strong></p>
+<p>The <em>Error Handler</em> is a method executed whenever the job fails</p>
+<p>It’s configured similarly to Related Action</p>
+<p>There is an OOTB handler: _call_webhook, which calls a webhook with configurable information.</p>
+<p>Example of using _call_webhook to call a webhook to Slack:</p>
+<pre class="code xml literal-block">
+<span class="nt">&lt;field</span><span class="w"> </span><span class="na">name=</span><span class="s">&quot;error_handler&quot;</span><span class="w"> </span><span class="na">eval=</span><span class="s">'{&quot;func_name&quot;: &quot;_call_webhook&quot;, &quot;kwargs&quot;: {&quot;webhook_url&quot;: &quot;XXX&quot;, &quot;only_if_max_retries_reached&quot;:True, &quot;payload&quot;: {&quot;text&quot;: &quot;Hello World!&quot;}}}'</span><span class="w"> </span><span class="nt">/&gt;</span>
+</pre>
 <p><strong>Job Context</strong></p>
 <p>The context of the recordset of the job, or any recordset passed in arguments of
 a job, is transferred to the job according to an allow-list.</p>
@@ -982,6 +991,7 @@ <h2><a class="toc-backref" href="#toc-entry-17">Contributors</a></h2>
 <li>Souheil Bejaoui &lt;<a class="reference external" href="mailto:souheil.bejaoui&#64;acsone.eu">souheil.bejaoui&#64;acsone.eu</a>&gt;</li>
 <li>Eric Antones &lt;<a class="reference external" href="mailto:eantones&#64;nuobit.com">eantones&#64;nuobit.com</a>&gt;</li>
 <li>Simone Orsi &lt;<a class="reference external" href="mailto:simone.orsi&#64;camptocamp.com">simone.orsi&#64;camptocamp.com</a>&gt;</li>
+<li>Tris Doan &lt;<a class="reference external" href="mailto:tridm&#64;trobz.com">tridm&#64;trobz.com</a>&gt;</li>
 </ul>
 </div>
 <div class="section" id="maintainers">

queue_job/tests/test_model_job_function.py

@@ -52,6 +52,9 @@ def test_function_job_config(self):
                 related_action_enable=True,
                 related_action_func_name="related_action_foo",
                 related_action_kwargs={"b": 1},
+                error_handler_enable=True,
+                error_handler_func_name=None,
+                error_handler_kwargs={},
                 job_function_id=job_function.id,
             ),
         )