Celery任务日志未显示在Google Cloud Logging的问题排查

我正尝试为Celery任务配置Google Cloud Logging，但日志无法显示在Google Cloud Logging控制台中。已在脚本中添加日志配置，但日志仍未上传。以下是相关配置：

日志配置 #

import json
import os
import logging
from google.cloud import bigquery
from google.cloud import logging as cloud_logging
from google.auth import exceptions as google_auth_exceptions

# Initialize Google Cloud Logging client and setup logging
client = cloud_logging.Client()
client.setup_logging()

# Logger configuration
def setup_logger():
    # Define a global logger variable
    global logger
    logger = logging.getLogger('cloudLogger')
    
    # Check if the logger has handlers already
    if not logger.hasHandlers():
        logger.setLevel(logging.INFO)
        
        # Create a stream handler
        handler = logging.StreamHandler()
        handler.setLevel(logging.INFO)
        
        # Define a formatter without including the script name in the log message
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        
        # Add the handler to the logger
        logger.addHandler(handler)
    
    return logger

# Example usage
if __name__ == "__main__":
    logger = setup_logger()
    logger.info("Logger is set up and ready to use.")

Celery任务代码 #

import uuid
import redis
import logging
from celery import Celery
import boto3
import json
import time
from datetime import datetime
from collections import deque
from google.cloud import bigquery
from google.cloud import logging as cloud_logging
from google.auth import exceptions as google_auth_exceptions
import os 
from logging_setup import setup_logger

    
app = Celery('upload_tasks')
app.config_from_object('celeryconfig')

config = load_config()
logger = setup_logger()

def upload_to_s3(file_path, s3_bucket, s3_key, logger):
    logger.info(f"Initializing S3 client for bucket {s3_bucket}...")
    s3_client = boto3.client('s3', region_name=S3_REGION,
                             aws_access_key_id=S3_ACCESS_KEY,
                             aws_secret_access_key=S3_SECRET_KEY)
    try:
        logger.info(f"Uploading {file_path} to s3://{s3_bucket}/{s3_key}...")
        s3_client.upload_file(file_path, s3_bucket, s3_key)
        logger.info(f"Successfully uploaded {file_path} to s3://{s3_bucket}/{s3_key}")
        return f"s3://{s3_bucket}/{s3_key}"
    except Exception as e:
        logger.error(f"Failed to upload to S3: {e}")
        return None
    
@app.task(bind=True, max_retries=50, queue='upload_files')
def upload_file(self, file_path, s3_bucket, s3_key, data, s3_location):
    # print("")
    logger = setup_logger()
    logger.info(f"Received task to upload {file_path} to S3 bucket {s3_bucket} with key {s3_key}.")
    try:
        s3_location = upload_to_s3(file_path, s3_bucket, s3_key, logger)
        if s3_location:
            store_metadata_in_bigquery(data, s3_location)
            os.remove(file_path)
        logger.info(f"Task completed for uploading {file_path} to s3://{s3_bucket}/{s3_key}")
    except Exception as exc:
        countdown = 5 * (2 ** self.request.retries)  # Exponential backoff
        logger.error(f"Error occurred during upload: {exc}. Retrying in {countdown} seconds...")
        raise self.retry(exc=exc, countdown=countdown)

尽管已配置日志并将其纳入Celery配置，日志仍未出现在Google Cloud Logging控制台中。已验证服务账号拥有必要权限，且GOOGLE_APPLICATION_CREDENTIALS环境变量设置正确。

已尝试的操作 #

• 确认Google Cloud Logging API已启用。
• 检查服务账号拥有「Logging Admin」角色。
• 验证GOOGLE_APPLICATION_CREDENTIALS环境变量设置正确。
• 在代码多个位置添加日志处理器以捕获日志。

环境 #

• Python 3.x
• Celery 4.x
• Google Cloud Logging库

问题排查与解决方案 #

1. 自定义Logger未关联Cloud Logging Handler

当前代码中client.setup_logging()仅为root logger添加Cloud Logging处理器，但你的自定义cloudLogger并未继承该处理器，也未手动添加。需修改setup_logger函数，直接给自定义logger绑定Cloud Logging处理器：

def setup_logger():
    global logger
    logger = logging.getLogger('cloudLogger')
    
    if not logger.hasHandlers():
        logger.setLevel(logging.INFO)
        
        # 添加Google Cloud Logging处理器
        client = cloud_logging.Client()
        cloud_handler = client.get_default_handler()
        logger.addHandler(cloud_handler)
        
        # 保留StreamHandler用于本地调试（可选）
        handler = logging.StreamHandler()
        handler.setLevel(logging.INFO)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    
    return logger

同时移除全局的client = cloud_logging.Client(); client.setup_logging()，避免重复配置。

2. Celery Worker日志劫持问题

Celery 4.x默认会劫持root logger，导致自定义日志处理器失效。需在celeryconfig.py中添加如下配置：

import logging
from google.cloud import logging as cloud_logging

# 给root logger添加Cloud Logging处理器
client = cloud_logging.Client()
cloud_handler = client.get_default_handler()
root_logger = logging.getLogger()
root_logger.addHandler(cloud_handler)

# 禁止Celery劫持root logger
CELERYD_HIJACK_ROOT_LOGGER = False

确保worker启动时加载该配置，让所有日志（包括Celery自身日志）都能流向Google Cloud Logging。

3. Worker进程环境变量验证

确认Celery Worker进程能读取到GOOGLE_APPLICATION_CREDENTIALS环境变量。如果通过systemd、supervisord等工具启动worker，需在启动配置中明确设置该变量，而非仅在当前shell中设置。

4. 日志级别检查

确保logger和所有处理器的日志级别不高于INFO（例如不要设置为WARNING），避免INFO及以下级别的日志被过滤。

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