Cloud Run Jobs容器提前退出致日志丢失问题求助

问题背景 #

查看Cloud Run Job特定任务日志时发现：多个任务显示Container called exit(0)（任务执行成功），但大量前置日志缺失，Log Explorer中也无法查看。推测容器未完成日志刷新就已退出。

用户尝试在脚本中添加日志刷新和延迟退出逻辑，但仅短运行任务能看到logger.info("Waiting for 5 seconds before exiting to allow logs to be flushed.")的日志，长任务仍存在日志丢失问题。相关代码片段如下：

任务执行脚本片段 #

try:
    logger.info("TESTS")
    task_params = get_task_params()
    run_task(TASK_INDEX, TASK_ATTEMPT, task_params)
    logger.info("Waiting for 5 seconds before exiting to allow logs to be flushed.")
    logging_config.flush_logs()
    time.sleep(5)  # Wait 5 seconds before exiting to allow logs to be flushed
except Exception as err:
    message = f"Task #{TASK_INDEX}, Attempt #{TASK_ATTEMPT} failed: {str(err)}"
    logger.info(message)
    sys.exit(1)  # Non-zero exit code for failure

日志配置文件config.py片段 #

def setup_logging(config):
    # Instantiates a client
    client = google.cloud.logging.Client()
    # Retrieves a Cloud Logging handler based on the environment
    # you're running in and integrates the handler with the
    # Python logging module. By default this captures all logs
    # at INFO level and higher
    client.setup_logging()

    if not os.path.exists(os.path.dirname(config.LOG_FILE)):
        os.makedirs(os.path.dirname(config.LOG_FILE))

    log_formatter = logging.Formatter(
        "%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]"
    )

    # Configure the file handler
    file_handler = RotatingFileHandler(
        config.LOG_FILE, maxBytes=1024 * 1024 * 10, backupCount=10
    )
    file_handler.setFormatter(log_formatter)
    file_handler.setLevel(getattr(logging, config.LOG_LEVEL.upper()))

    # Configure the stream handler
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(log_formatter)
    stream_handler.setLevel(getattr(logging, config.LOG_LEVEL.upper()))

    # Set the log level and handlers for the logger
    logger = logging.getLogger()  # Define logger at the module level
    logger.setLevel(getattr(logging, config.LOG_LEVEL.upper()))
    logger.addHandler(file_handler)
    logger.addHandler(stream_handler)

    # Log the initialization
    logger.info("Logging is set up")


def flush_logs(config=configurations["default"]):
    logger = logging.getLogger()
    handlers = logger.handlers[:]
    for handler in handlers:
        handler.flush()
        handler.close()
        logger.removeHandler(handler)

    logger.info("All logging handlers have been flushed and closed.")

    logging.shutdown()


# Initialize logging with the desired configuration
setup_logging(configurations["default"])

修复方案 #

1. 强制Cloud Logging Handler使用同步模式 #

client.setup_logging()默认添加异步Handler，长任务中异步日志可能未完成上传容器就已退出。修改日志初始化逻辑，改用同步Handler：

def setup_logging(config):
    client = google.cloud.logging.Client()
    # 替换默认异步Handler为同步Handler
    handler = google.cloud.logging.handlers.CloudLoggingHandler(client)
    logging.getLogger().addHandler(handler)
    
    # 保留原有文件和流Handler配置（省略重复代码）

2. 优化日志刷新逻辑，避免提前移除Handler #

当前flush_logs方法在刷新后直接关闭所有Handler，导致最后一条日志无法被处理。调整逻辑：

def flush_logs(config=configurations["default"]):
    logger = logging.getLogger()
    # 先记录刷新开始日志，再处理Handler
    logger.info("Starting to flush all logging handlers.")
    
    handlers = logger.handlers[:]
    for handler in handlers:
        handler.flush()
        # 仅关闭文件Handler，流和Cloud Logging Handler无需主动关闭
        if isinstance(handler, RotatingFileHandler):
            handler.close()
            logger.removeHandler(handler)
    
    logging.shutdown()
    # 注意：此时Logger已无有效Handler，不要再添加日志

3. 等待Cloud Logging队列排空，替代固定延迟 #

固定5秒延迟对长任务不可靠，直接等待Cloud Logging后台队列完成上传：

from google.cloud.logging.handlers.transports import BackgroundThreadTransport

# 在run_task执行完成后添加：
logger.info("Waiting for Cloud Logging to finish uploading logs.")
# 遍历Handler，找到Cloud Logging Handler并等待队列排空
for handler in logger.handlers:
    if isinstance(handler, google.cloud.logging.handlers.CloudLoggingHandler):
        if isinstance(handler.transport, BackgroundThreadTransport):
            handler.transport.worker.join()

# 再执行本地日志刷新
logging_config.flush_logs()

4. 禁用Stream Handler的缓冲 #

默认Stream Handler存在缓冲，修改为无缓冲模式：

import sys
import os

# 在创建Stream Handler时添加：
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(log_formatter)
stream_handler.setLevel(getattr(logging, config.LOG_LEVEL.upper()))
# 禁用缓冲
stream_handler.stream = os.fdopen(sys.stdout.fileno(), 'w', 0)

5. 延长Cloud Run Job的终止等待时间 #

在Job配置中增加terminationGracePeriodSeconds参数，给容器足够时间完成日志上传（最大可设为3600秒）：

# Job配置示例片段
apiVersion: run.googleapis.com/v1
kind: Job
metadata:
  name: your-job-name
spec:
  template:
    spec:
      containers:
      - name: your-container
        image: your-image
      terminationGracePeriodSeconds: 60  # 根据任务时长调整

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