Python 3.12 logging - 12 - makeLogRecord

Python 3.12 logging 的 makeLogRecord

引言

在 Python 的 logging 模块中，makeLogRecord 是一个用于从字典重建日志记录的函数。它将一个包含日志事件属性的字典转换为一个 LogRecord 对象，从而可以在本地日志系统中重新处理该事件。这在分布式日志收集、日志回放或测试等场景中非常有用。

函数签名

logging.makeLogRecord(attrs:dict)-> logging.LogRecord 

• 参数 attrs：一个字典，包含要创建的 LogRecord 的所有属性。常见的属性包括 name、levelno、pathname、lineno、msg、args、exc_info 等，也可以是自定义字段。
• 返回值：一个新的 LogRecord 实例，其属性由传入的字典决定。

内部实现解析

makeLogRecord 的源码如下：

defmakeLogRecord(dict):""" Make a LogRecord whose attributes are defined by the specified dictionary, This function is useful for converting a logging event received over a socket connection (which is sent as a dictionary) into a LogRecord instance. 创建一个日志记录实例，其属性由指定字典定义。 该函数可用于将通过套接字连接接收的日志事件（以字典形式发送）转换为日志记录实例。 """ rv = _logRecordFactory(None,None,"",0,"",(),None,None) rv.__dict__.update(dict)return rv 

• _logRecordFactory 是当前设置的 LogRecord 工厂（默认为 LogRecord 类）。调用它时传入一些默认参数，创建一个“空壳”记录对象。
• 然后，用传入的字典更新该对象的 __dict__，将字典中的键值对设为记录对象的属性。
• 最后返回这个填充好的记录对象。

这种设计保证了即使字典中缺少某些字段，记录对象仍然有一个有效的初始状态，不会因为缺少必需属性而崩溃。

应用场景

1. 远程日志处理

在分布式系统中，各服务可以将日志事件序列化为字典（如 JSON），通过网络发送到中央日志服务。中央服务使用 makeLogRecord 重建 LogRecord，然后应用本地配置的处理器进行统一存储和分析。

2. 日志回放与测试

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

Demo 1：模拟远程日志接收

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

import logging import pickle # 用于序列化（实际可用 JSON）# ---------- 服务器端配置 ----------defsetup_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 # ---------- 客户端模拟 ----------defclient_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 # ---------- 服务器接收 ----------defserver_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 # ---------- 配置回放用的日志系统 ----------defsetup_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 # ---------- 从文件读取日志字典 ----------defread_log_dicts_from_file(filename):"""生成器，逐行读取 JSON 文件并产生字典"""withopen(filename,'r')as f:for line in f:yield json.loads(line)# ---------- 回放函数 ----------defreplay_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'}]withopen('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 的桥梁，使得日志事件可以脱离原始进程进行传输、存储和重放。结合序列化工具，可以轻松构建分布式日志系统或用于测试和调试。掌握这一工具，能让你更灵活地处理日志数据。

本篇博客中，如有存在偏差或是错误的地方，请私信我哈，参与评论，欢迎指正。