Handle celery errors by re-running async task

CHANGELOG.md

@@ -5,6 +5,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic
 Versioning](https://semver.org/spec/v2.0.0.html).
 
+# [0.19.1] -
+### Fixed
+- [PR 14](/~https://github.com/salesforce/django-declarative-apis/pull/14) Retry async tasks if kombu.exceptions.OperationalError is encounterer
+- [PR 14](/~https://github.com/salesforce/django-declarative-apis/pull/14) Added DECLARATIVE_ENDPOINT_TASKS_FORCE_SYNCHRONOUS. This allows users to entirely skip asynchronous task processing if needed.
+- [PR 14](/~https://github.com/salesforce/django-declarative-apis/pull/14) Added DECLARATIVE_ENDPOINT_TASKS_SYNCHRONOUS_FALLBACK. This allows asynchronous tasks to automatically fall back to processing synchronously should asynchronous processing fail.
+
 # [0.19.0] - 2020-01-07
 ### Added
 - [PR 12](/~https://github.com/salesforce/django-declarative-apis/pull/12) Added support for logging correlation ids in deferred tasks.

django_declarative_apis/machinery/tasks.py

@@ -11,6 +11,7 @@
 import celery
 import time
 from typing import NamedTuple
+import kombu.exceptions
 
 from django.conf import settings
 from django.core.cache import cache
@@ -183,7 +184,29 @@ def schedule_future_task_runner(task_runner_args, task_runner_kwargs,
         countdown=countdown
     )
     task_runner_kwargs['correlation_id'] = _get_correlation_id()
-    future_task_runner.apply_async(task_runner_args, task_runner_kwargs, queue=queue, routing_key=routing_key, countdown=countdown+delay)
+
+    if settings.DECLARATIVE_ENDPOINT_TASKS_FORCE_SYNCHRONOUS:
+        logger.info('Processing tasks synchronously')
+        future_task_runner.apply(task_runner_args, task_runner_kwargs)
+    else:
+        MAX_ATTEMPTS = 3
+        for attempt in range(MAX_ATTEMPTS):
+            # XXX: This is an attempt to skirt around an unsolved, low repro issue somewhere in the celery/kombu/redis-py stack.
+            # Once in a while, a connection in the pool will timeout prior to a health check being called in redis-py and
+            # will result in an error being raised here. This should be removed once the issue has been sorted out.
+            # Note: This is around the use of redis-py in celery where celery's event loop is not running
+            # /~https://github.com/celery/kombu/issues/1019
+            try:
+                future_task_runner.apply_async(task_runner_args, task_runner_kwargs, queue=queue, routing_key=routing_key, countdown=countdown+delay)
+                return
+            except kombu.exceptions.OperationalError as err:
+                logger.warn('kombu.exceptions.OperationalError (attempt: %s)', attempt)
+                if attempt >= MAX_ATTEMPTS - 1:
+                    if settings.DECLARATIVE_ENDPOINT_TASKS_SYNCHRONOUS_FALLBACK:
+                        logger.warn('Falling back to executing task synchronously')
+                        future_task_runner.apply(task_runner_args, task_runner_kwargs)
+                        return
+                    raise err
 
 
 @celery_task(ignore_results=True,

setup.py

@@ -14,7 +14,7 @@
 
 setuptools.setup(
     name='django-declarative-apis',
-    version='0.19.0',
+    version='0.19.1',
     author='Drew Shafer',
     url='https://salesforce.com',
     description='Simple, readable, declarative APIs for Django',

tests/machinery/test_base.py

@@ -11,12 +11,15 @@
 import http
 import urllib.parse
 
+import kombu.exceptions
 import django.core.exceptions
 from django.db import models
 from django.core.cache import cache
+import django.test
 
 from django_declarative_apis import machinery, models as dda_models
 from django_declarative_apis.machinery import errors, filtering, tasks
+from django_declarative_apis.machinery.tasks import future_task_runner
 from django_declarative_apis.resources.utils import HttpStatusCode, rc
 
 import tests.models
@@ -740,7 +743,7 @@ def deferred_task_with_retry_and_filtered_exceptions_fail(inst):
             raise Exception("failing so task will retry")
 
 
-class DeferrableTaskTestCase(unittest.TestCase):
+class DeferrableTaskTestCase(django.test.TestCase):
     def setUp(self):
         _TestEndpoint.semaphore = {
             'status': None,
@@ -749,6 +752,132 @@ def setUp(self):
             'filtered_retry_count_2': 0
             }
 
+    def test_get_response_kombu_error_retried(self):
+        expected_response = {'foo': 'bar'}
+        endpoint = _TestEndpoint(expected_response)
+        manager = machinery.EndpointBinder.BoundEndpointManager(
+            machinery._EndpointRequestLifecycleManager(endpoint),
+            endpoint
+        )
+
+        conf = tasks.future_task_runner.app.conf
+        old_val = conf['task_always_eager']
+        conf['task_always_eager'] = True
+
+        cache.set(tasks.JOB_COUNT_CACHE_KEY, 0)
+
+        with mock.patch('django_declarative_apis.machinery.tasks.future_task_runner.apply_async') as mock_apply:
+            exceptions = iter([kombu.exceptions.OperationalError, kombu.exceptions.OperationalError])
+            def _side_effect(*args, **kwargs):
+                try:
+                    raise next(exceptions)
+                except StopIteration:
+                    return future_task_runner.apply(*args, **kwargs)
+            mock_apply.side_effect = _side_effect
+
+            try:
+                resp = manager.get_response()
+            finally:
+                conf['task_always_eager'] = old_val
+
+        self.assertEqual(resp, (http.HTTPStatus.OK, expected_response))
+        self.assertTrue(cache.get(tasks.JOB_COUNT_CACHE_KEY) != 0)
+
+        self.assertEqual('deferred task executed', _TestEndpoint.semaphore['status'])
+
+    def test_async_task_falls_back_to_synchronous_when_configured(self):
+        expected_response = {'foo': 'bar'}
+
+        endpoint = _TestEndpoint(expected_response)
+        manager = machinery.EndpointBinder.BoundEndpointManager(
+            machinery._EndpointRequestLifecycleManager(endpoint),
+            endpoint
+        )
+
+        conf = tasks.future_task_runner.app.conf
+        old_val = conf['task_always_eager']
+        conf['task_always_eager'] = True
+
+        with mock.patch('django_declarative_apis.machinery.tasks.future_task_runner.apply_async') as mock_apply_async:
+            mock_apply_async.side_effect = kombu.exceptions.OperationalError
+
+            cache.set(tasks.JOB_COUNT_CACHE_KEY, 0)
+
+            with self.settings(DECLARATIVE_ENDPOINT_TASKS_SYNCHRONOUS_FALLBACK=True):
+                try:
+                    resp = manager.get_response()
+                except kombu.exceptions.OperationalError:
+                    self.fail('OperationalError should not have been triggered')
+                finally:
+                    conf['task_always_eager'] = old_val
+
+        self.assertEqual('deferred task executed', _TestEndpoint.semaphore['status'])
+
+    def test_force_synchronous_tasks(self):
+        expected_response = {'foo': 'bar'}
+        endpoint = _TestEndpoint(expected_response)
+        manager = machinery.EndpointBinder.BoundEndpointManager(
+            machinery._EndpointRequestLifecycleManager(endpoint),
+            endpoint
+        )
+
+        conf = tasks.future_task_runner.app.conf
+        old_val = conf['task_always_eager']
+        conf['task_always_eager'] = True
+
+        cache.set(tasks.JOB_COUNT_CACHE_KEY, 0)
+
+        with mock.patch('django_declarative_apis.machinery.tasks.future_task_runner.apply_async') as mock_apply:
+            mock_apply.side_effect = kombu.exceptions.OperationalError
+
+            with self.settings(DECLARATIVE_ENDPOINT_TASKS_FORCE_SYNCHRONOUS=True):
+                try:
+                    resp = manager.get_response()
+                except kombu.exceptions.OperationalError:
+                    self.fail('OperationalError should not have been triggered')
+                finally:
+                    conf['task_always_eager'] = old_val
+
+        self.assertEqual(0, mock_apply.call_count)
+        self.assertEqual('deferred task executed', _TestEndpoint.semaphore['status'])
+
+    def test_get_response_kombu_error_attempts_exceeded(self):
+        expected_response = {'foo': 'bar'}
+        endpoint = _TestEndpoint(expected_response)
+        manager = machinery.EndpointBinder.BoundEndpointManager(
+            machinery._EndpointRequestLifecycleManager(endpoint),
+            endpoint
+        )
+
+        conf = tasks.future_task_runner.app.conf
+        old_val = conf['task_always_eager']
+        conf['task_always_eager'] = True
+
+        cache.set(tasks.JOB_COUNT_CACHE_KEY, 0)
+
+        with mock.patch('django_declarative_apis.machinery.tasks.future_task_runner.apply_async') as mock_apply:
+            exceptions = iter([
+                kombu.exceptions.OperationalError,
+                kombu.exceptions.OperationalError,
+                kombu.exceptions.OperationalError,
+            ])
+            def _side_effect(*args, **kwargs):
+                try:
+                    raise next(exceptions)
+                except StopIteration:
+                    return future_task_runner.apply(*args, **kwargs)
+            mock_apply.side_effect = _side_effect
+
+            try:
+                resp = manager.get_response()
+                self.fail('should have triggered a kombu.exceptions.OperationalError')
+            except kombu.exceptions.OperationalError:
+                pass
+            finally:
+                conf['task_always_eager'] = old_val
+
+        self.assertIsNone(_TestEndpoint.semaphore['status'])
+
     def test_get_response_success(self):
         expected_response = {'foo': 'bar'}
 

tests/settings.py

@@ -42,3 +42,5 @@
         'django_declarative_apis.authentication.oauthlib.oauth1.TwoLeggedOauth1',
     ),
 )
+DECLARATIVE_ENDPOINT_TASKS_FORCE_SYNCHRONOUS = False
+DECLARATIVE_ENDPOINT_TASKS_SYNCHRONOUS_FALLBACK = False