Python 日志系统设计：构建可观测的应用

Python 日志系统设计：构建可观测的应用

引言

大家好，我是一名正在从Rust转向Python的后端开发者。在构建大型应用程序时，日志系统是不可或缺的组成部分。良好的日志系统不仅能帮助我们追踪问题，还能提供应用运行时的关键洞察。作为从Rust过来的开发者，我发现Python的logging模块功能非常强大，但配置起来也相对复杂。今天，我想和大家分享一下我在设计Python日志系统方面的一些经验。

日志系统的重要性

为什么需要日志系统？

日志级别

Python的logging模块定义了以下日志级别：

基础配置

简单配置

import logging

# 基础配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)

logger = logging.getLogger(__name__)

logger.debug('这是一条调试信息')
logger.info('这是一条普通信息')
logger.warning('这是一条警告信息')
logger.error('这是一条错误信息')
logger.critical('这是一条严重错误信息')

输出到文件

import logging

logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
filename='app.log',
filemode='w' # 'w' 覆盖模式，'a' 追加模式
)

logger = logging.getLogger(__name__)
logger.info('应用启动')

高级配置

使用配置字典

import logging
from logging.config import dictConfig

logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
},
'detailed': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(module)s:%(lineno)d – %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'standard',
'level': 'INFO'
},
'file': {
'class': 'logging.FileHandler',
'filename': 'app.log',
'formatter': 'detailed',
'level': 'DEBUG'
},
'error_file': {
'class': 'logging.FileHandler',
'filename': 'error.log',
'formatter': 'detailed',
'level': 'ERROR'
}
},
'loggers': {
'': { # root logger
'handlers': ['console', 'file', 'error_file'],
'level': 'DEBUG',
'propagate': True
}
}
}

dictConfig(logging_config)
logger = logging.getLogger(__name__)
logger.debug('调试信息')
logger.info('普通信息')
logger.error('错误信息')

自定义日志格式

import logging

# 创建logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 创建formatter
formatter = logging.Formatter(
'%(asctime)s – %(name)s – %(levelname)s – %(process)d – %(thread)d – %(message)s'
)

# 创建handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)

file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)

# 添加handler到logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)

logger.info('应用启动')

实际应用场景

场景1：多模块日志管理

在大型项目中，每个模块应该有自己的logger：

# module_a.py
import logging

logger = logging.getLogger(__name__)

def do_something():
logger.debug('正在执行do_something')
try:
# 业务逻辑
logger.info('操作成功')
except Exception as e:
logger.error(f'操作失败: {e}', exc_info=True)

# main.py
import logging
from logging.config import dictConfig
import module_a

logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'standard',
'level': 'INFO'
}
},
'loggers': {
'module_a': {
'handlers': ['console'],
'level': 'DEBUG',
'propagate': False
}
}
}

dictConfig(logging_config)

if __name__ == '__main__':
logger = logging.getLogger(__name__)
logger.info('应用启动')
module_a.do_something()

场景2：日志轮转

对于长期运行的应用，日志文件会不断增大，需要进行轮转：

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 按文件大小轮转
rotating_handler = RotatingFileHandler(
'app.log',
maxBytes=1024 * 1024 * 5, # 5MB
backupCount=5 # 保留5个备份
)

# 按时间轮转
timed_handler = TimedRotatingFileHandler(
'app.log',
when='midnight', # 每天午夜轮转
interval=1,
backupCount=7 # 保留7天的日志
)

formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
rotating_handler.setFormatter(formatter)
timed_handler.setFormatter(formatter)

logger.addHandler(rotating_handler)
logger.addHandler(timed_handler)

场景3：结构化日志

在现代应用中，结构化日志（JSON格式）越来越流行：

import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
'timestamp': datetime.now().isoformat(),
'logger': record.name,
'level': record.levelname,
'message': record.getMessage(),
'module': record.module,
'line': record.lineno
}

if record.exc_info:
log_record['exception'] = self.formatException(record.exc_info)

return json.dumps(log_record)

