Introduction
Has your users complained about the loading issue on the web app you developed. That might be because of some long I/O bound call or a time consuming process. For example, when a customer signs up to website and we need to send confirmation email which in normal case the email will be sent and then reply 200 OK response is sent on signup POST. However we can send email later, after sending 200 OK response, right?. This is not so straight forward when you are working with a framework like Django, which is tightly binded to MVC paradigm.
So, how do we do it ? The very first thought in mind would be python threading module. Well, Python threads are implemented as pthreads (kernel threads), and because of the global interpreter lock (GIL), a Python process only runs one thread at a time. And again threads are hard to manage, maintain code and scale it.
Perequisite
Audience for this blog requires to have knowledge about Django and AWS elastic beanstalk.
Celery
Celery is here to rescue. It can help when you have a time consuming task (heavy compute or I/O bound tasks) between request-response cycle. Celery is an open source asynchronous task queue or job queue which is based on distributed message passing. In this post I will walk you through the celery setup procedure with django and SQS on elastic beanstalk.
Why Celery ?
Celery is very easy to integrate with existing code base. Just write a decorator above the definition of a function declaring a celery task and call that function with a .delay method of that function.
1 2 3 4 5 6 7 |
from celery import Celery app = Celery('hello', broker='amqp://guest@localhost//') @app.task def hello(): return 'hello world' |
1 2 |
# Calling a celery task hello.delay() |
Broker
To work with celery, we need a message broker. As of writing this blog, Celery supports RabbitMQ, Redis, and Amazon SQS (not fully) as message broker solutions. Unless you don’t want to stick to AWS ecosystem (as in my case), I recommend to go with RabbitMQ or Redis because SQS does not yet support remote control commands and events. For more info check here. One of the reason to use SQS is its pricing. One million SQS free request per month for every user.
Proceeding with SQS, go to AWS SQS dashboard and create a new SQS queues. Click on create new queue button.
Depending upon the requirement we can select any type of the queue. We will name queue as dev-celery.
Installation
Celery has a very nice documentation. Installation and configuration is described here. For convenience here are the steps
Activate your virtual environment, if you have configured one and install cerely.
pip install celery[sqs]
Configuration
Celery has built-in support of django. It will pick its setting parameter from django’s settings.py which are prepended by CELERY_ (‘CELERY’ word needs to be defined while initializing celery app as namespace). So put below setting parameter in settings.py
1 2 |
# Amazon credentials will be taken from environment variable. CELERY_BROKER_URL = 'sqs://' |
AWS login credentials should be present in the environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY
1 2 3 4 5 6 7 |
CELERY_BROKER_TRANSPORT_OPTIONS = {'region': 'us-west-2', 'visibility_timeout': 3600, 'polling_interval': 10, 'queue_name_prefix': '%s-' % {True: 'dev', False: 'production'}[DEBUG], 'CELERYD_PREFETCH_MULTIPLIER': 0, } |
Now let’s configure celery app within django code. Create a celery.py file besides django’s settings.py.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
from __future__ import absolute_import, unicode_literals import os from celery import Celery # set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') app = Celery('proj') # Using a string here means the worker doesn't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django app configs. app.autodiscover_tasks() @app.task(bind=True) def debug_task(self): print('Request: {0!r}'.format(self.request)) |
Now put below code in projects __init__.py
1 2 3 4 5 6 7 |
from __future__ import absolute_import, unicode_literals # This will make sure the app is always imported when # Django starts so that shared_task will use this app. from .celery import app as celery_app __all__ = ('celery_app',) |
Testing
Now let’s test the configuration. Open terminal start celery
Terminal 1
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
$ celery worker --app=proj --loglevel=INFO -------------- celery@lintel v4.1.0 (latentcall) ---- **** ----- --- * *** * -- Linux-4.15.0-24-generic-x86_64-with-Ubuntu-18.04-bionic 2018-07-04 11:18:57 -- * - **** --- - ** ---------- [config] - ** ---------- .> app: enq_web:0x7f0ba29fa3d0 - ** ---------- .> transport: sqs://localhost// - ** ---------- .> results: disabled:// - *** --- * --- .> concurrency: 4 (prefork) -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker) --- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery [tasks] . enq_web._celery.debug_task |
All the task which are registered to use celery using celery decorators appear here while starting celery. If you find that your task does not appear here then make sure that the module containing the task is imported on startup.
Now open django shell in another terminal
Terminal 2
1 2 3 4 5 |
$ python manage.py shell In [1]: from proj import celery In [2]: celery.debug_task() # ←← ← Not through celery In [3]: celery.debug_task.delay() # ←← ← This is through celery |
After executing the task function with delay method, that task should run in the worker process which is listening to events in other terminal. Here celery sent a message to SQS with details of the task and worker process which was listening to SQS, received it and task was executed in worker process. Below is what you should see in terminal 1
Terminal 1
1 |
Request: <Context: {'origin': 'gen14099@lintel', u'args': [], 'chain': None, 'root_id': '041be6c3-419d-4aa0-822f-d50da1b340a0', 'expires': None, u'is_eager': False, u'correlation_id': '041be6c3-419d-4aa0-822f-d50da1b340a0', 'chord': None, u'reply_to': 'd2e76b9b-094b-33b4-a873-db5d2ace8881', 'id': '041be6c3-419d-4aa0-822f-d50da1b340a0', 'kwargsrepr': '{}', 'lang': 'py', 'retries': 0, 'task': 'proj.celery.debug_task', 'group': None, 'timelimit': [None, None], u'delivery_info': {u'priority': 0, u'redelivered': None, u'routing_key': 'celery', u'exchange': u''}, u'hostname': u'celery@lintel', 'called_directly': False, 'parent_id': None, 'argsrepr': '()', 'errbacks': None, 'callbacks': None, u'kwargs': {}, 'eta': None, '_protected': 1}> |
Deploy celery worker process on AWS elastic beanstalk
Celery provides “multi” sub command to run process in daemon mode, but this cannot be used on production. Celery recommends various daemonization tools http://docs.celeryproject.org/en/latest/userguide/daemonizing.html
AWS elastic beanstalk already use supervisord for managing web server process. Celery can also be configured using supervisord tool. Celery’s official documentation has a nice example of supervisord config for celery. https://github.com/celery/celery/tree/master/extra/supervisord. Based on that we write quite a few commands under .ebextensions directory.
Create two files under .ebextensions directory. Celery.sh file extract the environment variable and forms celery configuration, which copied to /opt/python/etc/celery.conf file and supervisord is restarted. Here main celery command:
1 |
celery worker -A PROJECT_NAME -P solo --loglevel=INFO -n worker.%%h. |
At the time if writing this blog celery had https://github.com/celery/celery/issues/3759 issue. As a work around to this issue we add “-P solo”. This will run task sequentially for a single worker process.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
#!/usr/bin/env bash # Get django environment variables celeryenv=`cat /opt/python/current/env | tr '\n' ',' | sed 's/export //g' | sed 's/$PATH/%(ENV_PATH)s/g' | sed 's/$PYTHONPATH//g' | sed 's/$LD_LIBRARY_PATH//g'` celeryenv=${celeryenv%?} # Create celery configuraiton script celeryconf="[program:celeryd-worker] ; Set full path to celery program if using virtualenv command=/opt/python/run/venv/bin/celery worker -A PROJECT_NAME -P solo --loglevel=INFO -n worker.%%h directory=/opt/python/current/app/enq_web user=nobody numprocs=1 stdout_logfile=/var/log/celery/worker.log stderr_logfile=/var/log/celery/worker.log autostart=true autorestart=true startsecs=10 ; Need to wait for currently executing tasks to finish at shutdown. ; Increase this if you have very long running tasks. stopwaitsecs = 600 ; When resorting to send SIGKILL to the program to terminate it ; send SIGKILL to its whole process group instead, ; taking care of its children as well. killasgroup=true ; if rabbitmq is supervised, set its priority higher ; so it starts first priority=998 environment=$celeryenv " # Create the celery supervisord conf script echo "$celeryconf" | tee /opt/python/etc/celery.conf # Add configuration script to supervisord conf (if not there already) if ! grep -Fxq "[include]" /opt/python/etc/supervisord.conf then echo "[include]" | tee -a /opt/python/etc/supervisord.conf echo "files: celery.conf" | tee -a /opt/python/etc/supervisord.conf fi # Reread the supervisord config /usr/local/bin/supervisorctl -c /opt/python/etc/supervisord.conf reread # Update supervisord in cache without restarting all services /usr/local/bin/supervisorctl -c /opt/python/etc/supervisord.conf update # Start/Restart celeryd through supervisord /usr/local/bin/supervisorctl -c /opt/python/etc/supervisord.conf restart celeryd-worker |
Now create elastic beanstalk configuration file as below. Make sure you have pycurl and celery in requirements.txt. To install pycurl libcurl-devel needs to be installed from yum package manager.
1 2 3 4 5 6 7 8 9 10 11 12 |
packages: yum: libcurl-devel: [] container_commands: 01_mkdir_for_log_and_pid: command: "mkdir -p /var/log/celery/ /var/run/celery/" 02_celery_configure: command: "cp .ebextensions/celery-worker.sh /opt/elasticbeanstalk/hooks/appdeploy/post/ && chmod 744 /opt/elasticbeanstalk/hooks/appdeploy/post/celery-worker.sh" cwd: "/opt/python/ondeck/app" 03_celery_run: command: "/opt/elasticbeanstalk/hooks/appdeploy/post/celery-worker.sh" |
Add these files to git and deploy to elastic beanstalk.
Below is the figure describing the architecture with django, celery and elastic beanstalk.