Celery：将每个任务运行记录到独立文件？

Question

我想让每个正在运行的任务都记录到它自己专属的文件里，这些文件都放在logs/这个文件夹里，文件名就是任务的ID。

logger = get_task_logger(__name__)

@app.task(base=CallbackTask)
def calc(syntax):
    some_func()
    logger.info('started')

在我的工作程序中，我通过使用 -f 这个参数来设置日志文件的输出位置。我想确保每个任务都能输出到它自己的日志文件里。

Answer 1

这是我的代码：

import logging

from celery.app.log import TaskFormatter
from celery.signals import task_prerun, task_postrun

@task_prerun.connect
def overload_task_logger(task_id, task, args, **kwargs):
    logger = logging.getLogger("celery.task")
    file_handler = logging.FileHandler(f'/tmp/{task_id}.log')
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(
        TaskFormatter("[%(asctime)s: %(levelname)s/%(processName)s] %(task_name)s[%(task_id)s]: %(message)s")
    )
    logger.addHandler(file_handler)


@task_postrun.connect
def cleanup_logger(task_id, task, args, **kwargs):
    logger = logging.getLogger("celery.task")
    for handler in logger.handlers:
        if isinstance(handler, logging.FileHandler) and handler.baseFilename == f'/tmp/{task_id}.log':
            logger.removeHandler(handler)

我在 celery 版本 5.2.7 上测试过这个代码。

Answer 2

看起来我来晚了三年。不过，我还是想分享一下我的解决方案，这个灵感来自@Mikko Ohtamaa的想法。我稍微做了一些不同的调整，使用了Celery的信号和Python自带的日志框架来准备和清理日志处理。

from celery.signals import task_prerun, task_postrun
import logging

# to control the tasks that required logging mechanism
TASK_WITH_LOGGING = ['Proj.tasks.calc']
@task_prerun.connect(sender=TASK_WITH_LOGGING)
def prepare_logging(signal=None, sender=None, task_id=None, task=None, args=None, kwargs=None):
    logger = logging.getLogger(task_id)
    formatter = logging.Formatter('[%(asctime)s][%(levelname)s] %(message)s')
    # optionally logging on the Console as well as file
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(formatter)
    stream_handler.setLevel(logging.INFO)
    # Adding File Handle with file path. Filename is task_id
    task_handler = logging.FileHandler(os.path.join('/tmp/', task_id+'.log'))
    task_handler.setFormatter(formatter)
    task_handler.setLevel(logging.INFO)
    logger.addHandler(stream_handler)
    logger.addHandler(task_handler)

@task_postrun.connect(sender=TASK_WITH_LOGGING)
def close_logging(signal=None, sender=None, task_id=None, task=None, args=None, kwargs=None, retval=None, state=None):
    # getting the same logger and closing all handles associated with it
    logger = logging.getLogger(task_id)
    for handler in logger.handlers:
        handler.flush()
        handler.close()
    logger.handlers = []

@app.task(base=CallbackTask, bind=True)
def calc(self, syntax):
    # getting logger with name Task ID. This is already
    # created and setup in prepare_logging
    logger = logging.getLogger(self.request.id)
    some_func()
    logger.info('started')

这里的bind=True是必须的，这样才能在任务中使用到任务的ID。每次执行calc这个任务时，它都会创建一个独立的日志文件，文件名格式是<task_id>.log。

Answer 3

下面是我随便写的、没有经过测试的方法。可以把它当作一种指导，而不是可以直接用的生产级代码。

def get_or_create_task_logger(func):
    """ A helper function to create function specific logger lazily. """

    # https://docs.python.org/2/library/logging.html?highlight=logging#logging.getLogger
    # This will always result the same singleton logger
    # based on the task's function name (does not check cross-module name clash, 
    # for demo purposes only)
    logger = logging.getLogger(func.__name__)

    # Add our custom logging handler for this logger only
    # You could also peek into Celery task context variables here
    #  http://celery.readthedocs.org/en/latest/userguide/tasks.html#context
    if len(logger.handlers) == 0:
        # Log to output file based on the function name
        hdlr = logging.FileHandler('%s.log' % func.__name__)
        formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
        hdlr.setFormatter(formatter)
        logger.addHandler(hdlr) 
        logger.setLevel(logging.DEBUG)

    return logger

@app.task(base=CallbackTask)
def calc(syntax):
    logger = get_or_create_task_logger(calc)
    some_func()
    logger.info('started')