如何将Celery INFO级调度任务等框架日志调整为DEBUG级别

前置问题修正 #

你的现有日志配置存在一个小疏漏：propagate字段填写的是字符串'False'，Python logging模块会将非空字符串判定为真值，可能导致日志向上重复传播，先将其修改为布尔值False。

核心实现方案 #

核心思路是通过自定义日志过滤器，匹配Celery框架输出的指定INFO级日志，将其动态降级为DEBUG级别，不需要修改Celery源码，侵入性最低。

步骤1：编写自定义日志过滤器 #

import logging

class CeleryInfoToDebugFilter(logging.Filter):
    def filter(self, record):
        # 可根据实际需要调整要降级的日志关键词
        target_keywords = (
            "Received task:",
            "Scheduler: Sending due task",
            "succeeded in",
            "failed in",
            "retried",
        )
        # 仅处理Celery命名空间下的INFO级日志
        if record.name.startswith("celery.") and record.levelno == logging.INFO:
            log_msg = record.getMessage()
            if any(keyword in log_msg for keyword in target_keywords):
                # 动态修改日志级别为DEBUG
                record.levelno = logging.DEBUG
                record.levelname = "DEBUG"
        return True

步骤2：更新你的GCP日志配置 #

将过滤器注册到日志配置中，并绑定到你的GCP日志处理器：

LOGGING_GCP = {
    'version': 1,
    'disable_existing_loggers': True,
    # 新增过滤器配置
    'filters': {
        'celery_info_to_debug': {
            '()': CeleryInfoToDebugFilter,
        }
    },
    'formatters': {
        'standard': {
            'class': 'logging.Formatter',
            'format': '%(message)s | %(module)s:%(lineno)d',
        },
    },
    'handlers': {
        'gcp': {
            'level': 'DEBUG',
            'class': 'google.cloud.logging.handlers.ContainerEngineHandler',
            'formatter': 'standard',
            'filters': ['celery_info_to_debug'], # 绑定自定义过滤器
        },
    },
    'loggers': {
        '': {
            'handlers': ['gcp'],
            'level': 'DEBUG',
            'propagate': False, # 修正为布尔值
        }
    }
}

补充配置说明 #

如果使用的是Celery 4.x及以上版本，建议将旧版大写配置替换为官方推荐的小写配置，避免兼容性问题：

worker_hijack_root_logger = False
worker_redirect_stdouts = False

配置完成后重启Celery worker和beat，指定的Celery框架INFO日志就会自动转为DEBUG级别，不会和业务INFO日志混杂。

内容的提问来源于stack exchange，提问作者dzh