TrackingID的设计

前言

在实际业务中，根据 tracking_id 追查日志中一条请求的完整处理路径是一个比较常见的需求。不过 FastAPI 官方并没有提供相对应的功能，因此需要开发者自行实现。本文介绍如何基于 contextvars，为每次请求的完整流程都添加一个 tracking_id，并在日志中自动记录。

什么是 contextvars

Python 在 3.7 版本的标准库中加入了一个模块 contextvars，顾名思义就是 "(Context Variables) 上下文变量"，通常用来隐式地传递一些环境信息的变量，其作用跟 threading.local() 比较相似。不过 threading.local() 是针对线程的，隔离线程之间的数据状态，而 contextvars 可以用在 asyncio 生态的异步协程中。

提示

contextvars 不仅可以用在异步协程中，也可以替代 threading.local() 用在多线程函数中。

基本使用

1. 首先编写 context.py

import contextvars
from typing import Optional

TRACKING_ID: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
    'tracking_id', 
    default=None
)

def get_tracking_id() -> Optional[str]:
    """用于依赖注入"""
    return TRACKING_ID.get()

2. 编写中间件 middlewares.py，在请求头和响应头中添加 tracking_id 的信息。常见场景就是客户拿着 tracking_id 找碴。

import uuid

from starlette.middleware.base import (BaseHTTPMiddleware,
                                       RequestResponseEndpoint)
from starlette.requests import Request
from starlette.responses import Response

from context import TRACKING_ID


class TrackingIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        tracking_id = str(uuid.uuid4())
        token = TRACKING_ID.set(tracking_id)
        # HTTP 请求头习惯于使用 latin-1 编码
        request.scope["headers"].append((b"x-request-id", tracking_id.encode("latin-1")))

        try:
            resp = await call_next(request)
        finally:
            # 无论是否成功，每次请求结束时重置 tracking_id，避免泄露到下一次的请求中
            TRACKING_ID.reset(token)

        # 可选, 在响应中设置跟踪 ID 头
        resp.headers["X-Tracking-ID"] = tracking_id

        return resp

3. 编写 handler 函数 handlers.py，测试在 handler 函数中获取 tracking_id。

import asyncio

from context import TRACKING_ID


async def mock_db_query():
    await asyncio.sleep(1)
    current_id = TRACKING_ID.get()
    print(f"This is mock_db_query. Current tracking ID: {current_id}")
    await asyncio.sleep(1)

4. 编写主函数 main.py

import uvicorn
from fastapi import Depends, FastAPI
from fastapi.responses import PlainTextResponse
from starlette.background import BackgroundTasks

from context import TRACKING_ID, get_tracking_id
from handlers import mock_db_query
from middlewares import TrackingIDMiddleware

app = FastAPI()

app.add_middleware(TrackingIDMiddleware)


@app.get("/qwer")
async def get_qwer():
    """测试上下文变量传递"""
    current_id = TRACKING_ID.get()
    print(f"This is get qwer. Current tracking ID: {current_id}")
    return PlainTextResponse(f"Current tracking ID: {current_id}")


@app.get("/asdf")
async def get_asdf(tracking_id: str = Depends(get_tracking_id)):
    """测试依赖注入"""
    print(f"This is get asdf. tracking ID: {tracking_id}")
    await mock_db_query()
    return PlainTextResponse(f"Get request, tracking ID: {tracking_id}")

if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8000, workers=4)

5. 启动服务后用 curl 测试 api，在控制台可以看到 tracking_id 在请求中都能捕获到。

This is get qwer. Current tracking ID: 01b0153f-4877-4ca0-ac35-ed88ab406452
INFO:     127.0.0.1:55708 - "GET /qwer HTTP/1.1" 200 OK
This is get asdf. tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639
This is mock_db_query. Current tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639
INFO:     127.0.0.1:55722 - "GET /asdf HTTP/1.1" 200 OK

使用 curl 的控制台输出

$ curl -i http://127.0.0.1:8000/qwer
HTTP/1.1 200 OK
date: Sat, 29 Nov 2025 06:16:46 GMT
server: uvicorn
content-length: 57
content-type: text/plain; charset=utf-8
x-tracking-id: 01b0153f-4877-4ca0-ac35-ed88ab406452

Current tracking ID: 01b0153f-4877-4ca0-ac35-ed88ab406452

===== 另一个请求 =====

$ curl -i http://127.0.0.1:8000/asdf
HTTP/1.1 200 OK
date: Sat, 29 Nov 2025 06:16:49 GMT
server: uvicorn
content-length: 62
content-type: text/plain; charset=utf-8
x-tracking-id: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639

