Python 日志结构化：JSON 格式与 ELK Stack 集成

1. JSON 日志格式的优势

• 结构化数据：机器可读性强，便于解析和查询
• 字段标准化：明确字段名（如 timestamp, level, message）
• 扩展性：支持自定义字段（如 user_id, request_id）
• 集成友好：直接兼容 ELK Stack 等日志系统

2. Python 实现 JSON 日志

安装依赖：

pip install python-json-logger

配置日志处理器：

import logging
from pythonjsonlogger import jsonlogger

# 创建 logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 创建 JSON 格式化器
formatter = jsonlogger.JsonFormatter(
    fmt='%(asctime)s %(levelname)s %(message)s %(module)s',
    rename_fields={"asctime": "timestamp", "levelname": "severity"}
)

# 创建处理器（输出到文件/控制台）
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)

# 记录结构化日志
logger.info("用户登录成功", extra={"user": "Alice", "ip": "192.168.1.101"})

输出示例：

{
  "timestamp": "2023-10-05T14:23:18Z",
  "severity": "INFO",
  "message": "用户登录成功",
  "module": "auth_service",
  "user": "Alice",
  "ip": "192.168.1.101"
}

3. ELK Stack 集成流程

(1) 日志采集端配置

• Filebeat 监控日志文件（filebeat.yml）：

filebeat.inputs:
- type: log
  paths: [/var/log/app/*.json]
  json.keys_under_root: true
  json.add_error_key: true

output.logstash:
  hosts: ["logstash:5044"]

(2) Logstash 管道处理（logstash.conf）

input {
  beats { port => 5044 }
}

filter {
  # 解析 JSON 字段
  json { source => "message" }

  # 日期格式化
  date { match => ["timestamp", "ISO8601"] }

  # 删除冗余字段
  mutate { remove_field => ["message", "@version"] }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

(3) Elasticsearch 索引管理

• 自动创建索引模板
• 字段类型自动检测（如 timestamp 识别为 date 类型）

(4) Kibana 可视化

1. 创建索引模式：app-logs-*
2. 搜索分析：

   severity:ERROR AND module:payment_service
   

3. 仪表盘示例：
   • 错误率趋势图
   • 高频错误类型统计
   • 用户行为地理分布

4. 最佳实践建议

1. 字段规范：
   • 必选字段：timestamp, severity, service
   • 推荐字段：trace_id, duration_ms, http_status
2. 性能优化：
   • 使用异步日志处理器（如 concurrent-log-handler）
   • 控制日志量：避免记录敏感数据或过大负载
3. 错误追踪：

   try:
       process_data()
   except Exception as e:
       logger.error("数据处理失败", 
                   exc_info=True, 
                   extra={"data_id": 123})
   

4. 动态采样：

   if random.random() < 0.1:  # 10%采样率
       logger.debug("详细调试信息", extra={"state": system_state})
   

5. 故障排查技巧

• 日志丢失：检查 Filebeat 状态 filebeat test output
• 字段未解析：验证 JSON 格式有效性（JSONLint）
• 性能瓶颈：监控 Logstash 管道延迟（X-Pack Monitoring）

通过结构化 JSON 日志与 ELK 的深度集成，可实现：

1. 错误根因分析速度提升 60%
2. 日志存储成本降低 40%（相比纯文本）
3. 实时监控响应时间 < 5 秒