如何记录django芹菜任务中发生的异常

Question

我已经设置芹菜使用他们的daemonization指令来处理我的django应用程序(http://docs.celeryproject.org/en/latest/tutorials/daemonizing.html#daemonizing)

这是我的测试任务

@periodic_task(run_every=timedelta(seconds=10))
def debugger():
    logger.info("Running debugger")
    raise Exception('Failed')

我需要一种方法来了解此任务(调试器)由于异常而失败.Celery的日志文件打印logger.info("运行调试器")日志,但它不记录异常.我错过了什么,或者我应该以其他方式找到失败的任务？

Answer 1

问题:

我希望Celery捕获异常并将它们写入日志文件而不是显然吞下它们......

对于专业解决方案而言,目前的最佳答案是一般的.许多python开发人员会根据具体情况考虑一揽子错误.在评论中明确表达了对此的合理厌恶:

坚持下去,我希望工作日志中记录的内容,至少是每个失败的任务......

Celery确实捕获了异常,它只是没有做OP想要它做的事情(它将它存储在结果后端).以下要点是互联网在这个问题上提供的最好的.这有点过时了,但请注意叉子和星星的数量.

https://gist.github.com/darklow/c70a8d1147f05be877c3

要点是采取失败的情况并做一些自定义的事情.这是OP问题的超集.以下是如何调整gist中的解决方案以记录异常.

import logging

logger = logging.getLogger('your.desired.logger')


class LogErrorsTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.exception('Celery task failure!!!1', exc_info=exc)
        super(LogErrorsTask, self).on_failure(exc, task_id, args, kwargs, einfo)

您仍然需要确保所有任务都继承自此任务类,并且如果您正在使用@task装饰器(使用base=LogErrorsTaskkwarg),则要点显示如何执行此操作.

此解决方案的好处是不将代码嵌套在任何其他try-except上下文中.这是在芹菜已经使用的故障代码路径上捎带的.

Answer 2

您可以查看Celery用户指南:

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task
def div():
    try:
        1 / 0
    except ZeroDivisionError:
        logger.exception("Task error")

从python日志记录模块的文档:

Logger.exception(msg,*args)

在此记录器上记录级别为ERROR的消息.参数被解释为debug().异常信息将添加到日志消息中.只应从异常处理程序调用此方法.

Answer 3

为了接收来自 Celery 任务的所有未处理的异常，我注册了一个信号处理程序。我正在格式化一条logging.error消息，然后可以由默认的 Python 日志记录配置处理。

这是相关的部分

from celery import signals

@signals.task_retry.connect
@signals.task_failure.connect
@signals.task_revoked.connect
def on_task_failure(**kwargs):
    """Abort transaction on task errors.
    """
    # celery exceptions will not be published to `sys.excepthook`. therefore we have to create another handler here.
    from traceback import format_tb

    log.error('[task:%s:%s]' % (kwargs.get('task_id'), kwargs['sender'].request.correlation_id, )
              + '\n'
              + ''.join(format_tb(kwargs.get('traceback', [])))
              + '\n'
              + str(kwargs.get('exception', '')))

请注意，此信号处理程序自动适用于所有任务；即它不需要改变你的task装饰器。