logger = logging.getLogger('structured_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

logger.info('用户登录', extra={'user_id': 123, 'ip': '192.168.1.1'})
logger.error('数据库连接失败', extra={'db_host': 'localhost', 'db_port': 5432})

场景4：日志过滤

根据条件过滤日志：

import logging

class ErrorFilter(logging.Filter):
def filter(self, record):
# 只允许ERROR级别及以上的日志通过
return record.levelno >= logging.ERROR

logger = logging.getLogger('filtered_logger')
logger.setLevel(logging.DEBUG)

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.addFilter(ErrorFilter())

formatter = logging.Formatter('%(asctime)s – %(levelname)s – %(message)s')
console_handler.setFormatter(formatter)

logger.addHandler(console_handler)

logger.debug('调试信息') # 不会输出
logger.info('普通信息') # 不会输出
logger.error('错误信息') # 会输出

最佳实践

1. 使用结构化日志

结构化日志便于日志分析和查询：

import logging

logger = logging.getLogger(__name__)

# 使用extra参数添加额外信息
logger.info('用户登录成功', extra={
'user_id': 123,
'username': 'john_doe',
'ip_address': '192.168.1.100',
'user_agent': 'Mozilla/5.0'
})

2. 记录异常信息

try:
# 可能出错的代码
result = risky_operation()
except Exception as e:
# 记录完整的异常堆栈
logger.error(f'操作失败: {e}', exc_info=True)
# 或者使用logger.exception自动记录堆栈
logger.exception('操作失败')

3. 使用日志包装器

import logging
from functools import wraps

def log_function_call(func):
@wraps(func)
def wrapper(*args, **kwargs):
logger = logging.getLogger(func.__module__)
logger.debug(f'调用 {func.__name__}, 参数: args={args}, kwargs={kwargs}')
try:
result = func(*args, **kwargs)
logger.debug(f'{func.__name__} 执行成功, 返回: {result}')
return result
except Exception as e:
logger.error(f'{func.__name__} 执行失败: {e}', exc_info=True)
raise
return wrapper

@log_function_call
def process_data(data):
# 处理数据
return data

4. 配置分离

将日志配置放在单独的配置文件中：

# logging_config.py
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'standard',
'level': 'INFO'
}
},
'loggers': {
'': {
'handlers': ['console'],
'level': 'DEBUG',
'propagate': True
}
}
}

# main.py
from logging.config import dictConfig
from logging_config import LOGGING_CONFIG

dictConfig(LOGGING_CONFIG)

与Rust日志系统的对比

实战项目：完整的日志系统

import logging
from logging.config import dictConfig
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
'timestamp': datetime.now().isoformat(),
'logger': record.name,
'level': record.levelname,
'message': record.getMessage(),
'module': record.module,
'line': record.lineno,
'process': record.process,
'thread': record.thread
}

if record.exc_info:
log_record['exception'] = self.formatException(record.exc_info)

if hasattr(record, 'extra'):
log_record.update(record.extra)

return json.dumps(log_record)

def setup_logging():
logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'json': {
'()': JsonFormatter
},
'standard': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'standard',
'level': 'INFO'
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'app.log',
'formatter': 'json',
'level': 'DEBUG',
'maxBytes': 1024 * 1024 * 5,
'backupCount': 5
},
'error_file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'error.log',
'formatter': 'json',
'level': 'ERROR',
'maxBytes': 1024 * 1024 * 5,
'backupCount': 5
}
},
'loggers': {
'app': {
'handlers': ['console', 'file', 'error_file'],
'level': 'DEBUG',
'propagate': False
},
'app.database': {
'handlers': ['file'],
'level': 'DEBUG',
'propagate': False
},
'app.api': {
'handlers': ['console', 'file'],
'level': 'INFO',
'propagate': False
}
}
}

dictConfig(logging_config)

if __name__ == '__main__':
setup_logging()

logger = logging.getLogger('app')
db_logger = logging.getLogger('app.database')
api_logger = logging.getLogger('app.api')

logger.info('应用启动')
db_logger.debug('连接数据库')
api_logger.info('处理API请求')

总结

构建一个良好的日志系统是构建可观测应用的关键。通过合理配置日志级别、输出格式和存储策略，我们可以实现：

作为从Rust转向Python的开发者，我发现Python的日志系统虽然配置复杂，但灵活性很高。通过合理使用，可以构建出满足各种需求的日志系统。

延伸阅读：

• Python官方文档 – logging模块
• structlog – 结构化日志库
• 日志最佳实践指南