Python生产环境日志与监控实战指南_Python

一、你大概率也在用print调试

print("开始处理订单:", order_id)
print("调用支付接口...")
print("支付成功:", response)

这个习惯在本地开发没事，但如果你的flask应用已经在生产环境跑了，print有三个致命问题:

日志级别混在一起——你分不清哪个是info、哪个是error
多进程日志错乱——gunicorn 4个worker，print的输出全搅在一起
没有结构化数据——运维问你"今天有多少笔超时订单"，你没法查

这篇文章从print开始，一步步演进到一套可观测的生产级日志+监控方案。代码都可以直接跑。

二、阶段1: print → logging (最基础的升级)

import logging

# 基础配置
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[
        logging.filehandler('app.log'),
        logging.streamhandler()
    ]
)

logger = logging.getlogger(__name__)

# 使用
logger.info("开始处理订单: %s", order_id)
logger.warning("支付接口响应时间: %.2fs", elapsed)
logger.error("支付失败: order=%s, reason=%s", order_id, reason, exc_info=true)

输出效果:

2026-08-06 10:23:15,421 [info] app.payment: 开始处理订单: ord-12345
2026-08-06 10:23:17,892 [warning] app.payment: 支付接口响应时间: 2.47s
2026-08-06 10:23:18,103 [error] app.payment: 支付失败: order=ord-12345, reason=timeout
traceback (most recent call last):
  file "payment.py", line 42, in process_payment
    raise paymenttimeout("gateway timeout")

但这还不够——运维要从日志里提取数据还得用grep。下一步: 结构化日志。

三、阶段2: 结构化日志 (可查询)

import structlog
import json

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.positionalargumentsformatter(),
        structlog.processors.timestamper(fmt="iso"),
        structlog.processors.jsonrenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.loggerfactory(),
)

logger = structlog.get_logger()

# 绑定请求上下文
logger = logger.bind(order_id="ord-12345", user_id="u-567")

# 结构化日志
logger.info("payment_initiated", amount=1250.00, currency="hkd")
logger.warning("payment_slow", elapsed_sec=2.47, gateway="stripe_hk")
logger.error("payment_failed", reason="timeout", retry_count=3)

输出(json格式):

{"timestamp": "2026-08-06t10:23:15.421z", "level": "info", "event": "payment_initiated",
 "order_id": "ord-12345", "user_id": "u-567", "amount": 1250.00, "currency": "hkd"}

现在可以用 jq 或任何日志分析工具直接查询:

# 查今天所有支付失败的订单
cat app.log | jq 'select(.event == "payment_failed") | {order_id, reason}'
# 统计支付接口p99延迟
cat app.log | jq 'select(.event == "payment_slow") | .elapsed_sec' | sort -n | tail -1

四、阶段3: 日志轮转 (别让日志撑爆磁盘)

from logging.handlers import rotatingfilehandler

handler = rotatingfilehandler(
    'app.log',
    maxbytes=50 * 1024 * 1024,  # 50mb
    backupcount=10,               # 保留10个历史文件
    encoding='utf-8'
)
handler.setformatter(logging.formatter(
    '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
))

# 配合timedrotatingfilehandler按时间轮转
from logging.handlers import timedrotatingfilehandler
time_handler = timedrotatingfilehandler(
    'app.log',
    when='midnight',    # 每天午夜轮转
    backupcount=30,     # 保留30天
    encoding='utf-8'
)

root_logger.addhandler(time_handler)

五、阶段4: 自定义metrics (可观测性)

光有日志不够——你需要能在dashbord上看的指标:

from prometheus_client import counter, histogram, gauge, generate_latest
import time
from flask import flask, response

app = flask(__name__)

# 定义指标
payment_total = counter('payment_total', 'total payments', ['status'])
payment_duration = histogram('payment_duration_seconds', 'payment processing time')
active_orders = gauge('active_orders', 'currently processing orders')