Get request, tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639

后台任务型 API

FastAPI 中有 from starlette.background import BackgroundTasks 可以让接口直接响应，将实际流程放到后台异步处理。因为上面的中间件在响应时会重置 tracking_id，所以后台的协程函数可能不会获取到 tracking_id。理论上是这样的，但是本地测试时发现在异步协程中还是能获取到 tracking_id，这可能是本地低并发的问题。在生产环境高并发的情况下，最好还是强制解耦，显式传递 contextvars。

1. 自定义的中间件类保持不变。
2. 添加后台任务的 api

from starlette.background import BackgroundTasks
from handlers import mock_backgroud_task

@app.get("/zxcv")
async def get_zxcv(tasks: BackgroundTasks):
    """测试后台任务"""
    current_id = TRACKING_ID.get()
    print(f"This is get zxcv. The current id is {current_id}")

    # 显式传递 tracking_id
    tasks.add_task(mock_backgroud_task, current_id)

    return PlainTextResponse(f"This is get zxcv. The current id is {current_id}")

3. 后台协程函数中显式处理。任务在启动时，使用传入的参数值，在自己的任务执行上下文中，重新设置 TRACKING_ID。在任务结束时，对任务自己设置的上下文进行清理。

async def mock_backgroud_task(request_tracking_id: Optional[str]):
    if request_tracking_id is None:
        request_tracking_id = str(uuid4())
        print(f"WARNING: No tracking ID found. Generate a new one: {request_tracking_id}")
    token = TRACKING_ID.set(request_tracking_id)
    try:
        # 模拟耗时的后台异步任务
        await asyncio.sleep(5)
        print(f"This is mock backgroud task. Current tracking ID: {request_tracking_id}")
    finally:
        # 确保 tracking ID 被重置
        TRACKING_ID.reset(token)

4. 启动主程序并测试 tracking_id 是否一致。

日志记录 tracking_id

tracking_id 在前面是手动添加到 print() 里面的，未免还是麻烦了点，不注意的话还可能忘了。因此，最好是用日志记录器自动获取 tracking_id，这样代码的侵入性改动就大大降低了。

首先需要封装一个日志模块。这里选用标准库 logging，喜欢 loguru 的话也可以试试用 loguru 封装（如果生产环境有频繁且严苛的安全扫描，个人建议还是尽量用标准库比较好，免得经常升级第三方依赖）。如果你的应用运行在 docker 或者 k8s 上，有现成的控制台日志采集工具，那么这个日志模块只需要输出到控制台就行了。如果应用运行在 server 上，需要用比如 filebeat 这样的工具读取日志文件，那么需要解决日志文件体积增长、uvicorn 多工作进程运行同时写日志文件的竞争问题、以及高并发情况下写日志文件阻塞异步协程的问题。

以下日志模块的示例解决了上述问题，主要特点：

• 自动获取 tracking_id，无需手动记录。
• 日志为 JSON 格式，便于在 ELK 这样的日志聚合系统中查日志。
• 日志会同时输出到标准输出和日志文件，也可以配置输出到其中一个。
• 日志文件按天轮转，默认保留最近 7 天的日志，避免日志文件体积增长的问题。
• 主进程和工作进程输出到各自的日志文件中，避免同时写日志文件的竞争问题。
• 日志会先输出到队列，避免写文件阻塞异步协程的问题。

1. 编辑文件 pkg/log/log.py。主要内容都在这个部分。需要对外提供的对象只有 setup_logger 和 close_log_queue。

import json
import logging
import os
import re
import sys
from logging.handlers import QueueHandler, QueueListener, TimedRotatingFileHandler
from multiprocessing import current_process
from pathlib import Path
from queue import Queue
from typing import Optional

from context import TRACKING_ID

_queue_listener = None
_queue_logger: Optional[Queue] = None
PATTERN_PROCESS_NAME = re.compile(r"SpawnProcess-(\d+)")


class JSONFormatter(logging.Formatter):
    """A logging formatter that outputs logs in JSON format."""
    def format(self, record: logging.LogRecord) -> str:
        log_record = {
            "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "name": record.name,
            # "taskName": getattr(record, "taskName", None),  # Record task name if needed
            "processName": record.processName,  # Record process name if needed
            "tracking_id": getattr(record, "tracking_id", None),
            "loc": "%s:%d" % (record.filename, record.lineno),
            "func": record.funcName,
            "message": record.getMessage(),
        }

        return json.dumps(log_record, ensure_ascii=False, default=str)


