DjangoでCelery+Redisのジョブキューを構築したときにハマったポイント

こんにちはカカドゥの増田です。

Djangoでジョブキューを使おうと思って、Celery(セロリ)+Redisで構築しようとしたら結構ハマったので記録しておきます。

環境

	バージョン
Ubuntu	15.04
Python	2.7.10
Celery	3.1.18
Redis	2.8.19

ファイルレイアウト

- project/
  - project/__init__.py
  - project/settings.py
  - project/urls.py
  - project/celery.py
- manage.py

発生した問題

• アプリケーション側からは問題なくキューが入ったように見えるのに、Redisにキューが入らない
• Redisを試す前にRabbitMQを試していたことが問題を複雑にしていた

原因

公式ドキュメントの説明に従って、project/__init__.pyを作成していなかったから。

from __future__ import absolute_import

# 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

解説

Djangoのproject/__init__.pyの中でfrom celery import appしなかった場合、fallbackの仕組みとしてBROKER_URLにamqp://guest:**@localhost:5672//を設定した状態として動作します。

そのあたりの実装は、以下のあたりを読むとわかります。

celery/_state.py at 43ef0321058f318310cb0abd994b82047a25751e · celery/celery · GitHub

def _get_current_app():
    if default_app is None:
        #: creates the global fallback app instance.
        from celery.app import Celery
        set_default_app(Celery(
            'default', fixups=[], set_as_current=False,
            loader=os.environ.get('CELERY_LOADER') or 'default',
        ))
    return _tls.current_app or default_app

celery/base.py at 43ef0321058f318310cb0abd994b82047a25751e · celery/celery · GitHub

    def connection(self, hostname=None, userid=None, password=None,
                   virtual_host=None, port=None, ssl=None,
                   connect_timeout=None, transport=None,
                   transport_options=None, heartbeat=None,
                   login_method=None, failover_strategy=None, **kwargs):
        conf = self.conf
        return self.amqp.Connection(
            hostname or conf.BROKER_URL,
            userid or conf.BROKER_USER,
            password or conf.BROKER_PASSWORD,
            virtual_host or conf.BROKER_VHOST,
            port or conf.BROKER_PORT,
            transport=transport or conf.BROKER_TRANSPORT,
            ssl=self.either('BROKER_USE_SSL', ssl),
            heartbeat=heartbeat,
            login_method=login_method or conf.BROKER_LOGIN_METHOD,
            failover_strategy=(
                failover_strategy or conf.BROKER_FAILOVER_STRATEGY
            ),
            transport_options=dict(
                conf.BROKER_TRANSPORT_OPTIONS, **transport_options or {}
            ),
            connect_timeout=self.either(
                'BROKER_CONNECTION_TIMEOUT', connect_timeout
            ),
        )

self.amqp.Connectionは内々でkombu.Connection()を呼び出しているのですが、Pythonインタプリタで引数を何も渡さずにConnectionを呼び出してやると、fallbackが働いたときの挙動がよく分かります。

>>> from kombu import Connection
>>> Connection()
<Connection: amqp://guest:**@localhost:5672// at 0x7f55f3b56810>

補足

今回僕の場合、Redisの前にRabbitMQを試していたため、RabbitMQデーモンが起動した状態になっており、それが災いして原因特定を困難にしました。

project/celery.pyを作らなかったためにfallback(縮退)モードとしてCeleryが動き、アプリケーションから入れたキューはRedisではなくRabbitMQに入ってしまったのでした。

RabbitMQを停止しておけば以下のようにキューを入れる段階でInternal Server Errorが出るのでもう少し調査が捗ったかもしれません。