@app.route('/process_payment', methods=['post'])
def process_payment():
    active_orders.inc()  # 正在处理的订单+1
    start = time.time()

    try:
        # 支付逻辑...
        payment_total.labels(status='success').inc()
        payment_duration.observe(time.time() - start)
        return {'status': 'ok'}
    except exception:
        payment_total.labels(status='failed').inc()
        raise
    finally:
        active_orders.dec()  # 处理完成-1

# prometheus采集端点
@app.route('/metrics')
def metrics():
    return response(generate_latest(), mimetype='text/plain')

收藏本文——下次搭新服务时直接复制这套日志+监控模板，省一个下午的配置时间。

六、可视化: 生产环境可观测性全景

import matplotlib.pyplot as plt
import matplotlib
matplotlib.rcparams['font.sans-serif'] = ['pingfang sc', 'simhei']
matplotlib.rcparams['axes.unicode_minus'] = false

categories = ['无日志\n(print)', '基础logging', '结构化日志', '+日志轮转', '+metrics']
levels = [0, 1, 2, 3, 4]
# 各维度的能力评分 (0-10)
queryability = [0, 3, 9, 9, 9]
alert_ready = [0, 1, 5, 5, 10]
disk_safety = [0, 3, 3, 10, 10]
ops_efficiency = [0, 2, 7, 8, 10]

fig, ax = plt.subplots(figsize=(14, 6))

x = range(len(categories))
w = 0.2
bars1 = ax.bar([i - 1.5*w for i in x], queryability, w, color='#3498db', edgecolor='white', label='可查询性')
bars2 = ax.bar([i - 0.5*w for i in x], alert_ready, w, color='#e74c3c', edgecolor='white', label='告警就绪')
bars3 = ax.bar([i + 0.5*w for i in x], disk_safety, w, color='#2ecc71', edgecolor='white', label='磁盘安全')
bars4 = ax.bar([i + 1.5*w for i in x], ops_efficiency, w, color='#f39c12', edgecolor='white', label='运维效率')

ax.set_xticks(x)
ax.set_xticklabels(categories, fontsize=11)
ax.set_ylabel('能力评分 (0-10)', fontsize=12)
ax.set_title('python日志方案的五阶段能力演进', fontsize=13, fontweight='bold')
ax.legend(fontsize=9, loc='upper left')
ax.grid(axis='y', alpha=0.3)
ax.set_ylim(0, 12)

for bars in [bars1, bars2, bars3, bars4]:
    for bar in bars:
        if bar.get_height() > 1:
            ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.2,
                    str(int(bar.get_height())), ha='center', fontsize=8, color='#2c3e50')

plt.tight_layout()
plt.savefig('python_logging_evolution.png', dpi=120, bbox_inches='tight', facecolor='white')

七、生产环境日志配置模板 (开箱即用)

# production_logging.py —— 直接复制到你的项目里
import logging.config
import structlog

logging_config = {
    'version': 1,
    'disable_existing_loggers': false,
    'formatters': {
        'json': {
            '()': structlog.stdlib.processorformatter,
            'processor': structlog.processors.jsonrenderer(),
        },
        'console': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        }
    },
    'handlers': {
        'file_json': {
            'class': 'logging.handlers.rotatingfilehandler',
            'filename': 'app.json.log',
            'maxbytes': 50_000_000,
            'backupcount': 10,
            'formatter': 'json',
        },
        'console': {
            'class': 'logging.streamhandler',
            'formatter': 'console',
        }
    },
    'root': {
        'handlers': ['file_json', 'console'],
        'level': 'info',
    },
}

def setup_production_logging():
    logging.config.dictconfig(logging_config)
    structlog.configure(
        processors=[structlog.stdlib.filter_by_level,
                    structlog.stdlib.add_log_level,
                    structlog.processors.timestamper(fmt="iso"),
                    structlog.stdlib.processorformatter.wrap_for_formatter],
        context_class=dict,
        logger_factory=structlog.stdlib.loggerfactory(),
    )
    return structlog.get_logger()

# 一行初始化
logger = setup_production_logging()
logger.info("app_started", env="production", version="2.4.1")