如何为logging.Formatter注入特定变量，无需额外参数实现全链路日志标识

要实现多线程下给目标函数及其所有子调用（包括仅用logging的第三方库）的日志统一添加foo_id，核心思路是用线程本地存储隔离不同线程的标识，再通过日志过滤器自动给每条日志注入标识，无需手动传递参数。

实现步骤 #

1. 线程本地存储：用threading.local()存储当前线程的foo_id，保证多线程之间标识互不干扰。
2. 日志过滤器：自定义过滤器，从线程本地存储中取出foo_id并添加到日志记录的extra字段。
3. 上下文管理器：封装标识的设置与清理逻辑，让函数调用时自动生效，避免污染后续线程或调用。

完整代码示例 #

boo.py（无需修改，模拟第三方库代码）

import logging

def fn():
    logging.debug('Some module debug message')
    logging.info('Some module info message')

main.py（核心实现）

import logging
import threading
from contextlib import contextmanager
import boo

# 线程本地存储，保存当前线程的foo_id
thread_local = threading.local()

# 自定义日志过滤器：自动给日志添加fooid字段
class FooIdFilter(logging.Filter):
    def filter(self, record):
        # 从线程本地取foo_id，不存在则设为空字符串
        record.fooid = getattr(thread_local, 'foo_id', '')
        return True

# 上下文管理器：自动设置/清理线程本地的foo_id
@contextmanager
def set_foo_id(foo_id):
    original_id = getattr(thread_local, 'foo_id', None)
    thread_local.foo_id = foo_id
    try:
        yield
    finally:
        # 恢复原有状态，避免污染后续调用
        if original_id is not None:
            thread_local.foo_id = original_id
        else:
            del thread_local.foo_id

# 日志基础配置
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# 给根logger添加过滤器，让所有子logger（包括第三方库的）都生效
root_logger = logging.getLogger()
root_logger.addFilter(FooIdFilter())

formatter = logging.Formatter('%(asctime)s - %(fooid)s - %(levelname)s - %(message)s')

file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)

logger.addHandler(file_handler)

def foo(foo_id: str):
    # 用上下文管理器自动绑定foo_id
    with set_foo_id(foo_id):
        logger.debug('Debug message')
        logger.info('Info message')
        logger.warning('Warning message')
        logger.error('Error message')
        logger.critical('Critical message')

        boo.fn()

# 单线程测试
foo('test run')

# 多线程测试（验证线程隔离）
import threading
def thread_task(foo_id):
    foo(foo_id)

thread1 = threading.Thread(target=thread_task, args=('thread-1',))
thread2 = threading.Thread(target=thread_task, args=('thread-2',))
thread1.start()
thread2.start()
thread1.join()
thread2.join()

关键说明 #

• 线程隔离：threading.local()确保每个线程的foo_id独立，多线程同时调用foo时不会互相覆盖。
• 全局生效：给根logger添加过滤器后，所有基于根logger的子日志实例（包括第三方库直接使用的logging模块）都会自动带上fooid字段，无需修改第三方代码。
• 安全清理：上下文管理器会在函数退出（包括异常退出）时恢复线程原有状态，避免标识污染后续请求。

输出效果（app.log） #

2024-04-12 19:05:49,792 - test run - DEBUG - Debug message
2024-04-12 19:05:49,792 - test run - INFO - Info message
2024-04-12 19:05:49,792 - test run - WARNING - Warning message
2024-04-12 19:05:49,792 - test run - ERROR - Error message
2024-04-12 19:05:49,792 - test run - CRITICAL - Critical message
2024-04-12 19:05:49,792 - test run - DEBUG - Some module debug message
2024-04-12 19:05:49,792 - test run - INFO - Some module info message
2024-04-12 19:06:00,123 - thread-1 - DEBUG - Debug message
... # thread-1的其他日志内容
2024-04-12 19:06:00,125 - thread-2 - DEBUG - Debug message
... # thread-2的其他日志内容

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