Attempt to use AMQP again, but with CloudAMQP's settings. (#823)

I originally tried using AMQP in #819 but this used an exorbitant number of messages. However, I just found a CloudAMQP post about running Celery which includes some settings that reduce the number of messages processed by RabbitMQ. Given the fact that our current broker, AWS SQS, is labeled as "experimental" in some places and might not be very reliable, we're switching back to AMQP, but with vastly lower message use.

Dockerfile.worker

@@ -4,4 +4,7 @@ ARG DOCKERFILE_WEB
 
 FROM $DOCKERFILE_WEB
 
-CMD celery worker -l warning -A project
+# Note that we are running Celery without a heartbeat as
+# per https://www.cloudamqp.com/docs/celery.html, to reduce
+# messaging costs when using AMQP.
+CMD celery worker -l warning -A project --without-heartbeat

README.md

@@ -419,15 +419,4 @@ For example, to start up all services with Celery integration enabled, you can r
 docker-compose -f docker-compose.yml -f docker-compose.celery.yml up
 ```
 
-You will also need to add the following to your `.justfix-env`:
-
-```
-AWS_ACCESS_KEY_ID=<your access key ID>
-AWS_SECRET_ACCESS_KEY=<your secret access key>
-JUSTFIX_CELERY_BROKER_URL=justfix-sqs:///?queue_name_prefix=throwaway-dev-prefix-
-```
-
-You should change `throwaway-dev-prefix-` to something like your name, as it
-will be used as the prefix for the SQS queue used by your development instance.
-
 [Multiple Compose files]: https://docs.docker.com/compose/extends/

docker-compose.celery.yml

@@ -1,13 +1,22 @@
-# Remember to set JUSTFIX_CELERY_BROKER_URL in your .justfix-env
-# when using this file!
 version: '2'
 services:
+  app:
+    environment:
+      - JUSTFIX_CELERY_BROKER_URL=amqp://rabbitmq
+    links:
+      - db
+      - rabbitmq
+  rabbitmq:
+    image: rabbitmq:3.7.17-alpine
   celery:
     extends:
       file: docker-services.yml
       service: base_app
+    environment:
+      - JUSTFIX_CELERY_BROKER_URL=amqp://rabbitmq
     volumes:
       - .:/tenants2:delegated
     links:
         - db
+        - rabbitmq
     command: python manage.py runcelery

project/management/commands/runcelery.py

@@ -11,7 +11,7 @@
 def restart_celery():
     cmd = 'pkill -9 celery'
     subprocess.call(shlex.split(cmd))
-    cmd = f'celery worker -l info --quiet -A {project.__name__}'
+    cmd = f'celery worker -l info --quiet -A {project.__name__} --without-heartbeat'
     subprocess.call(shlex.split(cmd))
 
 

project/settings.py

@@ -437,10 +437,41 @@
     default_queue_name_prefix="tenants2-"
 )
 CELERY_RESULT_BACKEND = 'django-db'
+
+if CELERY_BROKER_URL.startswith('amqp://'):
+    # By default, using Celery with AMQP consumes *tons* of messages,
+    # which quickly becomes expensive with services like CloudAMQP
+    # that charge based on message usage. The following settings are
+    # taken from https://www.cloudamqp.com/docs/celery.html to reduce
+    # message usage and increase efficiency.
+
+    # This will decrease connection usage.
+    CELERY_BROKER_POOL_LIMIT = 1
+
+    # We're using TCP keep-alive instead.
+    CELERY_BROKER_HEARTBEAT = None
+
+    # This may require a long timeout due to Linux DNS timeouts etc.
+    CELERY_BROKER_CONNECTION_TIMEOUT = 30
+
+    # This will delete all celeryev. queues without consumers after 1 minute.
+    CELERY_EVENT_QUEUE_EXPIRES = 60
+
+    # Disable prefetching, it causes problems and doesn't help performance.
+    CELERY_WORKER_PREFETCH_MULTIPLIER = 1
+
+# Our tasks are generally network-bound rather than CPU-bound, so we'll
+# increase concurrency substantially.
+CELERY_WORKER_CONCURRENCY = 5
+
+# We want to use Django logging.
 CELERY_WORKER_HIJACK_ROOT_LOGGER = False
+
+# When executing tasks synchronously, make sure exceptions propagate.
 CELERY_TASK_EAGER_PROPAGATES = True
 
 if not CELERY_BROKER_URL:
+    # If Celery integration is disabled, just execute tasks synchronously.
     CELERY_TASK_ALWAYS_EAGER = True
 
 if AWS_STORAGE_STATICFILES_BUCKET_NAME: