Updates

docker/local/local-config.sed

@@ -1,5 +1,5 @@
 s/CELERY_BROKER = .*/CELERY_BROKER = 'redis:\/\/redis'/g
-s/CELERY_BACKEND = .*/CELERY_BACKEND = 'redis:\/\/redis'/g
+s/CELERY_BACKEND = .*/CELERY_BACKEND = 'redis:\/\/redis\/1'/g
 s/REDIS_HOST = .*/REDIS_HOST = 'redis'/g
 s/PROMETHEUS_ENABLED = .*/PROMETHEUS_ENABLED = False/g
 s/STATE_MANAGER = .*/STATE_MANAGER = 'Redis'/g

turbinia/config/turbinia_config_tmpl.py

@@ -317,7 +317,7 @@
 CELERY_BROKER = f'redis://{REDIS_HOST}'
 
 # Storage for task results/status
-CELERY_BACKEND = f'redis://{REDIS_HOST}'
+CELERY_BACKEND = f'redis://{REDIS_HOST}/1'
 
 # Task expiration (in seconds). Tasks will be revoked
 # after the expiration time elapses. Revoked tasks will not

turbinia/tcelery.py

@@ -50,14 +50,18 @@ def setup(self):
     self.app = celery.Celery(
         'turbinia', broker=config.CELERY_BROKER, backend=config.CELERY_BACKEND)
     self.app.conf.update(
+        accept_content=['json'],
         broker_connection_retry_on_startup=True,
+        # Store Celery task results metadata
+        result_backend=config.CELERY_BACKEND,
         task_default_queue=config.INSTANCE_ID,
-        accept_content=['json'],
+        # Re-queue task if Celery worker abruptly exists
+        task_reject_on_worker_lost=True,
         worker_cancel_long_running_tasks_on_connection_loss=True,
         worker_concurrency=1,
         worker_prefetch_multiplier=1,
-        task_reject_on_worker_lost=True,  # Re-queue task if celery worker abruptly exists
-        worker_deduplicate_successful_tasks=True)  # avoid task duplication
+        # Avoid task duplication
+        worker_deduplicate_successful_tasks=True)
 
 
 class TurbiniaKombu(TurbiniaMessageBase):