Bug 1278858 - Add retryable_task decorator that provides the ability to not retry for given exceptions (#1574)

tests/test_worker/test_task.py

@@ -0,0 +1,27 @@
+import pytest
+
+from treeherder.workers.task import retryable_task
+
+
+@retryable_task()
+def successful_task(x, y):
+    return x + y
+
+
+def test_retryable_task():
+    "Test celery executes a task properly"
+
+    result = successful_task.delay(7, 3)
+    assert result.wait() == 10
+
+
+@retryable_task()
+def throwing_task(x, y):
+    raise TypeError
+
+
+def test_retryable_task_throws():
+    "Test celery executes a task properly"
+
+    with pytest.raises(TypeError):
+        throwing_task.delay(7, 3)

treeherder/autoclassify/tasks.py

@@ -1,36 +1,30 @@
 import logging
 
 import newrelic.agent
-from celery import task
 from django.core.management import call_command
 
 from treeherder import celery_app
+from treeherder.workers.task import retryable_task
 
 logger = logging.getLogger(__name__)
 
 
-@task(name='autoclassify', max_retries=10)
+@retryable_task(name='autoclassify', max_retries=10)
 def autoclassify(project, job_guid):
     newrelic.agent.add_custom_parameter("project", project)
     newrelic.agent.add_custom_parameter("job_guid", job_guid)
-    try:
-        logger.info('Running autoclassify')
-        call_command('autoclassify', project, job_guid)
-        celery_app.send_task('detect-intermittents',
-                             [project, job_guid],
-                             routing_key='detect_intermittents')
-    except Exception as e:
-        autoclassify.retry(exc=e, countdown=(1 + autoclassify.request.retries) * 60)
+    logger.info('Running autoclassify')
+    call_command('autoclassify', project, job_guid)
+    celery_app.send_task('detect-intermittents',
+                         [project, job_guid],
+                         routing_key='detect_intermittents')
 
 
-@task(name='detect-intermittents', max_retries=10)
+@retryable_task(name='detect-intermittents', max_retries=10)
 def detect_intermittents(project, job_guid):
     newrelic.agent.add_custom_parameter("project", project)
     newrelic.agent.add_custom_parameter("job_guid", job_guid)
-    try:
-        logger.info('Running detect intermittents')
-        # TODO: Make this list configurable
-        if project == "mozilla-inbound":
-            call_command('detect_intermittents', project, job_guid)
-    except Exception as e:
-        detect_intermittents.retry(exc=e, countdown=(1 + detect_intermittents.request.retries) * 60)
+    logger.info('Running detect intermittents')
+    # TODO: Make this list configurable
+    if project == "mozilla-inbound":
+        call_command('detect_intermittents', project, job_guid)

treeherder/workers/task.py

@@ -0,0 +1,27 @@
+import random
+from functools import wraps
+
+from celery import task
+from django.db.utils import IntegrityError
+
+
+class retryable_task(object):
+    def __init__(self, *args, **kwargs):
+        self.raise_exceptions = kwargs.pop("raise_exceptions", (TypeError, IntegrityError))
+        self.task_args = args
+        self.task_kwargs = kwargs
+
+    def __call__(self, f):
+        @wraps(f)
+        def inner(*args, **kwargs):
+            try:
+                return f(*args, **kwargs)
+            except self.raise_exceptions:
+                raise
+            except Exception as e:
+                # Implement exponential backoff with some randomness to prevent
+                # thundering herd type problems. Constant factor chosen so we get
+                # reasonable pause between the fastest retries.
+                f.retry(exc=e, countdown=10 * int(random.uniform(2, 3) ** f.request.retries))
+
+        return task(*self.task_args, **self.task_kwargs)(inner)