多Granian Worker共享日志文件致FastAPI日志丢失问题排查

问题场景 #

使用Granian启动6个Worker的FastAPI服务，部署在Docker容器中，存在请求日志丢失现象：确认有请求产生，但轮转日志中缺失对应记录，初步怀疑问题源于多进程写入同一文件或日志轮转环节。

核心原因分析 #

1. 多进程日志写入竞争：每个Granian Worker都是独立操作系统进程，Python标准库的TimedRotatingFileHandler未提供进程间同步机制，多进程同时写入同一文件时，会出现日志覆盖、截断或丢失的情况。
2. 轮转后的句柄失效：当某一个Worker触发日志轮转后，其他Worker的文件句柄仍指向已被轮转的旧文件，后续日志会写入旧文件而非新的主日志文件，导致这些日志无法被正常收集到当前轮转周期的日志中，表现为"丢失"。

解决方案 #

方案一：使用进程安全的日志处理库 #

concurrent-log-handler专门解决多进程环境下的日志写入冲突问题，通过文件锁机制保证进程间同步，同时支持日志轮转与压缩。

1. 安装依赖：

pip install concurrent-log-handler

2. 修改日志配置中的文件处理器：

import gzip
import logging
import os
import shutil
import tempfile
from pathlib import Path
from typing import Final
from concurrent_log_handler import ConcurrentRotatingFileHandler  # 新增导入

class LoggerConfig:
    _LOG_FMT: Final[str] = "%(asctime)s - %(levelname)s - %(message)s"
    _COPY_BUFSIZE: Final[int] = 1024 * 1024  # 1 MiB

    # 保留原有的_gzip_namer和_gzip_rotator方法不变

    @classmethod
    def setup_logger(
        cls,
        name: str = "my_app",
        level: int = logging.INFO,
        log_file: str = "logs/my_app.log",
        backup_count: int = 365,
        when: str = "midnight",
        interval: int = 1,
    ) -> logging.Logger:
        logger = logging.getLogger(name)
        logger.setLevel(level)
        logger.propagate = False

        if logger.handlers:
            return logger

        formatter = logging.Formatter(cls._LOG_FMT)

        stream_handler = logging.StreamHandler()
        stream_handler.setLevel(level)
        stream_handler.setFormatter(formatter)
        logger.addHandler(stream_handler)

        log_path = Path(log_file)
        log_path.parent.mkdir(parents=True, exist_ok=True)

        # 替换为ConcurrentRotatingFileHandler
        file_handler = ConcurrentRotatingFileHandler(
            filename=str(log_path),
            maxBytes=0,  # 配合when/interval实现定时轮转
            backupCount=backup_count,
            encoding="utf-8",
            errors="backslashreplace",
            delay=True,
            when=when,
            interval=interval,
        )
        file_handler.setLevel(level)
        file_handler.suffix = "%Y-%m-%d"
        file_handler.namer = cls._gzip_namer
        file_handler.rotator = cls._gzip_rotator
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        return logger

方案二：日志输出到stdout/stderr（容器化最佳实践） #

放弃文件日志直接写入，改为将所有日志输出到容器的标准输出/错误流，由Docker统一收集管理，从根源避免多进程写入冲突。

1. 修改日志配置，仅保留StreamHandler：

import logging
from typing import Final

class LoggerConfig:
    _LOG_FMT: Final[str] = "%(asctime)s - %(levelname)s - %(message)s"

    @classmethod
    def setup_logger(
        cls,
        name: str = "my_app",
        level: int = logging.INFO,
    ) -> logging.Logger:
        logger = logging.getLogger(name)
        logger.setLevel(level)
        logger.propagate = False

        if logger.handlers:
            return logger

        formatter = logging.Formatter(cls._LOG_FMT)

        stream_handler = logging.StreamHandler()
        stream_handler.setLevel(level)
        stream_handler.setFormatter(formatter)
        logger.addHandler(stream_handler)

        return logger

2. 配置Docker Compose日志策略：

services:
  my-app:
    build: .
    container_name: my-app
    network_mode: "host"
    restart: unless-stopped
    logging:
      driver: "json-file"
      options:
        max-size: "10m"  # 单个日志文件最大10MB
        max-file: "30"   # 最多保留30个日志文件

方案三：为每个Worker分配独立日志文件 #

通过Granian提供的GRANIAN_WORKER_ID环境变量，让每个Worker写入专属日志文件，避免进程间写入冲突。

修改日志文件名生成逻辑：

@classmethod
def setup_logger(
    cls,
    name: str = "my_app",
    level: int = logging.INFO,
    backup_count: int = 365,
    when: str = "midnight",
    interval: int = 1,
) -> logging.Logger:
    # 获取Worker ID，默认0
    worker_id = os.getenv("GRANIAN_WORKER_ID", "0")
    log_file = f"logs/my_app_worker_{worker_id}.log"
    
    # 后续逻辑与原配置一致，使用生成的log_file路径

方案优先级建议 #

1. 优先选择方案二，符合容器化部署的最佳实践，简化日志管理流程，彻底规避多进程日志冲突问题。
2. 若必须保留本地文件日志，选择方案一，通过专业库实现进程安全的日志写入与轮转。
3. 方案三适合需要按Worker维度排查问题的场景，但会增加日志文件数量，需额外处理日志合并需求。

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