Python Logging 通过 JSON 配置文件及 Logstash 集成指南

山海之间

发布时间：2026-10-13| 2人浏览过

原创

本文详解如何在 Python 中使用 JSON 文件配置 logging 模块，实现日志的灵活管理。同时介绍如何将日志格式化为 JSON 字符串以对接 Logstash，避免 Grok 解析带来的 CPU 开销。内容涵盖配置文件编写、代码调用示例以及 python-logstash 模块的使用，帮助开发者在 Kibana 中更高效地搜索和分析结构化日志数据。

Python 日志 JSON 配置文件与基础调用

logconfig.json

JSON 配置文件结构详解


{
 "version":1,
 "disable_existing_loggers":false,
 "formatters":{
 "simple":{
  "format":"[%(asctime)s - %(levelname)s - line(%(lineno)d) - %(filename)s]: %(message)s",
  "datefmt":"%Y-%m-%d %H:%M:%S"
 }
 },
 "handlers":{
 "console":{
  "class":"logging.StreamHandler",
  "level":"DEBUG",
  "formatter":"simple",
  "stream":"ext://sys.stdout"
 },
 "info_file_handler":{
  "class":"logging.handlers.TimedRotatingFileHandler",
  "level":"INFO",
  "formatter":"simple",
  "filename":"../log/info.log",
  "when":"H",
  "interval":1,
  "backupCount":50,
  "encoding":"utf8"
 },
 "error_file_handler":{
  "class":"logging.handlers.TimedRotatingFileHandler",
  "level":"ERROR",
  "formatter":"simple",
  "filename":"../log/errors.log",
  "when":"H",
  "interval":1,
  "backupCount":50,
  "encoding":"utf8"
 }
 },
 "loggers":{
 "my_module":{
  "level":"ERROR",
  "handlers":["info_file_handler"],
  "propagate":"no"
 }
 },
 "root":{
 "level":"INFO",
 "handlers":["console","info_file_handler","error_file_handler"]
 }
}

log_utility.py

日志加载工具代码实现


import os
import json
import logging
import logging.config


def setup_logging(default_path="logconfig.json",default_level=logging.DEBUG):
 path = default_path
 if os.path.exists(path):
 with open(path,"r") as f:
  config = json.load(f)
  logging.config.dictConfig(config)
 else:
 logging.basicConfig(level=default_level)

调用

config_path = sys.path[0] + '/logconfig.json' log_utility.setup_logging(config_path)

补充知识：python logging定制logstash的json日志格式

近期完成了日志收集系统的优化工作，本文总结一种针对 Python 环境优化 Logstash 的方案。

众所周知，Logstash 使用 Grok 插件解析日志时需要消耗大量 CPU 资源，因为涉及正则表达式的匹配过程。

基于 Logstash 的调优思路，我们可以预先将日志生成为 JSON 格式。对于 Python 程序，具体如何实现呢？

主要有两种实现方式：第一种是在生成 JSON 后，直接发送至 Logstash 的接收端口；第二种是将 JSON 格式的日志写入文件，让 Logstash 在执行 tail 操作时直接读取整行 JSON 数据。

Python 标准库中的 logging 模块修改为输出 JSON 格式相对容易。此外，已有不少开发者针对此需求开发了 Python 专用的 Logstash 模块。

Logstash TCP 连接示例


import logging
import logstash
import sys

host = 'localhost'

test_logger = logging.getLogger('python-logstash-logger')
test_logger.setLevel(logging.INFO)
test_logger.addHandler(logstash.LogstashHandler(host, 5959, version=1))
# test_logger.addHandler(logstash.TCPLogstashHandler(host, 5959, version=1))

test_logger.error('python-logstash: test logstash error message.')
test_logger.info('python-logstash: test logstash info message.')
test_logger.warning('python-logstash: test logstash warning message.')

# add extra field to logstash message
extra = {
 'test_string': 'python version: ' + repr(sys.version_info),
 'test_boolean': True,
 'test_dict': {'a': 1, 'b': 'c'},
 'test_float': 1.23,
 'test_integer': 123,
 'test_list': [1, 2, '3'],
}

test_logger.info('python-logstash: test extra fields', extra=extra)

python-logstash 库内置了 AMQP 协议的支持方案。

Logstash AMQP 协议示例


import logging
import logstash

# AMQP parameters
host = 'localhost'
username = 'guest'
password= 'guest'
exchange = 'logstash.py'

# get a logger and set logging level
test_logger = logging.getLogger('python-logstash-logger')
test_logger.setLevel(logging.INFO)

# add the handler
test_logger.addHandler(logstash.AMQPLogstashHandler(version=1,
       host=host,
       durable=True,
       username=username,
       password=password,
       exchange=exchange))

# log
test_logger.error('python-logstash: test logstash error message.')
test_logger.info('python-logstash: test logstash info message.')
test_logger.warning('python-logstash: test logstash warning message.')

try:
 1/0
except:
 test_logger.exception('python-logstash: test logstash exception with stack trace')

无论采用何种方式，最终生成的日志格式符合规范即可。

Logstash 输出格式示例


{
 "@source"=>"unknown",
 "@type"=>"nginx",
 "@tags"=>[],
 "@fields"=>{
 "remote_addr"=>"192.168.0.1",
 "remote_user"=>"-",
 "body_bytes_sent"=>"13988",
 "request_time"=>"0.122",
 "status"=>"200",
 "request"=>"GET /some/url HTTP/1.1",
 "request_method"=>"GET",
 "http_referrer"=>"http://www.example.org/some/url",
 "http_user_agent"=>"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/21.0.1180.79 Safari/537.1"
 },
 "@timestamp"=>"2012-08-23T10:49:14+02:00"
}

此处简要说明，虽然该模块可用，但体验并不完美。我在 Python 中将日志输出为 JSON 字符串，原本期望在 Elasticsearch 中能像 Grok 解析那样将日志拆分为独立字段，而非将所有内容堆积在 message 字段中。相信读者已理解我的意图，这会导致在 Kibana 中搜索变得困难。

在 Kibana 进行查询时，我常使用 source:xxx AND level:INFO 这样的语法。然而正如前述，整条日志都被包裹在 @message 字段中，导致无法直接按字段筛选。

以上关于 Python logging 通过 JSON 文件配置及 Logstash 集成的步骤即为全部内容，希望能为读者提供参考，欢迎支持本站。