4、配置结构化日志：使用 python-json-logger 库输出 JSON 格式日志

在集成测试与微服务架构中，非结构化的文本日志难以被日志收集系统（如 ELK、Loki）高效解析。使用 python-json-logger 库可以将日志记录直接输出为 JSON 格式，使每条日志都成为结构化的键值对集合，便于后续的过滤、聚合与告警。

4.1 安装与基础配置

首先通过 pip 安装依赖：

pip install python-json-logger

在 Python 代码中，核心配置步骤包括：创建 JsonFormatter 实例、将其绑定到 StreamHandler，最后将 Handler 添加到 Logger。以下是一个可直接运行的配置示例：

import logging
from pythonjsonlogger import jsonlogger

# 1. 获取或创建 Logger
logger = logging.getLogger("integration_test")
logger.setLevel(logging.DEBUG)

# 2. 创建控制台 Handler
handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)

# 3. 定义 JSON 格式化器
formatter = jsonlogger.JsonFormatter(
    fmt="%(asctime)s %(name)s %(levelname)s %(message)s %(module)s %(funcName)s",
    datefmt="%Y-%m-%dT%H:%M:%S"
)
handler.setFormatter(formatter)

# 4. 将 Handler 添加到 Logger
if not logger.handlers:
    logger.addHandler(handler)

# 5. 使用示例
logger.info("用户登录成功", extra={"user_id": 1001, "ip": "192.168.1.10"})
logger.error("数据库连接超时", extra={"db_host": "pg-master-01", "timeout_ms": 5000})

输出效果：

{"asctime": "2025-03-21T14:32:10", "name": "integration_test", "levelname": "INFO", "message": "用户登录成功", "module": "test_auth", "funcName": "test_login", "user_id": 1001, "ip": "192.168.1.10"}
{"asctime": "2025-03-21T14:32:11", "name": "integration_test", "levelname": "ERROR", "message": "数据库连接超时", "module": "test_db", "funcName": "test_connection", "db_host": "pg-master-01", "timeout_ms": 5000}
最佳实践： 在集成测试中，建议在 conftest.py 的 pytest_configure 钩子中一次性完成 Logger 配置，避免每个测试模块重复初始化。

4.2 关键参数与字段映射

JsonFormatter 的 fmt 参数使用标准 %(key)s 占位符，支持所有 LogRecord 属性。下表列出最常用的内置字段与自定义扩展方式：

字段名 占位符 说明 示例值
时间戳 %(asctime)s ISO 8601 格式，通过 datefmt 控制精度 2025-03-21T14:32:10
日志级别 %(levelname)s INFO / WARNING / ERROR 等 ERROR
Logger 名称 %(name)s 通常使用模块名或测试套件名 integration_test
消息正文 %(message)s 日志描述文本 数据库连接超时
模块名 %(module)s 自动提取调用栈中的模块 test_db
函数名 %(funcName)s 自动提取调用栈中的函数 test_connection
自定义字段 extra={} 通过 extra 字典传入任意键值对 {"user_id": 1001}
注意： 如果 extra 中的键与 fmt 中的内置占位符同名，extra 值会覆盖内置字段。建议自定义字段使用下划线前缀（如 _user_id）或全小写命名以避免冲突。

4.3 结构化日志在集成测试中的典型流程

下图展示了从测试执行到日志消费的完整链路，其中 JSON 格式化器是连接测试代码与日志基础设施的关键桥梁：

结构化日志集成测试流程 测试代码 logger.info(...) JsonFormatter 序列化为 JSON StreamHandler stdout / stderr 日志收集系统 ELK / Loki 测试报告与告警 失败分析 / 告警触发 说明：JsonFormatter 将 LogRecord 转为 JSON 字符串，经 StreamHandler 输出后由日志收集系统统一消费

4.4 进阶配置：自定义字段与日志上下文

在复杂的集成测试场景中，往往需要为每条日志注入全局上下文（如测试用例 ID、环境名称、构建版本）。推荐使用 logging.Filter 或 extra 参数实现：

import logging
from pythonjsonlogger import jsonlogger

class TestContextFilter(logging.Filter):
    """为所有日志记录注入测试上下文"""
    def filter(self, record):
        record.test_run_id = getattr(record, 'test_run_id', 'unknown')
        record.environment = getattr(record, 'environment', 'staging')
        return True

# 配置 Logger
logger = logging.getLogger("integration_test")
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    fmt="%(asctime)s %(levelname)s %(message)s %(test_run_id)s %(environment)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.addFilter(TestContextFilter())

# 使用：自动携带上下文
logger.info("外部 API 调用成功", extra={"api": "/v2/orders", "latency_ms": 120})

输出：

{"asctime": "2025-03-21T14:32:12", "levelname": "INFO", "message": "外部 API 调用成功", "test_run_id": "unknown", "environment": "staging", "api": "/v2/orders", "latency_ms": 120}
关键要点：
  • JsonFormatter 的 fmt 中声明的字段必须存在于 LogRecord 中，否则输出 null。
  • 使用 Filter 注入全局字段，避免在每个 extra 中重复传递。
  • 生产环境建议将日志输出到文件或直接发送到日志收集 Agent，而非仅输出到控制台。

4.5 配置清单速查

步骤 操作 关键代码
1 安装库 pip install python-json-logger
2 创建 Logger logger = logging.getLogger("name")
3 创建 Handler handler = logging.StreamHandler()
4 实例化 JsonFormatter JsonFormatter(fmt="...", datefmt="...")
5 绑定 Formatter 与 Handler handler.setFormatter(formatter)
6 添加 Handler 到 Logger logger.addHandler(handler)
7 （可选）添加 Filter logger.addFilter(MyFilter())
8 记录日志 logger.info("msg", extra={...})

完成以上配置后，集成测试的所有日志将自动以 JSON 格式输出，为后续的自动化分析、告警与故障排查奠定坚实基础。

← 上一章 📖 返回目录 下一章 →