Python logging 模块：makeLogRecord 函数解析与应用

2. 日志回放与测试

在开发或调试阶段，你可能需要重现某个日志事件。可以将之前保存的日志字典（例如从日志文件中提取）用 makeLogRecord 重建，再通过当前的日志系统处理，以验证日志处理逻辑是否正确。

Demo 1：模拟远程日志接收

本示例模拟一个客户端发送日志字典，服务器接收后重建并输出。

import logging
import pickle  # 用于序列化（实际可用 JSON）

# ---------- 服务器端配置 ----------
def setup_server_logging():
    """配置服务器 logger，输出到控制台"""
    logger = logging.getLogger('server')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return logger

# ---------- 客户端模拟 ----------
def client_send_log():
    """创建日志字典并序列化（模拟网络发送）"""
    log_dict = {
        'name': 'client_app',
        'levelno': logging.INFO,
        'levelname': 'INFO',
        'pathname': '/path/to/client.py',
        'lineno': 42,
        'msg': 'User logged in',
        'args': (),
        'exc_info': None,
        'func': 'login',
        'sinfo': None,
        # 自定义属性
        'user_id': 'alice',
        'session': 'abc123'
    }
    # 使用 pickle 序列化（实际传输中可用 JSON 等）
    data = pickle.dumps(log_dict)
    return data

# ---------- 服务器接收 ----------
def server_receive(data):
    """反序列化并用 makeLogRecord 重建，然后交给 logger 处理"""
    log_dict = pickle.loads(data)
    record = logging.makeLogRecord(log_dict)
    logger = logging.getLogger('server')
    logger.handle(record)  # 将记录发送给所有已注册的处理器

# ---------- 主程序 ----------
if __name__ == '__main__':
    server_logger = setup_server_logging()
    data = client_send_log()
    server_receive(data)

逐行解析

• setup_server_logging：创建名为 'server' 的 logger，添加控制台处理器，设置格式。这是接收端的日志配置。
• client_send_log：
  • 构造一个字典 log_dict，包含标准 LogRecord 属性（如 name、levelno、msg）和两个自定义字段 user_id 和 session。
  • 使用 pickle.dumps 将字典序列化为字节流，模拟网络传输。
• server_receive：
  • 反序列化字节流得到字典。
  • 调用 logging.makeLogRecord(log_dict) 重建 LogRecord 对象。
  • 获取名为 'server' 的 logger，调用其 handle 方法将记录交给处理器（控制台输出）。
• 主程序依次执行配置、获取数据、接收处理。

运行后控制台输出：

2025-02-24 xx:xx:xx,xxx - server - INFO - User logged in

注意：自定义字段未在格式中出现，但它们已存在于记录中，可在自定义格式化器中使用。

Demo 2：从文件读取日志字典进行回放

假设已有保存的日志字典文件（每行一个 JSON 对象），现在需要重新处理它们以测试新的日志处理器。

import logging
import json

# ---------- 配置回放用的日志系统 ----------
def setup_replay_logging():
    """配置一个输出到文件的 logger，格式中包含自定义字段 user_id"""
    logger = logging.getLogger('replay')
    logger.setLevel(logging.DEBUG)
    handler = logging.FileHandler('replay.log')
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s - user=%(user_id)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return logger

# ---------- 从文件读取日志字典 ----------
def read_log_dicts_from_file(filename):
    """生成器，逐行读取 JSON 文件并产生字典"""
    with open(filename, 'r') as f:
        for line in f:
            yield json.loads(line)

# ---------- 回放函数 ----------
def replay_logs(logger, dicts):
    """遍历字典，重建 LogRecord 并交给 logger 处理"""
    for d in dicts:
        record = logging.makeLogRecord(d)
        logger.handle(record)

# ---------- 主程序 ----------
if __name__ == '__main__':
    # 准备示例日志字典并写入文件
    sample_dicts = [
        {'name': 'app', 'levelno': 20, 'levelname': 'INFO', 'pathname': 'main.py', 'lineno': 10, 'msg': 'Start', 'args': (), 'exc_info': None, 'func': 'main', 'user_id': 'alice'},
        {'name': 'app', 'levelno': 30, 'levelname': 'WARNING', 'pathname': 'main.py', 'lineno': 15, 'msg': 'Disk low', 'args': (), 'exc_info': None, 'func': 'check', 'user_id': 'bob'}
    ]
    with open('logs.jsonl', 'w') as f:
        for d in sample_dicts:
            f.write(json.dumps(d) + '\n')
    
    # 设置回放 logger
    logger = setup_replay_logging()
    # 读取字典并回放
    dicts = read_log_dicts_from_file('logs.jsonl')
    replay_logs(logger, dicts)

逐行解析

• setup_replay_logging：创建名为 'replay' 的 logger，添加一个写入 replay.log 的文件处理器，格式中包含 user_id 字段。
• read_log_dicts_from_file：生成器函数，逐行读取 JSON Lines 文件，每行解析为一个字典。
• replay_logs：遍历字典生成器，对每个字典调用 logging.makeLogRecord 重建记录，然后使用 logger 的 handle 方法处理。
• 主程序：
  • 首先创建一个示例文件 logs.jsonl，包含两条日志字典。
  • 调用 setup_replay_logging 获取 logger。
  • 从文件读取字典并执行回放。

运行后，replay.log 文件中将包含两条记录，例如：

2025-02-24 xx:xx:xx,xxx - replay - INFO - Start - user=alice
2025-02-24 xx:xx:xx,xxx - replay - WARNING - Disk low - user=bob

验证了回放成功。

注意事项

1. 字典必须包含足够的信息：虽然 makeLogRecord 会创建默认值，但如果缺少关键属性（如 name、levelno、msg），后续处理可能出错。
2. 自定义字段的处理：重建的记录包含自定义字段，但需要在 Formatter 中显式引用（如 %(user_id)s）才能输出。
3. 异常信息：如果字典中包含 exc_info，通常应为三元组或 None。直接使用 pickle 序列化异常对象可能不可靠，建议使用字符串形式。
4. 与 setLogRecordFactory 的交互：makeLogRecord 使用当前的 _logRecordFactory，因此如果自定义了工厂，它会影响重建的记录。

总结

makeLogRecord 是一个简单但功能强大的函数，它提供了从字典到 LogRecord 的桥梁，使得日志事件可以脱离原始进程进行传输、存储和重放。结合序列化工具，可以轻松构建分布式日志系统或用于测试和调试。掌握这一工具，能让你更灵活地处理日志数据。