class TrackingIDFilter(logging.Filter):
    """A logging filter that adds tracking_id to log records.
    """
    def filter(self, record):
        record.tracking_id = TRACKING_ID.get()
        return True


def _setup_console_handler(level: int) -> logging.StreamHandler:
    """Setup a StreamHandler for console logging.
    
    Args:
        level (int): The logging level.
    """
    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(level)
    handler.setFormatter(JSONFormatter())
    return handler


def _setup_file_handler(
    level: int, log_path: str, rotate_days: int
) -> TimedRotatingFileHandler:
    """Setup a TimedRotatingFileHandler for logging.
    
    Args:
        level (int): The logging level.
        log_path (str): The log path.
        rotate_days (int): The number of days to keep log files.
    """
    handler = TimedRotatingFileHandler(
        filename=log_path,
        when="midnight",
        interval=1,
        backupCount=rotate_days,
        encoding="utf-8",
    )
    handler.setLevel(level)
    handler.setFormatter(JSONFormatter())
    return handler


def _setup_queue_handler(level: int, log_queue: Queue) -> QueueHandler:
    """Setup a QueueHandler for logging.
    
    Args:
        level (int): The logging level.
        log_queue (Queue): The log queue.
    """
    handler = QueueHandler(log_queue)
    handler.setLevel(level)
    return handler


def _get_spawn_process_number(name: str) -> str:
    """
    Get the spawn process number for log file naming.
    The server should be started with multiple processes using uvicorn's --workers option.
    Prevent issues caused by multiple processes writing to the same log file.

    Args:
        name (str): The name of the log file.

    Returns:
        str: The spawn process number for log file naming.
    """
    try:
        process_name = current_process().name
        pid = os.getpid()

        if process_name == "MainProcess":
            return name
        elif m := PATTERN_PROCESS_NAME.match(process_name):
            return f"{name}-sp{m.group(1)}"
        else:
            return f"{name}-{pid}"

    except:
        return f"{name}-{os.getpid()}"


def _setup_logpath(log_dir: str, name: str) -> str:
    """Setup the log path.
    
    Args:
        log_dir (str): The log directory.
        name (str): The name of the log file. Example: "app"

    Returns:
        str: The log path.
    """
    main_name = _get_spawn_process_number(name)
    log_file = f"{main_name}.log"
    log_path = Path(log_dir) / log_file

    if not log_path.parent.exists():
        try:
            log_path.parent.mkdir(parents=True, exist_ok=True)
        except Exception as e:
            raise RuntimeError(
                f"Failed to create log directory: {log_path.parent}"
            ) from e
    return str(log_path)


def _validate(level: int, enable_console: bool, enable_file: bool, rotate_days: int) -> None:
    """Validate the log configuration.
    
    Args:
        level (int): The logging level.
        enable_console (bool): Whether to enable console logging.
        enable_file (bool): Whether to enable file logging.
        rotate_days (int): The number of days to keep log files.

    Raises:
        ValueError: If the log configuration is invalid.
    """
    if not enable_console and not enable_file:
        raise ValueError("At least one of enable_console or enable_file must be True.")

    if level not in [
        logging.DEBUG,
        logging.INFO,
        logging.WARNING,
        logging.ERROR,
        logging.CRITICAL,
    ]:
        raise ValueError("Invalid logging level specified.")

    if not isinstance(rotate_days, int) or rotate_days <= 0:
        raise ValueError("rotate_days must be a positive integer.")


def setup_logger(
    name: str = "app",
    level: int = logging.DEBUG,
    enable_console: bool = True,
    enable_file: bool = True,
    log_dir: str = "logs",
    rotate_days: int = 7,
) -> logging.Logger:
    """Setup a logger with console and/or file handlers.
    
    Args:
        name (str): The name of the logger. This will be used as the log file name prefix. Defaults to "app".
        level (int): The logging level. Defaults to logging.DEBUG.
        enable_console (bool): Whether to enable console logging. Defaults to True.
        enable_file (bool): Whether to enable file logging. Defaults to True.
        log_dir (str): The log directory. Defaults to "logs".
        rotate_days (int): The number of days to keep log files. Defaults to 7.

    Returns:
        logging.Logger: The configured logger.
    """
    logger = logging.getLogger(name)

    if logger.hasHandlers():
        return logger  # Logger is already configured

    _validate(level, enable_console, enable_file, rotate_days)

    logger.setLevel(level)
    logger.propagate = False  # Prevent log messages from being propagated to the root logger

    log_path = _setup_logpath(log_dir, name)

    handlers = []

    if enable_console:
        handlers.append(_setup_console_handler(level))

    if enable_file:
        handlers.append(_setup_file_handler(level, log_path, rotate_days))

    global _queue_logger, _queue_listener
    if not _queue_logger:
        _queue_logger = Queue(-1)

    queue_handler = _setup_queue_handler(level, _queue_logger)

    if not _queue_listener:
        _queue_listener = QueueListener(
            _queue_logger, *handlers, respect_handler_level=True
        )
        _queue_listener.start()

    logger.addHandler(queue_handler)
    logger.addFilter(TrackingIDFilter())

    return logger


