日志系统最佳实践：从入门到精通

在软件开发过程中，日志系统往往是最容易被忽视但又最重要的基础设施之一。良好的日志系统能够帮助我们快速定位问题、监控系统健康状况、分析用户行为。本文将总结多年开发实践中积累的日志系统最佳实践。

日志级别选择

合理使用日志级别是日志设计的第一步。大多数日志框架都提供以下标准级别：

ERROR（错误）

仅记录需要立即处理的错误情况，例如：

应用程序无法启动或继续运行的致命错误
关键业务流程失败（支付失败、数据保存失败）
系统资源耗尽（内存溢出、磁盘空间不足）

示例：

logger.error("Failed to process payment: %s, transaction_id: %s",
            str(e), transaction_id)

WARN（警告）

记录潜在的问题，但系统仍能继续运行：

业务规则异常但仍可继续处理
使用了过时的 API 或配置
资源使用接近阈值

示例：

logger.warning("Using deprecated API endpoint: /api/v1/old-endpoint")
logger.warning("Memory usage at 85%%, consider scaling up")

INFO（信息）

记录重要的业务流程和系统状态：

关键操作的开始和结束（用户登录、订单创建）
系统状态变化（服务启动、配置更新）
定期健康检查结果

示例：

logger.info("User login successful: user_id=%s", user_id)
logger.info("Order created: order_id=%s, amount=%.2f", order_id, amount)

DEBUG（调试）

记录详细的调试信息，用于问题排查：

方法入口和出口（重要方法）
关键变量的值
复杂逻辑的分支路径

示例：

logger.debug("Processing request: method=%s, path=%s", method, path)
logger.debug("Query execution time: %.2fms", query_time)

日志内容设计

包含上下文信息

日志应该包含足够的上下文信息，以便在出现问题时能够快速定位：

# ❌ 不好的日志
logger.error("Database error")

# ✅ 好的日志
logger.error("Database error: query=%s, error=%s, user_id=%s",
             query, str(e), user_id)

使用结构化日志

结构化日志（JSON 格式）便于日志聚合和分析：

import json
import logging

class StructuredFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name,
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }
        if hasattr(record, 'extra'):
            log_data.update(record.extra)
        return json.dumps(log_data)

logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger.addHandler(handler)

避免敏感信息

永远不要在日志中记录敏感信息：

密码、密钥、令牌
完整的信用卡号
个人身份信息（身份证号、护照号）
完整的用户数据

# ❌ 不好的日志
logger.info("User password: %s", password)

# ✅ 好的日志
logger.info("User login attempt: user_id=%s", user_id)

性能优化

使用延迟求值

日志格式化字符串应该使用延迟求值，避免不必要的字符串拼接：

# ❌ 不好的方式
logger.debug("User data: " + str(user_dict))

# ✅ 好的方式
logger.debug("User data: %s", user_dict)

这样，当日志级别高于 DEBUG 时，字符串格式化操作不会执行，节省性能。

异步日志

在生产环境中，应该使用异步日志处理，避免 I/O 操作阻塞主线程：

import logging.handlers

# 使用异步处理器
handler = logging.handlers.QueueHandler(queue)
async_handler = logging.handlers.QueueListener(queue)
logger.addHandler(handler)

合理控制日志量

避免在同一位置输出过多重复日志：

# ❌ 不好的方式
for item in items:
    logger.debug("Processing item: %s", item)

# ✅ 好的方式
logger.debug("Processing batch: total_items=%d", len(items))
for item in items:
    process_item(item)
logger.debug("Batch processing completed")

日志轮转与归档

按大小轮转

当日志文件达到一定大小时进行轮转：

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'app.log',
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5          # 保留5个备份
)

按时间轮转

每天或每小时创建新的日志文件：

from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',       # 每天午夜轮转
    interval=1,            # 间隔1天
    backupCount=30         # 保留30天
)

监控与告警

错误率监控

监控 ERROR 级别日志的数量，设置告警阈值：

class ErrorRateMonitor:
    def __init__(self, threshold=10, window=60):
        self.threshold = threshold
        self.window = window
        self.errors = []

    def log_error(self):
        now = time.time()
        self.errors.append(now)
        # 清理超时的错误记录
        self.errors = [t for t in self.errors if now - t < self.window]

        if len(self.errors) >= self.threshold:
            self.send_alert()

    def send_alert(self):
        # 发送告警通知
        pass

关键指标追踪

为关键业务操作添加专门的成功/失败日志：

logger.info("METRIC: payment_processed, status=success, amount=%.2f", amount)
logger.info("METRIC: payment_processed, status=failed, error_code=%d", error_code)

分布式系统中的日志

请求追踪 ID

在分布式系统中，每个请求应该有唯一的追踪 ID：

import uuid

def process_request(request):
    trace_id = request.headers.get('X-Trace-ID', str(uuid.uuid4()))
    logger.info("Processing request: trace_id=%s", trace_id)

    # 在所有相关的日志中包含 trace_id
    logger.info("Query database: trace_id=%s", trace_id)

服务标识

确保每条日志都能追溯到具体的服务实例：

import socket

hostname = socket.gethostname()
logger = logging.getLogger(f"{__name__}.{hostname}")

调试技巧

条件日志

在复杂的调试场景中，使用条件日志：

if logger.isEnabledFor(logging.DEBUG):
    logger.debug("Complex object: %s", json.dumps(complex_obj, indent=2))

异常堆栈

在记录异常时，包含完整的堆栈跟踪：

try:
    risky_operation()
except Exception as e:
    logger.exception("Failed to execute risky operation")
    # 等价于 logger.error("...", exc_info=True)

最佳实践清单

在日常开发中，应该遵循以下清单：

为每个模块创建独立的 logger
使用合适的日志级别
日志消息清晰、简洁、信息完整
不记录敏感信息
生产环境使用结构化日志
配置日志轮转和归档
实施异步日志处理
为关键业务操作添加专门日志
在分布式系统中使用追踪 ID
定期审查和优化日志策略

总结

日志系统是软件开发的"黑匣子"，良好的日志实践能够大幅提升系统的可维护性和可调试性。记住，日志的最终目的是帮助理解系统的运行状态和快速定位问题，而不是简单地记录所有事情。合理的设计日志，让每一条日志都有其存在的价值。

在编写日志时，多问自己几个问题：

这条日志的目的是什么？
它是否包含了足够的上下文信息？
它的级别是否恰当？
它是否会影响系统性能？

通过持续的实践和优化，我们就能建立起一个既高效又实用的日志系统。

龙鳞