Celeryの実行ログを取る


## 発端

Django ＋ Redis（時に ElastiCashe）＋ Celery の組み合わせは、よく使います。

* 非同期処理は Python スレッドは禁止（結果として重い物を流すと gunicorn/uWSGI の反応がなくなるので）。Celery workerを使う。 
* cron ジョブ的なバッチ処理も、Celery beat で書く。

という感じでやってます。

簡単に書けるのですが、流れたかどうか知りたい場合、Celery worker が１台だけならログファイルを見ればいいとして、複数台になってくると、どこで流れたんだ、という所から調べるのが大変。

## 何を依頼して、何が失敗したか知りたい

ということで、依頼 -> 成功時／失敗時の実行ログを RDB のモデルに出して、adminサイトで確認できるようにします。

もちろん、Celery worker が1台だけの場合でも、ログファイルを追うより楽になります。

既に Celery Task をいっぱい書いてしまっているので、なるべく簡単に実現したい。

## 継承元のTask

通常は、`from celery import Task` で書き始めますが、エラーハンドリング対応の Task を書くことにします。

呼び出し側は、delay() の次に呼ばれる apply_async() で依頼時のログを出しています。
呼ばれた側は、on_success() で正常時、on_failure() で異常時のログを出しています。

使い方は、SampleTask のように継承元を Task -> BaseHandlingTask に変えればOK。

ちなみに、delay() の非同期で作っておいて、やっぱり同期でということで、run() に直した場合は何も記録されません。

app/tasks/base/handling.py

```python
import json
import logging
import socket
import sys
import traceback
from celery import Task
from django.db import transaction

from app.models import JobState


class BaseHandlingTask(Task):
    """エラー ハンドリングを行うベースタスク"""
    logger = logging.getLogger('prj')

    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None, link=None, link_error=None, **options):
        """Apply tasks asynchronously by sending a message."""
        async_result = None
        try:
            async_result = super().apply_async(args, kwargs, task_id, producer, link, link_error, **options)
        except:     # Redis が上がっていない等
            # 呼び出し側が atomic ブロック内だと、一緒に rollback されてしまうので、２つめの接続を使う
            with transaction.atomic(using='force'):
                exc_type, exc_value, exc_traceback = sys.exc_info()
                job_state = JobState()  # task_id はない
                job_state.task_name = self.name
                job_state.name = self.name.split('.')[-1]
                if args:
                    job_state.args = json.dumps(list(args))
                if kwargs:
                    job_state.kwargs = json.dumps(kwargs)
                job_state.origin = socket.gethostname()
                job_state.exception_class = exc_value.__class__.__name__
                job_state.exception_msg = exc_value.args[0] if exc_value.args else exc_type.__module__ + '.' + exc_type.__name__
                job_state.traceback = traceback.format_exc()
                job_state.save(using='force')
            raise

        # 起動 成功時 - Redis は上がっている。Celery は上がっている／上がっていない(is_complete == False のまま残る)
        job_state, is_created = JobState.objects.get_or_create(task_id=async_result.id)
        job_state.task_name = async_result.task_name
        job_state.name = async_result.task_name.split('.')[-1]
        if args:
            job_state.args = json.dumps(list(args))
        if kwargs:
            job_state.kwargs = json.dumps(kwargs)
        job_state.origin = socket.gethostname()
        job_state.save()

        return async_result

    def on_success(self, retval, task_id, args, kwargs):
        """Success handler - 正常時のハンドラー - Celery worker 側で呼ばれる"""
        job_state, is_created = JobState.objects.get_or_create(task_id=task_id)
        job_state.task_name = self.name
        job_state.name = self.name.split('.')[-1]
        if args:
            job_state.args = json.dumps(list(args))
        if kwargs:
            job_state.kwargs = json.dumps(kwargs)
        if self.request.get('origin'):
            job_state.origin = self.request.get('origin')
        job_state.hostname = self.request.get('hostname')
        job_state.is_complete = True
        job_state.save()

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """Error handler - エラー時のハンドラー - Celery worker 側で呼ばれる
            - run() の atomic ブロック内であっても、別途 worker から呼ばれるので、記録は rollback されない
        """
        job_state, is_created = JobState.objects.get_or_create(task_id=task_id)
        job_state.task_name = self.name
        job_state.name = self.name.split('.')[-1]
        if args:
            job_state.args = json.dumps(list(args))
        if kwargs:
            job_state.kwargs = json.dumps(kwargs)
        if self.request.get('origin'):
            job_state.origin = self.request.get('origin')
        job_state.hostname = self.request.get('hostname')
        job_state.exception_class = exc.__class__.__name__
        job_state.exception_msg = str(exc)
        job_state.traceback = str(einfo)
        job_state.save()

    def run(self, *args, **kwargs):
        """The body of the task executed by workers."""
        raise NotImplementedError('Tasks must define the run method.')


class SampleTask(BaseHandlingTask):
    """サンプル タスク"""
    logger = logging.getLogger('prj')

    def run(self, is_error=False):
        self.logger.info('SampleTask start...')
        with transaction.atomic():
            if is_error:
                raise ValueError('エラーです')
        self.logger.info('SampleTask end.')

```

