Python 日志处理：logging 模块配置、JSON 日志、ELK 集成

小飞兽 Python 7 次阅读 2026-07-24

Introduction

日志是生产环境调试的第一工具。本文讲解 logging 模块的全面配置、自定义 Formatter、JSON 日志格式（方便 ELK 采集）、日志轮转（RotatingFileHandler）、以及多模块日志配置。

---

logging 模块基础

import logging, sys

# 基本配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)

logger = logging.getLogger(__name__)

logger.debug('调试信息')
logger.info('一般信息')
logger.warning('警告信息')
logger.error('错误信息')
logger.critical('严重错误')

自定义 Formatter

import logging, sys
from datetime import datetime

class MyFormatter(logging.Formatter):
    colors = {
        'DEBUG': '[36m',    # 青色
        'INFO': '[32m',     # 绿色
        'WARNING': '[33m', # 黄色
        'ERROR': '[31m',   # 红色
        'CRITICAL': '[35m',# 紫色
    }
    RESET = '[0m'

    def format(self, record):
        levelname = record.levelname
        if levelname in self.colors:
            record.levelname = f'{self.colors[levelname]}{levelname}{self.RESET}'
        return super().format(record)

# 带颜色的日志
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(MyFormatter(
    fmt='%(asctime)s [%(levelname)s] %(name)s - %(message)s',
    datefmt='%H:%M:%S'
))

JSON 日志（ELK 友好）

import logging, json, sys
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
        }
        if record.exc_info:
            log_obj['exception'] = self.formatException(record.exc_info)
        if hasattr(record, 'user_id'):
            log_obj['user_id'] = record.user_id
        if hasattr(record, 'request_id'):
            log_obj['request_id'] = record.request_id
        return json.dumps(log_obj, ensure_ascii=False)

# 配置
logger = logging.getLogger('api')
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# 使用
logger.info('用户登录', extra={'user_id': 123})  # extra 字段会进入 JSON

日志轮转（防止日志文件过大）

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

# 按大小轮转（超过 10MB 则创建新文件，保留 5 个备份）
rotating = RotatingFileHandler(
    'app.log', maxBytes=10*1024*1024, backupCount=5,
    encoding='utf-8'
)
rotating.setLevel(logging.INFO)

# 按时间轮转（每天零点创建新文件，保留 30 天）
timed = TimedRotatingFileHandler(
    'app.log', when='midnight', interval=1, backupCount=30,
    encoding='utf-8'
)
timed.setLevel(logging.INFO)

logger = logging.getLogger('myapp')
logger.addHandler(timed)

多模块日志配置

# main.py（统一配置）
import logging, logging.config, yaml

with open('logging.yaml') as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)

# mylib.py（直接使用即可，继承 main.py 的配置）
import logging
logger = logging.getLogger('mylib')
logger.info('来自 mylib 的日志')

# logging.yaml
version: 1
disable_existing_loggers: false
formatters:
  standard:
    format: '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
handlers:
  console:
    class: logging.StreamHandler
    formatter: standard
    stream: ext://sys.stdout
  file:
    class: logging.handlers.RotatingFileHandler
    formatter: standard
    filename: app.log
    maxBytes: 10485760
    backupCount: 5
root:
  level: INFO
  handlers: [console, file]
loggers:
  mylib:
    level: DEBUG
    handlers: [file]
    propagate: false

structlog（更现代的日志方案）

pip install structlog

import structlog, logging

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt='iso'),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

log = structlog.get_logger()
log.info('user_action', user_id=123, action='login', duration_ms=250)

常见问题

Q1: 线上日志太多影响性能？用 logging.getLogger('mylib') 设置合适的 level（INFO 而非 DEBUG），并用 RotatingFileHandler 限制文件大小。

Q2: 日志怎么和 ELK（Elasticsearch + Logstash + Kibana）集成？JSON 格式日志 → Filebeat 采集 → Logstash 解析 → Elasticsearch 存储 → Kibana 可视化。

Q3: 如何给日志加请求追踪 ID？用 contextvars 或在请求入口生成 UUID，放进 structlog 的 context，后续所有日志自动带上 request_id。