def close_log_queue() -> None:
    """Close the log queue and stop the listener.
    This function should be called when the application is shutting down to ensure that the log queue is closed and the listener is stopped.
    """
    global _queue_listener
    if _queue_listener:
        _queue_listener.stop()
        _queue_listener = None

2. 编辑 pkg/log/__init__.py，便于其它模块导入，个人一般也是通过这个文件告诉别人这个模块哪些对象是 public 的。logger 对象是一个单例，其它模块直接使用即可，一般无需使用 setup_logger 单独创建日志记录器对象。不过有的需求是各个类对象创建各自的日志记录器，所以也对外提供出去了（个人认为没啥必要）。

from pathlib import Path

from .log import close_log_queue, setup_logger

logger = setup_logger(
    log_dir=str(Path(__file__).parent.parent.parent / "logs"),
)

__all__ = [
    "logger",
    "setup_logger",
    "close_log_queue",
]

3. 通过 FastAPI 的 lifespan 来调用 close_log_queue

from contextlib import asynccontextmanager

from pkg.log import close_log_queue

@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        yield
    finally:
        close_log_queue()
        print("Shutdown!")


app = FastAPI(lifespan=lifespan)

4. 将 print() 改成 logger.info()，测试日志中是否自动输出 tracking_id。其中部分改动如下

async def mock_db_query():
    await asyncio.sleep(1)
    current_id = TRACKING_ID.get()
    logger.info(f"This is mock_db_query. Current tracking ID: {current_id}")
    await asyncio.sleep(1)

5. 使用 curl 测试并查看日志

curl 命令输出如下：

$ curl -i  http://127.0.0.1:8000/asdf
HTTP/1.1 200 OK
date: Sat, 29 Nov 2025 13:07:09 GMT
content-length: 62
content-type: text/plain; charset=utf-8
x-tracking-id: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb

Get request, tracking ID: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb

服务端的控制台输出如下：

{"@timestamp": "2025-11-29T21:07:09+0800", "level": "INFO", "name": "app", "processName": "SpawnProcess-3", "tracking_id": "38a015f7-b0d3-41ea-a2b3-179bf608b4bb", "loc": "main.py:39", "func": "get_asdf", "message": "This is get asdf. tracking ID: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb"}
{"@timestamp": "2025-11-29T21:07:10+0800", "level": "INFO", "name": "app", "processName": "SpawnProcess-3", "tracking_id": "38a015f7-b0d3-41ea-a2b3-179bf608b4bb", "loc": "handlers.py:12", "func": "mock_db_query", "message": "This is mock_db_query. Current tracking ID: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb"}

日志文件输出如下：

{"@timestamp": "2025-11-29T21:07:09+0800", "level": "INFO", "name": "app", "processName": "SpawnProcess-3", "tracking_id": "38a015f7-b0d3-41ea-a2b3-179bf608b4bb", "loc": "main.py:39", "func": "get_asdf", "message": "This is get asdf. tracking ID: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb"}
{"@timestamp": "2025-11-29T21:07:10+0800", "level": "INFO", "name": "app", "processName": "SpawnProcess-3", "tracking_id": "38a015f7-b0d3-41ea-a2b3-179bf608b4bb", "loc": "handlers.py:12", "func": "mock_db_query", "message": "This is mock_db_query. Current tracking ID: 38a015f7-b0d3-41ea-a2b3-179bf608b4bb"}

根据以上三条输出可见，tracking_id 在日志中和日志消息中保持一致。

补充: 另一种解决多进程文件写入问题的方法

费劲调试完上面的日志模块中，想到另一个解决多进程文件写入问题的方法，那就是由外部工具写日志，服务只输出到控制台，类似 docker 和 k8s 运行环境的解决方案。以下只是思路，大致想来应该没什么问题，感兴趣的话可以尝试一下。

1. 移除日志模块中的写文件功能，保留输出到控制台的部分。注意标准输出也有可能阻塞异步协程，所以队列处理器还是要保留的。
2. 启动时用 nohup 将控制台输出重定向到文件中。比如：nohup python main.py > logs/start.log 2>&1 &
3. 配置 logrotate 来做日志轮转。