TL;DR

Celeryのデフォルトの設定ではCeleryでブローカーにSSL有りのRedisを使えません。
SSL有りのRedisを使うには次の設定が必要です。
- BROKER_USE_SSL: ブローカーとの接続とSSL設定のSSL使用を切り替えます。例: {'ssl_cert_reqs': 'none'}
- CELERY_REDIS_BACKEND_USE_SSL: Redisがブローカーの場合はこの設定も必要です。例: {'ssl_cert_reqs': ssl.CERT_CERT_REQUIRED (sslは標準ライブラリのssl)
Django + Celeryの設定例を https://github.com/TakesxiSximada/TIL/master/django/django-celery-redis-ssl に示しました。

背景

Azure Redis Cacheをスタンダードプランで使う場合、グローバルIPを割り振らなければアクセスできません。パスワードの設定などを行って使用することになるのですが、SSLも有効にする必要があります。このSSL有りのRedisをCeleryのブローカーに指定すると、Celeryのデフォルトの設定ではブローカーとの接続時にエラーします。Celeryのドキュメントには当然記述があるのですが、Celeryの設定項目はとてもたくさんありわかりづらいです。また解説自体が機能の説明になっているため、どのように使えばいいのか把握しづらい状態です。そこでこの記事では、使用するシーンとしてDjango + Celeryの組み合わせで使う場合を想定し、サンプルコードを元に解説を進めます。その中でCeleryのブローカーにSSL有りRedisを指定する方法を解説します。

環境

対象	バージョン
OS	macOS 10.12.4
Python	Python3.6.2
依存ライブラリ	requirementsを参照

requirements

実行には以下のライブラリを使用しました。こちらは明示的に使ったライブラリです。

celery
custom_settings
django
ipdb
redis

依存ライブラリを含む実際にインストールしたライブラリの一覧はこちらです。

#
# This file is autogenerated by pip-compile
# To update, run:
#
#    pip-compile --output-file requirements.txt requirements.in
#
amqp==2.2.2               # via kombu
appnope==0.1.0            # via ipython
billiard==3.5.0.3         # via celery
celery==4.1.0
custom-settings==1.0.post1
decorator==4.1.2          # via ipython, traitlets
django==1.11.6
ipdb==0.10.3
ipython-genutils==0.2.0   # via traitlets
ipython==6.2.1            # via ipdb
jedi==0.11.0              # via ipython
kombu==4.1.0              # via celery
parso==0.1.0              # via jedi
pexpect==4.2.1            # via ipython
pickleshare==0.7.4        # via ipython
prompt-toolkit==1.0.15    # via ipython
ptyprocess==0.5.2         # via pexpect
pygments==2.2.0           # via ipython
pytz==2017.2              # via celery, django
redis==2.10.6
simplegeneric==0.8.1      # via ipython
six==1.11.0               # via custom-settings, prompt-toolkit, traitlets
traitlets==4.3.2          # via ipython
vine==1.1.4               # via amqp
wcwidth==0.1.7            # via prompt-toolkit
zope.dottedname==4.2      # via custom-settings

Djangoプロジェクトの作成

Djangoプロジェクトを作成します。プロジェクト名はprojにします。

django-admin startproject proj .

celeryconfig.py

celeryconfig.py には Celeryの設定を記述します。まずはブローカー、返却値をを保持しておくためのバックエンド、データの受け渡しのための形式を指定します。これらはCeleryをつかう場合には必ず必要になります。ここで使用しているcustom_settingsは設定のためのヘルパーライブラリです。詳しくは https://pypi.python.org/pypi/custom_settings をご覧ください。

import custom_settings

custom = custom_settings.load('settings_custom')
BROKER_URL = custom.get('BROKER_URL', default='redis://:REDIS_PASS@localhost:6379/0')
CELERY_RESULT_BACKEND = custom.get('BROKER_URL', default='redis://:REDIS_PASS@localhost:6379/0')
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

続いてSSL接続のための設定です。 BROKER_USE_SSL と　CELERY_REDIS_BACKEND_USE_SSL を設定します。

import ssl

BROKER_USE_SSL = {'ssl_cert_reqs': 'none'}
CELERY_REDIS_BACKEND_USE_SSL = {'ssl_cert_reqs': ssl.CERT_REQUIRED}

全体はこのようになります。

# <CELERY-BASE>
import custom_settings

custom = custom_settings.load('settings_custom')
BROKER_URL = custom.get('BROKER_URL', default='redis://:REDIS_PASS@localhost:6379/0')
CELERY_RESULT_BACKEND = custom.get('BROKER_URL', default='redis://:REDIS_PASS@localhost:6379/0')
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
# </CELERY-BASE>


# <CELERY-SSL>
import ssl

BROKER_USE_SSL = {'ssl_cert_reqs': 'none'}
CELERY_REDIS_BACKEND_USE_SSL = {'ssl_cert_reqs': ssl.CERT_REQUIRED}
# </CELERY-SSL>

Celeryの準備

先ほど作成したprojプロジェクトでCeleryを使えるようにしていきます。まずは proj/celery_.py を作成して次のように記述します。

from __future__ import absolute_import

import os

from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')


# <CELERY-APP>
app = Celery('proj')
app.config_from_object('celeryconfig')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
# </CELERY-APP>


# <CELERY-TASK>
@app.task
def debug():
    print('OK')
# </CELERY-TASK>

Celeryオブジェクトを生成して設定をします。

app = Celery('proj')
app.config_from_object('celeryconfig')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

app.config_from_object(’celeryconfig') で設定を読み込んでいます。今回はCeleryの設定は celeryconfig.py に記述します。Djangoの場合は proj/settings.py に同様の設定をして、 app.config_from_obj('celeryconfig') の第一引数を 'django.conf:settings' にしても同じです。

検証用にタスクを一つ定義しておきます。このタスクは後ほど celery call コマンドで呼び出します。

@app.task
def debug():
    print('OK')

動作確認

実行する環境を作成します。venvなどはここでは割愛しますので適宜作成してください。依存ライブラリをインストールします。

pip install -r requirements.txt

サンプルコードはDjangoプロジェクトなのでマイグレーションを実行します。

python manage.py migrate

Celeryワーカーを起動します。ワーカー起動は -D を付けているのでバックグラウンドで実行します。 --logfile で指定したファイルにログを出力します。

celery -A proj.celery_app worker -D --pidfile .worker.pid --logfile worker.log

準備できました。 proj.celery_.debug をデバッグ用のタスクとして実装しています。タスクを実行します。

celery -A proj.celery_app call proj.celery_.debug

次のようなハッシュ値が表示されます。これはタスクがRedisに登録されてWorkerでの処理が終了した状態です。

$ celery -A proj.celery_app call proj.celery_.debug
b77a6165-4458-4540-a3a7-0824f4d6ca30

もしハッシュ値が表示されない場合はどこかにおかしな箇所があります。特に次の項目を確認してください。

Redisは起動しているか？
Redisへの接続情報は間違っていないか？
設定ファイルの記述は正しいか？

後片付け

ワーカーがまだ起動したままなので停止します。ワーカーの起動時に --pidfile で指定したファイルにPIDが出力されています。そのプロセスをkillコマンドでkillします。

kill `cat .worker.pid`