Celery任务内GCP Logging无法正常输出日志问题

Celery默认采用fork模式启动worker子进程，主进程中初始化的CloudLoggingHandler无法在子进程中正常工作：

1. Fork操作会复制主进程内存，但GCP Logging依赖的网络连接类资源无法跨进程复用；
2. Celery会在子进程启动时重置日志配置，将原有handler替换为StreamHandler，导致任务内日志仅输出到控制台；
3. 主进程的OUTSIDE Task日志重复推送，是因为fork时主进程模块级代码的日志操作被子进程重复执行。

方案1：利用Celery Worker初始化信号配置日志 #

通过worker_init信号，在每个子进程启动时重新初始化GCP Logging，确保每个worker进程都拥有有效的日志handler：

from celery import Celery
from celery.signals import worker_init
import google.cloud.logging
import logging

app = Celery('celery_tasks', backend='redis://localhost//0', broker='redis://localhost//1')

def setup_gcp_logging(**kwargs):
    # 清空root logger原有handler，避免重复输出
    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    # 重新初始化GCP Logging
    client = google.cloud.logging.Client()
    client.setup_logging()

# 绑定子进程初始化信号
worker_init.connect(setup_gcp_logging)

@app.task
def dummyFunc(firstArg, secondArg):
    logging.info('INSIDE Task!!!')
    result = {
        'sum': firstArg + secondArg,
        'id': 15 
    }
    return result

# 仅主进程启动时执行一次
logging.info('OUTSIDE Task')

方案2：任务内按需初始化日志 #

在任务内部检查是否已配置GCP handler，未配置则初始化（适合任务执行频率较低的场景）：

from celery import Celery
import google.cloud.logging
import logging

app = Celery('celery_tasks', backend='redis://localhost//0', broker='redis://localhost//1')

@app.task
def dummyFunc(firstArg, secondArg):
    root_logger = logging.getLogger()
    # 检查是否已有CloudLoggingHandler
    has_gcp_handler = any(
        isinstance(h, google.cloud.logging.handlers.CloudLoggingHandler) 
        for h in root_logger.handlers
    )
    if not has_gcp_handler:
        client = google.cloud.logging.Client()
        client.setup_logging()
    
    logging.info('INSIDE Task!!!')
    result = {
        'sum': firstArg + secondArg,
        'id': 15 
    }
    return result

logging.info('OUTSIDE Task')

方案3：修改Celery全局日志配置 #

将GCP Logging handler直接加入Celery的日志系统配置：

from celery import Celery
import google.cloud.logging
import logging

app = Celery('celery_tasks', backend='redis://localhost//0', broker='redis://localhost//1')

# 初始化GCP Logging handler
client = google.cloud.logging.Client()
gcp_handler = client.get_default_handler()

# 配置Celery使用自定义日志handler
app.conf.update(
    CELERYD_LOG_HANDLERS = [gcp_handler],
    CELERYD_LOG_LEVEL = logging.INFO
)

@app.task
def dummyFunc(firstArg, secondArg):
    logging.info('INSIDE Task!!!')
    result = {
        'sum': firstArg + secondArg,
        'id': 15 
    }
    return result

logging.info('OUTSIDE Task')

保持原有worker启动命令不变：

celery -A celery_tasks worker --concurrency=1 --loglevel=info

执行任务后，GCP Logs Explorer中即可看到INSIDE Task!!!的日志，且不会重复推送OUTSIDE Task。

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