30

我已经使用他们的守护进程指令（ http://docs.celeryproject.org/en/latest/tutorials/daemonizing.html#daemonizing）设置了 celery 来使用我的 django 应用程序

这是我的测试任务

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

我需要一种方法来知道该任务（调试器）由于异常而失败。Celery 的日志文件打印 logger.info("running debugger") 日志，但它不记录异常。我是否遗漏了什么，或者我应该以其他方式找到失败的任务？

4

5 回答 5

14

问题：

我希望 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 上下文中。这是在 celery 已经在使用的故障代码路径上的捎带。

于 2017-07-26T17:01:43.860 回答
8

您可以查看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()。异常信息被添加到日志消息中。此方法只能从异常处理程序中调用。

于 2014-02-24T15:46:49.997 回答
8

为了从 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装饰器。

于 2018-10-23T18:41:24.467 回答
1

使用 traceback 模块将跟踪捕获为字符串并将其发送到记录器。

try:
    ...
except:
    import traceback
    logger.info(traceback.format_exc())
于 2013-06-25T19:39:51.520 回答
1

您还可以覆盖 celery 应用程序，以避免向base每个装饰器添加 kwarg @app.task：

import logging
from celery import Celery, Task

logger = logging.getLogger(__name__)

class LoggingTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.exception('Task failed: %s' % exc, exc_info=exc)
        super(LoggingTask, self).on_failure(exc, task_id, args, kwargs, einfo)

class LoggingCelery(Celery):
    def task(self, *args, **kwargs):
        kwargs.setdefault('base', LoggingTask)
        return super(LoggingCelery, self).task(*args, **kwargs)

app = LoggingCelery(__name__)
于 2018-01-19T11:04:33.883 回答