Further separate task queues and increase timeouts

Procfile

@@ -7,6 +7,6 @@ release: ./manage.py migrate --check || ./manage.py migrate
 # https://devcenter.heroku.com/articles/http-routing#http-validation-and-restrictions
 # long request lines are useful for long DSL search queries
 web: gunicorn --timeout 120 --limit-request-line 8192 --bind 0.0.0.0:$PORT isic.wsgi
-worker: REMAP_SIGTERM=SIGQUIT celery --app isic.celery worker --loglevel INFO --without-heartbeat --concurrency 2 -X low-priority
-low_priority_worker: REMAP_SIGTERM=SIGQUIT celery --app isic.celery worker --loglevel INFO --without-heartbeat --concurrency 2 -Q low-priority
+worker: REMAP_SIGTERM=SIGQUIT ./deploy/worker.sh
+low_priority_worker: REMAP_SIGTERM=SIGQUIT ./deploy/low-priority-worker.sh
 beat: REMAP_SIGTERM=SIGQUIT celery --app isic.celery beat --loglevel INFO

deploy/low-priority-worker.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+set -e
+
+celery --app isic.celery worker \
+    --loglevel INFO \
+    --without-heartbeat \
+    --concurrency 2 \
+    --queues s3-log-processing,stats-aggregation,es-indexing

deploy/worker.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+set -e
+
+celery --app isic.celery worker \
+    --loglevel INFO \
+    --without-heartbeat \
+    --concurrency 2 \
+    --queues celery

isic/core/tasks.py

@@ -51,12 +51,13 @@ def share_collection_with_users_task(collection_pk: int, grantor_pk: int, user_p
 
 
 @shared_task(
-    soft_time_limit=900,
-    time_limit=910,
+    soft_time_limit=1200,
+    time_limit=1210,
     autoretry_for=(ConnectionError, TimeoutError),
     retry_backoff=True,
     retry_backoff_max=600,
     retry_kwargs={"max_retries": 3},
+    queue="es-indexing",
 )
 def sync_elasticsearch_index_task():
     bulk_add_to_search_index(Image.objects.with_elasticsearch_properties())

isic/stats/tasks.py

@@ -161,7 +161,7 @@ def _cdn_access_log_records(log_file_bytes: BytesIO) -> Iterable[dict]:
             }
 
 
-@shared_task(queue="low-priority")
+@shared_task(queue="stats-aggregation")
 def collect_image_download_records_task():
     """
     Collect CDN logs to record image downloads.
@@ -181,7 +181,11 @@ def collect_image_download_records_task():
 
 
 @shared_task(
-    soft_time_limit=600, time_limit=630, max_retries=5, retry_backoff=True, queue="low-priority"
+    soft_time_limit=600,
+    time_limit=630,
+    max_retries=5,
+    retry_backoff=True,
+    queue="s3-log-processing",
 )
 def process_s3_log_file_task(s3_log_object_key: str):
     logger.info("Processing s3 log file %s", s3_log_object_key)