Merge branch 'hotfix/2.6.3'

Author: chrisseto

project/settings.py

@@ -322,6 +322,9 @@
 ELASTICSEARCH_TIMEOUT = int(os.environ.get('ELASTICSEARCH_TIMEOUT', '45'))
 ELASTICSEARCH_INDEX_VERSIONS = tuple(v for v in os.environ.get('ELASTICSEARCH_INDEX_VERSIONS', '').split(',') if v)
 
+# Seconds, not an actual celery settings
+CELERY_RETRY_BACKOFF_BASE = int(os.environ.get('CELERY_RETRY_BACKOFF_BASE', 2 if DEBUG else 10))
+
 # Celery Settings
 
 BROKER_URL = os.environ.get('BROKER_URL', 'amqp://'),

share/tasks.py

@@ -1,8 +1,8 @@
 import abc
-import random
 import datetime
 import functools
 import logging
+import random
 import threading
 
 import pendulum
@@ -309,12 +309,18 @@ def do_run(self, start=None, end=None, limit=None, force=False, superfluous=Fals
                 # if not created and log.task_id != self.task_id and log.date_modified - datetime.datetime.utcnow() > datetime.timedelta(minutes=10):
                 #     pass
                 log.reschedule()
+                # Use random to add jitter to help break up locking issues
                 # Kinda hacky, allow a stupidly large number of retries as there is no options for infinite
-                raise self.retry(countdown=random.randrange(10, 30), exc=e, max_retries=99999)
+                raise self.retry(
+                    exc=e,
+                    max_retries=99999,
+                    countdown=min((random.random() + 1) * (settings.CELERY_RETRY_BACKOFF_BASE ** self.request.retries), 60 * 15)
+                )
             except Exception as e:
                 log.fail(e)
                 logger.exception('Failed harvester task (%r, %s, %s)', self.config, start, end)
-                raise self.retry(countdown=10, exc=e)
+                # No log needed here, max retries is 3
+                raise self.retry(countdown=min(settings.CELERY_RETRY_BACKOFF_BASE ** self.request.retries, 60 * 15), exc=e)
 
             if force and error:
                 log.forced(error)