FastAPI中Correlation/request_id重叠问题排查求助

问题根源 #

你的代码出现request_id跨请求重叠的核心原因是全局Logger的Filter被重复添加且未做上下文隔离：

• FastAPI的异步处理基于单线程事件循环，并非每个请求分配独立线程，所有请求共享同一个全局logger对象。
• 每次处理请求时都调用logger.addFilter(context_info)，但从未移除这些Filter，导致Logger上绑定的Filter数量持续累加。
• 异步环境下，当/wait请求执行await asyncio.sleep(10)时，事件循环会切换到处理其他请求，新请求添加的Filter会修改后续所有日志的request_id——包括旧请求在sleep结束后输出的日志，最终导致request_id跨请求重叠。

正确解决方案：使用ContextVars实现上下文隔离 #

Python的contextvars模块专门用于异步/多线程场景下的上下文变量隔离，能确保每个请求的request_id独立存储，不会互相干扰。

修改后的代码实现 #

main.py

import logging
import uuid
import contextvars
from logging import LogRecord
from fastapi import FastAPI, Depends

import common

# 创建上下文变量存储request_id
request_id_var = contextvars.ContextVar("request_id", default=None)

app = FastAPI()

# Setup Logger
class ContextInfo(logging.Filter):
    def filter(self, record: LogRecord) -> bool:
        # 从当前上下文获取request_id
        record.request_id = request_id_var.get()
        return True

logger = logging.getLogger('test-app')
logger.setLevel(logging.DEBUG)

formatter = logging.Formatter('%(name)-12s: %(asctime)s - %(levelname)-8s - request_id: %(request_id)-15s - %(funcName)s - %(message)s')

stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)

logger.addHandler(stream_handler)
# 只添加一次全局Filter，不再在请求中重复添加
logger.addFilter(ContextInfo())

logger.propagate = False

# 定义依赖：生成并设置request_id到上下文
def get_request_id():
    req_id = str(uuid.uuid4())
    request_id_var.set(req_id)
    return req_id

@app.get("/wait")
async def wait(request_id: str = Depends(get_request_id)):
    logger.info("Request comes from client to get long-waited resource access")
    await common.wait_longer()
    logger.info("Request /wait completed successfully")
    return {"message": "I am come after long wait", "request_id": request_id}

@app.get("/quick")
async def quick(request_id: str = Depends(get_request_id)):
    logger.info("Request comes from client to get quick resource access")
    common.quick()
    logger.info("Request /quick completed successfully")
    return {"message": "Quick response", "request_id": request_id}

common.py（无需修改）

import logging
import asyncio

logger = logging.getLogger('test-app')

async def wait_longer():
    logger.info('Starting long process')
    await asyncio.sleep(10)
    return None

def quick():
    logger.info('Starting quick process')
    return None

方案说明 #

1. 上下文变量隔离：request_id_var作为上下文变量，每个请求的request_id会被存储在独立的上下文中，异步任务切换时不会互相覆盖。
2. 全局Filter复用：只给Logger添加一次ContextInfo Filter，Filter每次从当前上下文获取request_id，确保日志中的request_id属于当前请求。
3. 依赖注入管理：通过FastAPI的依赖get_request_id统一生成和设置request_id，避免重复代码，同时确保请求开始时就初始化上下文变量。

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