## JobState モデル

ログの出力先のモデルです。

* タスク名
* 依頼時の引数
* どのホストから起動されたか
* どのホストで実行されたか
* 正常に完了したか
* Exception 時は、例外のクラス／メッセージ／traceback

がわかるようになっています。

app/models/job_state.py

```python
from django.db import models


class JobState(models.Model):
    """ジョブ状態"""
    task_id = models.CharField('タスクID', max_length=255, blank=True, null=True, db_index=True)   # UUID
    task_name = models.CharField('タスク名', max_length=255, blank=True, null=True)     # 例: app.tasks.handling.SampleTask
    name = models.CharField('クラス名', max_length=255, blank=True, null=True)          # 例: SampleTask
    args = models.TextField('args', null=True, blank=True)
    kwargs = models.TextField('kwargs', null=True, blank=True)
    is_complete = models.BooleanField('完了', default=False)
    origin = models.CharField('origin', max_length=255, blank=True, null=True)  # Name of host that sent this task.
    hostname = models.CharField('hostname', max_length=255, blank=True, null=True)  # Node name of the worker instance executing the task.
    exception_class = models.CharField('例外クラス', max_length=255, null=True, blank=True, default='')
    exception_msg = models.CharField('例外メッセージ', max_length=255, null=True, blank=True, default='')
    traceback = models.TextField('traceback', null=True, blank=True, default='')

    created_at = models.DateTimeField('登録日時', auto_now_add=True, blank=True, null=True)
    updated_at = models.DateTimeField('更新日時', auto_now=True, blank=True, null=True)

    def __str__(self):
        return self.task_id if self.task_id else str(self.id)
```

## adminサイト登録

adminサイトで JobState を見れるようにします。
そのままだと長くなるので、所々短縮してたりします。

app/admin.py

```python
from django.contrib import admin
from app.models import JobState

class JobStateAdmin(admin.ModelAdmin):
    """ジョブ状態"""
    list_display = ('id', 'task_id_shorten', 'name', 'x_args', 'x_kwargs', 'is_complete', 'exception_class', 'origin', 'hostname', 'created_at')
    list_display_links = ('id', 'task_id_shorten', 'name')
    list_filter = ('is_complete',)
    search_fields = ['task_id', 'task_name', 'name']

    def task_id_shorten(self, obj):
        return obj.task_id[:8] + '...' if obj.task_id else ''
    task_id_shorten.short_description = 'タスクID'

    def x_args(self, obj):
        return obj.args[:20] + '...' if obj.args and len(obj.args) > 20 else obj.args
    x_args.short_description = 'args'

    def x_kwargs(self, obj):
        return obj.kwargs[:20] + '...' if obj.kwargs and len(obj.kwargs) > 20 else obj.kwargs
    x_kwargs.short_description = 'kwargs'

admin.site.register(JobState, JobStateAdmin)
```

## settings

呼び出し側が transaction.atomic() だと、異常時に一緒に JobState も rollback されてしまうので、ちょっと強引ですが、もう一つ強制書き込み用の Database 接続を増やしてます。

そこまでしなくていいよ、という場合はお好みで。

DATABASE = {} の下に、以下を追加

prj/settings/local.py

```python
# atmic ブロック内で Exception しても、JobState には書き込みできるよう、同じ設定で２つ目の接続を作る
DATABASES.update({'force': DATABASES['default']})
```

## 以上

荒削りのコードですが、一目瞭然で見れるようになりました。

正常時も含めると多すぎるという場合は、正常時に出している所をコメントアウトする。

また、cron ジョブ（Celery beat）で15分単位で流れるものは、これも多すぎるので、そもそも BaseHandlingTask を継承しないで、素の Task を継承する。

等で調整して下さい。