1. Python日志记录Logging最佳实践日志记录是任何应用程序不可或缺的一部分它就像应用程序的黑匣子记录了系统运行时的关键信息。作为一名有多年Python开发经验的工程师我见过太多因为日志记录不当而导致的调试噩梦。今天我将分享Python标准库logging模块的最佳实践这些经验来自于我参与过的多个大型项目包括处理日均千万级请求的Web服务。日志记录不仅仅是简单的print语句替代品它是一个完整的系统需要考虑日志级别、格式、输出目标、性能影响等多个方面。合理的日志记录可以帮助你快速定位生产环境中的问题而不当的日志配置则可能导致磁盘爆满或关键信息丢失。2. 日志系统核心概念解析2.1 日志级别及其应用场景Python的logging模块定义了6个日志级别每个级别都有特定的使用场景DEBUG(10)最详细的日志信息通常只在开发阶段使用。比如记录函数的输入参数、中间计算结果等。INFO(20)确认程序按预期运行的信息性消息。比如用户登录成功、订单已创建等业务流程关键点。WARNING(30)表明可能有问题发生但程序还能继续运行。比如磁盘空间不足80%、使用默认配置等。ERROR(40)由于更严重的问题软件已不能执行某些功能。比如数据库连接失败、支付接口调用异常等。CRITICAL(50)严重的错误可能导致程序无法继续运行。比如内存耗尽、关键配置文件丢失等。NOTSET(0)所有消息都会被记录通常只在根logger上设置。提示在生产环境中通常将日志级别设置为INFO或WARNING在开发环境可以设置为DEBUG。避免在生产环境使用DEBUG级别因为它会产生大量日志并影响性能。2.2 Logger、Handler、Filter和FormatterPython的日志系统由四个主要组件组成Logger应用程序直接使用的接口负责产生日志记录。可以设置日志级别和传播规则。Handler决定日志记录的输出位置如控制台、文件、网络等。每个handler可以有自己的日志级别和格式。Filter提供了更细粒度的控制决定哪些日志记录会被输出。Formatter指定日志记录的最终输出格式。这种组件化的设计使得日志系统非常灵活你可以为不同的模块设置不同的日志级别或者将错误日志发送到邮件而将调试日志写入文件。3. 基础配置与使用3.1 快速开始基础日志配置最简单的日志配置方式是使用basicConfig()方法这适合小型应用或脚本import logging # 基本配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, filenameapp.log, filemodea ) # 使用日志 logging.info(程序启动) try: 1 / 0 except ZeroDivisionError: logging.error(除零错误发生, exc_infoTrue)这段代码配置了日志级别为INFO格式包含时间、logger名称、级别和消息输出到app.log文件。exc_infoTrue会在日志中记录完整的异常堆栈。3.2 高级配置字典配置与文件配置对于复杂的应用程序推荐使用字典或配置文件来配置日志系统。以下是使用字典配置的例子import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { standard: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s, datefmt: %Y-%m-%d %H:%M:%S }, }, handlers: { console: { class: logging.StreamHandler, level: DEBUG, formatter: standard, stream: ext://sys.stdout }, file: { class: logging.handlers.RotatingFileHandler, level: INFO, formatter: standard, filename: app.log, maxBytes: 10485760, # 10MB backupCount: 5, encoding: utf8 }, }, loggers: { : { # root logger handlers: [console, file], level: DEBUG, propagate: False }, my_module: { handlers: [file], level: INFO, propagate: False } } } logging.config.dictConfig(LOGGING_CONFIG) logger logging.getLogger(__name__)这种配置方式提供了更精细的控制包括不同的formatter用于不同的handlerRotatingFileHandler实现日志轮转防止单个文件过大为不同模块设置不同的日志级别和handler控制日志传播行为4. 生产环境最佳实践4.1 结构化日志记录在现代应用程序中特别是使用ELKElasticsearch, Logstash, Kibana等日志分析系统时结构化日志变得非常重要。Python的logging模块可以通过自定义Formatter实现JSON格式的日志import json import logging from datetime import datetime class JsonFormatter(logging.Formatter): def format(self, record): log_record { timestamp: datetime.utcnow().isoformat(), level: record.levelname, message: record.getMessage(), logger: record.name, module: record.module, function: record.funcName, line: record.lineno, } if record.exc_info: log_record[exception] self.formatException(record.exc_info) return json.dumps(log_record) # 配置使用JsonFormatter handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger logging.getLogger() logger.addHandler(handler) logger.setLevel(logging.INFO) # 使用示例 logger.info(用户登录成功, extra{user_id: 12345, ip: 192.168.1.1})结构化日志使得日志分析工具能够更容易地解析和索引日志内容实现更强大的查询和可视化功能。4.2 日志轮转与归档策略在生产环境中必须考虑日志文件的轮转和归档以防止磁盘空间被占满。Python提供了几种Handler来实现这一功能RotatingFileHandler基于文件大小的轮转from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes10*1024*1024, # 10MB backupCount5, encodingutf-8 )TimedRotatingFileHandler基于时间的轮转from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( app.log, whenmidnight, interval1, backupCount7, encodingutf-8 )注意在Windows系统上由于文件锁定机制可能需要先关闭现有的handler才能成功轮转文件。4.3 性能优化技巧日志记录虽然重要但不合理的实现可能成为性能瓶颈。以下是一些优化建议避免在热路径上进行字符串格式化# 不推荐 - 即使日志级别高于DEBUG也会执行字符串格式化 logger.debug(User %s logged in from %s, user, ip) # 推荐 - 先检查日志级别 if logger.isEnabledFor(logging.DEBUG): logger.debug(User %s logged in from %s, user, ip)使用适当的日志级别生产环境避免使用DEBUG级别它会生成大量日志并影响性能。异步日志记录对于高吞吐量应用考虑使用异步handler如concurrent-log-handler包提供的ConcurrentRotatingFileHandler。合理设置传播避免不必要的日志传播特别是当使用多个handler时。5. 高级应用场景5.1 多模块应用的日志管理在大型应用中通常需要为不同模块设置不同的日志级别。Python的logging模块支持层次化的logger命名# 主模块 import logging logging.basicConfig(levellogging.WARNING) logger logging.getLogger(__name__) # 子模块 module_logger logging.getLogger(main.submodule) module_logger.setLevel(logging.DEBUG) # 只为这个子模块设置DEBUG级别logger名称使用点号分隔形成层次结构例如app (主应用)app.models (模型层)app.views (视图层)父logger的设置可以被子logger继承除非子logger明确设置了不同的值。5.2 上下文感知的日志记录在Web应用或异步任务中通常需要在日志中包含请求ID、用户ID等上下文信息。有几种实现方式使用Filter添加上下文class ContextFilter(logging.Filter): def filter(self, record): record.request_id get_current_request_id() # 从线程局部存储等获取 return True logger.addFilter(ContextFilter()) formatter logging.Formatter(%(asctime)s [%(request_id)s] %(message)s)使用LoggerAdapterlogger logging.getLogger(__name__) adapter logging.LoggerAdapter(logger, {request_id: 12345}) adapter.info(Processing request)使用structlog等第三方库import structlog structlog.configure( processors[ structlog.processors.JSONRenderer() ] ) log structlog.get_logger() log.info(user_login, user_id123, ip192.168.1.1)5.3 分布式系统中的日志收集在微服务架构中日志分散在各个服务中需要集中收集和分析。常见的解决方案包括ELK Stack(Elasticsearch Logstash Kibana)FluentdSplunkAWS CloudWatch LogsPython应用可以通过配置适当的handler将日志发送到这些系统。例如使用python-logstash将日志发送到Logstashfrom logstash_async.handler import AsynchronousLogstashHandler logstash_handler AsynchronousLogstashHandler( hostlocalhost, port5959, database_pathlogstash.db ) logger.addHandler(logstash_handler)6. 常见问题与解决方案6.1 日志不显示或输出不正确问题现象配置了日志但看不到输出或者格式不符合预期。排查步骤检查logger的有效级别logger.getEffectiveLevel()确认handler是否添加到loggerlogger.handlers检查handler的级别是否高于logger的级别确认日志没有因为传播而被父logger过滤典型解决方案# 确保handler已添加 if not logger.handlers: logger.addHandler(logging.StreamHandler()) # 确保日志级别设置正确 logger.setLevel(logging.DEBUG) for handler in logger.handlers: handler.setLevel(logging.DEBUG)6.2 日志文件权限问题问题现象应用无法写入日志文件特别是在使用不同用户运行的应用中。解决方案确保运行用户对日志目录有写权限使用os.umask设置适当的文件创建权限考虑使用logging.handlers.WatchedFileHandler它会在文件被移动或删除后自动重新打开import os os.umask(0o002) # 设置文件创建时的权限掩码 from logging.handlers import WatchedFileHandler handler WatchedFileHandler(app.log)6.3 日志性能问题问题现象应用性能下降特别是在高负载时。优化建议使用异步日志handler减少不必要的日志记录特别是DEBUG级别避免在日志消息中进行复杂的计算或IO操作考虑使用更快的JSON序列化库如orjson代替标准库的json# 使用orjson加速JSON日志格式化 import orjson class FastJsonFormatter(logging.Formatter): def format(self, record): log_record {...} # 同上 return orjson.dumps(log_record).decode(utf-8)7. 日志安全注意事项日志中可能包含敏感信息如密码、API密钥、个人信息等必须小心处理避免记录敏感数据# 不推荐 logger.info(f用户登录密码: {password}) # 推荐 logger.info(用户登录, extra{user: username})实现敏感数据过滤class SensitiveDataFilter(logging.Filter): def filter(self, record): if hasattr(record, password): record.password ***REDACTED*** return True logger.addFilter(SensitiveDataFilter())控制日志访问权限确保日志文件只能被授权用户访问在生产环境中避免将日志写入web可访问的目录考虑使用加密的日志存储合规性考虑根据GDPR等法规要求可能需要定期清理包含个人数据的日志确保日志保留策略符合组织政策和法律法规8. 实战完整的日志配置示例下面是一个综合了上述最佳实践的完整日志配置示例适合大多数Python Web应用import logging import logging.config from logging.handlers import RotatingFileHandler import os import json from datetime import datetime def setup_logging(): LOG_DIR logs os.makedirs(LOG_DIR, exist_okTrue) class JsonFormatter(logging.Formatter): def format(self, record): log_record { timestamp: datetime.utcnow().isoformat(), level: record.levelname, logger: record.name, message: record.getMessage(), module: record.module, function: record.funcName, line: record.lineno, } if record.exc_info: log_record[exception] self.formatException(record.exc_info) return json.dumps(log_record, ensure_asciiFalse) # 基础配置 logging_config { version: 1, disable_existing_loggers: False, formatters: { json: { (): JsonFormatter, }, simple: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s, datefmt: %Y-%m-%d %H:%M:%S }, }, handlers: { console: { class: logging.StreamHandler, level: INFO, formatter: simple, stream: ext://sys.stdout }, file: { (): RotatingFileHandler, level: INFO, formatter: json, filename: os.path.join(LOG_DIR, app.log), maxBytes: 10 * 1024 * 1024, # 10MB backupCount: 5, encoding: utf8 }, error_file: { (): RotatingFileHandler, level: ERROR, formatter: json, filename: os.path.join(LOG_DIR, error.log), maxBytes: 10 * 1024 * 1024, backupCount: 5, encoding: utf8 }, }, loggers: { : { # root logger handlers: [console, file, error_file], level: INFO, }, app.db: { handlers: [file], level: DEBUG, propagate: False, }, app.api: { handlers: [console, file], level: INFO, propagate: False, }, }, } logging.config.dictConfig(logging_config) # 添加全局过滤器 class SensitiveDataFilter(logging.Filter): def filter(self, record): if hasattr(record, password): record.password ***REDACTED*** return True for handler in logging.root.handlers: handler.addFilter(SensitiveDataFilter()) # 初始化日志配置 setup_logging() # 获取logger实例 logger logging.getLogger(__name__) db_logger logging.getLogger(app.db) api_logger logging.getLogger(app.api)这个配置实现了多级别日志记录INFO到文件ERROR单独文件JSON格式的结构化日志日志轮转防止磁盘空间耗尽敏感数据过滤不同模块的不同日志级别配置控制台和文件双重输出9. 日志监控与告警仅仅记录日志是不够的还需要监控日志并设置适当的告警。常见的做法包括关键错误告警监控ERROR及以上级别的日志通过邮件、Slack等通知开发团队class AlertHandler(logging.Handler): def emit(self, record): if record.levelno logging.ERROR: send_alert(record.getMessage()) # 实现发送告警的逻辑 logger.addHandler(AlertHandler())异常模式检测使用ELK或Splunk等工具检测异常日志模式如突然增加的ERROR日志心跳监控确保应用定期产生心跳日志没有心跳则触发告警性能指标提取从日志中提取请求处理时间等指标监控性能变化日志采样在高负载情况下可以对DEBUG日志进行采样而不是记录全部以平衡详细度和性能10. 日志与应用程序健康检查日志可以作为应用程序健康检查的重要组成部分。一些实践建议启动日志应用启动时记录关键配置和环境信息logger.info(应用启动, extra{ config: sanitized_config, environment: os.environ.get(ENV, development) })健康检查端点在Web应用中添加/health端点检查依赖服务状态并记录app.route(/health) def health_check(): try: db_status check_database() cache_status check_cache() logger.info(健康检查, extra{ db_status: db_status, cache_status: cache_status }) return jsonify({status: healthy}), 200 except Exception as e: logger.error(健康检查失败, exc_infoTrue) return jsonify({status: unhealthy}), 500定期状态日志记录内存使用、线程数等系统指标import psutil import threading def log_system_stats(): logger.info(系统状态, extra{ memory: psutil.virtual_memory().percent, cpu: psutil.cpu_percent(), threads: threading.active_count() }) threading.Timer(300, log_system_stats).start() # 每5分钟记录一次 # 启动状态监控 log_system_stats()优雅关闭捕获终止信号并记录关闭原因import signal def handle_shutdown(signum, frame): logger.info(接收到关闭信号 %s, signum) # 执行清理操作 sys.exit(0) signal.signal(signal.SIGTERM, handle_shutdown) signal.signal(signal.SIGINT, handle_shutdown)11. 日志测试与验证确保日志系统正常工作同样需要测试。一些测试策略单元测试日志输出import unittest from io import StringIO class TestLogging(unittest.TestCase): def setUp(self): self.stream StringIO() handler logging.StreamHandler(self.stream) logger logging.getLogger(test) logger.setLevel(logging.INFO) logger.handlers [handler] self.logger logger def test_log_output(self): self.logger.info(测试消息) self.assertIn(测试消息, self.stream.getvalue())集成测试日志配置验证生产配置是否正确加载def test_log_config(): setup_logging() logger logging.getLogger(__name__) assert logger.level logging.INFO assert any(isinstance(h, RotatingFileHandler) for h in logger.handlers)日志内容断言在功能测试中验证特定日志是否产生def test_login_logging(): with self.assertLogs(app.auth, levelINFO) as cm: login_user(test, password) self.assertIn(用户test登录成功, cm.output[0])性能测试评估日志系统对应用性能的影响import timeit def test_log_performance(): def log_messages(): for i in range(1000): logger.info(测试消息 %d, i) time timeit.timeit(log_messages, number10) assert time 1.0 # 10000条日志应在1秒内完成12. 日志与审计跟踪在某些领域如金融、医疗日志不仅用于调试还用于合规审计。这时需要考虑不可篡改性确保日志一旦写入就不能被修改可以考虑将日志写入只读介质使用区块链技术存储日志哈希定期将日志归档到安全存储完整时间戳使用高精度、同步的时间源from datetime import datetime, timezone logger.info(关键操作, extra{ timestamp: datetime.now(timezone.utc).isoformat(), user: current_user.id, action: delete_record })操作关联使用唯一ID关联相关操作import uuid request_id str(uuid.uuid4()) def process_request(request): logger.info(开始处理请求, extra{request_id: request_id}) # ...处理逻辑... logger.info(请求处理完成, extra{request_id: request_id})长期保留根据法规要求保留日志足够长时间实现自动归档和过期策略考虑冷热存储分层近期日志在快速存储旧日志在廉价存储访问控制限制谁可以访问审计日志日志文件设置严格的文件权限查询接口实现认证和授权记录所有对审计日志的访问13. 日志与错误追踪系统集成现代错误追踪系统如Sentry、Rollbar可以与日志系统集成提供更强大的错误监控能力Sentry集成示例import sentry_sdk from sentry_sdk.integrations.logging import LoggingIntegration # 所有日志级别ERROR的事件都会发送到Sentry sentry_logging LoggingIntegration( levellogging.INFO, # 捕获INFO及以上作为breadcrumbs event_levellogging.ERROR # 发送ERROR及以上作为事件 ) sentry_sdk.init( dsnyour-dsn-here, integrations[sentry_logging], traces_sample_rate1.0 ) # 普通日志使用方式不变 logger.error(发生了严重错误, extra{data: {user_id: 123}})错误上下文增强try: risky_operation() except Exception as e: logger.error(操作失败, exc_infoTrue, extra{ context: { user: current_user.id, data: sanitized_data } }) raise性能监控集成# 在Sentry中跟踪事务 with sentry_sdk.start_transaction(optask, nameprocess_data): logger.info(开始处理数据) # ...处理逻辑... logger.info(数据处理完成)自定义指纹控制错误分组方式try: process_payment() except PaymentError as e: logger.error(支付处理失败, exc_infoTrue, extra{ fingerprint: [payment, str(e.code)] })14. 日志与指标监控系统集成日志可以与Prometheus、StatsD等指标监控系统结合提供更全面的可观测性从日志提取指标from prometheus_client import Counter LOG_ERRORS Counter(app_log_errors, Number of ERROR logs, [logger]) class MetricsFilter(logging.Filter): def filter(self, record): if record.levelno logging.ERROR: LOG_ERRORS.labels(loggerrecord.name).inc() return True logger.addFilter(MetricsFilter())记录性能指标import time from prometheus_client import Histogram REQUEST_TIME Histogram(request_processing_seconds, Time spent processing request) def process_request(request): start_time time.time() logger.info(开始处理请求) try: # ...处理逻辑... logger.info(请求处理成功) finally: duration time.time() - start_time REQUEST_TIME.observe(duration) logger.info(请求处理完成, extra{duration: duration})结构化日志与指标关联# 日志 logger.info(API响应, extra{ endpoint: /users, status_code: 200, response_time: 0.45 }) # 对应的Prometheus指标 API_RESPONSES Counter(api_responses, API responses, [endpoint, status_code]) API_DURATION Histogram(api_duration, API response time, [endpoint]) class ApiMetricsFilter(logging.Filter): def filter(self, record): if hasattr(record, endpoint): API_RESPONSES.labels( endpointrecord.endpoint, status_codegetattr(record, status_code, 0) ).inc() if hasattr(record, response_time): API_DURATION.labels( endpointrecord.endpoint ).observe(record.response_time) return True15. 日志与分布式追踪系统集成在微服务架构中分布式追踪系统如Jaeger、Zipkin可以帮助跟踪请求在多个服务间的流转传播追踪上下文from opentelemetry import trace from opentelemetry.propagate import inject, extract # 在发送请求前注入追踪上下文 headers {} inject(headers) logger.info(发送请求, extra{headers: headers}) # 在接收请求时提取追踪上下文 context extract(headers) tracer trace.get_tracer(__name__) with tracer.start_as_current_span(process, contextcontext): logger.info(处理请求)在日志中包含追踪IDclass TracingFilter(logging.Filter): def filter(self, record): span trace.get_current_span() if span.is_recording(): record.trace_id format_trace_id(span.get_span_context().trace_id) record.span_id format_span_id(span.get_span_context().span_id) return True logger.addFilter(TracingFilter()) formatter logging.Formatter(%(trace_id)s %(span_id)s %(message)s)关联日志与追踪with tracer.start_as_current_span(database.query): logger.info(执行数据库查询, extra{ query: sanitized_query, parameters: parameters }) # ...执行查询...采样决策传播def should_log(record): span trace.get_current_span() if span.is_recording(): # 如果span被采样则记录详细日志 return True # 否则只记录错误 return record.levelno logging.ERROR class SamplingFilter(logging.Filter): def filter(self, record): return should_log(record)16. 日志与机器学习日志数据可以用于训练机器学习模型用于异常检测、预测性维护等场景日志模式挖掘from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import KMeans # 从日志中提取文本特征 logs [log.message for log in query_logs()] vectorizer TfidfVectorizer(max_features1000) X vectorizer.fit_transform(logs) # 聚类分析日志模式 kmeans KMeans(n_clusters10) clusters kmeans.fit_predict(X) # 分析每个簇的特征 for i in range(10): cluster_logs [log for log, c in zip(logs, clusters) if c i] logger.info(f日志簇{i}示例, extra{examples: cluster_logs[:3]})异常检测from sklearn.ensemble import IsolationForest # 训练异常检测模型 model IsolationForest(contamination0.01) model.fit(X) # 预测新日志是否异常 new_logs [log.message for log in query_recent_logs()] X_new vectorizer.transform(new_logs) anomalies model.predict(X_new) -1 for log, is_anomaly in zip(new_logs, anomalies): if is_anomaly: logger.warning(异常日志检测, extra{log: log})预测性维护import pandas as pd from prophet import Prophet # 将错误日志按时间聚合 error_logs query_error_logs() df pd.DataFrame({ ds: [log.timestamp for log in error_logs], y: 1 }).groupby(ds).sum().reset_index() # 预测未来错误趋势 model Prophet() model.fit(df) future model.make_future_dataframe(periods30) forecast model.predict(future) # 如果预测错误率上升触发告警 if forecast.tail(30)[yhat].mean() df[y].mean() * 1.5: logger.error(预测错误率将显著上升)日志分类from sklearn.pipeline import Pipeline from sklearn.linear_model import LogisticRegression # 准备已标记的训练数据 X_train [log.message for log in training_logs] y_train [log.label for log in training_logs] # 训练分类器 text_clf Pipeline([ (tfidf, TfidfVectorizer()), (clf, LogisticRegression()) ]) text_clf.fit(X_train, y_train) # 分类新日志 new_logs [log.message for log in unlabeled_logs] predictions text_clf.predict(new_logs) for log, label in zip(new_logs, predictions): logger.info(自动分类日志, extra{ log: log, predicted_label: label })17. 日志与安全分析日志是安全分析的重要数据源可以用于检测入侵、异常行为等暴力破解检测from collections import defaultdict failed_logins defaultdict(int) class SecurityFilter(logging.Filter): def filter(self, record): if (record.getMessage().startswith(登录失败) and hasattr(record, ip)): failed_logins[record.ip] 1 if failed_logins[record.ip] 5: logger.warning(可能的暴力破解尝试, extra{ ip: record.ip, count: failed_logins[record.ip] }) return True logger.addFilter(SecurityFilter())敏感操作监控SENSITIVE_ACTIONS [delete, grant, revoke, admin] class SensitiveActionFilter(logging.Filter): def filter(self, record): if hasattr(record, action): if any(action in record.action for action in SENSITIVE_ACTIONS): logger.info(敏感操作记录, extra{ user: getattr(record, user, unknown), action: record.action, target: getattr(record, target, unknown) }) return True异常时间活动检测import datetime class UnusualTimeFilter(logging.Filter): def filter(self, record): if hasattr(record, timestamp): hour datetime.datetime.fromisoformat(record.timestamp).hour if hour 8 or hour 20: # 非工作时间 logger.warning(非工作时间活动, extra{ user: getattr(record, user, unknown), time: record.timestamp, action: getattr(record, action, unknown) }) return True数据泄露检测import re PATTERNS [ r\b\d{4}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b, # 信用卡 r\b\d{3}[ -]?\d{2}[ -]?\d{4}\b, # SSN # 其他敏感数据模式 ] class DataLeakFilter(logging.Filter): def filter(self, record): message record.getMessage() for pattern in PATTERNS: if re.search(pattern, message): logger.error(可能的敏感数据泄露, extra{ pattern: pattern, log: message }) # 可以进一步redact敏感数据 record.msg re.sub(pattern, ***REDACTED***, message) return True18. 日志与合规性在许多行业日志记录是合规性要求的一部分。需要考虑数据保留策略from datetime import datetime, timedelta def cleanup_old_logs(): cutoff datetime.now() - timedelta(days365) # 保留1年 old_logs query_logs(beforecutoff) for log in old_logs: if not is_required_for_compliance(log): delete_log(log)访问审计class AccessAuditFilter(logging.Filter): def filter(self, record): if hasattr(record, access_type): logger.info(访问审计, extra{ user: record.user, resource: record.resource, action: record.action, timestamp: datetime.now(timezone.utc).isoformat() }) return True不可否认性import hashlib def create_log_hash(record): log_data json.dumps({ timestamp: record.timestamp, message: record.message, user: record.user }, sort_keysTrue).encode(utf-8) return hashlib.sha256(log_data).hexdigest() class NonRepudiationFilter(logging.Filter): def filter(self, record): record.hash create_log_hash(record) return True合规性报告def generate_compliance_report(): report { period: { Python日志记录最佳实践与高级应用
1. Python日志记录Logging最佳实践日志记录是任何应用程序不可或缺的一部分它就像应用程序的黑匣子记录了系统运行时的关键信息。作为一名有多年Python开发经验的工程师我见过太多因为日志记录不当而导致的调试噩梦。今天我将分享Python标准库logging模块的最佳实践这些经验来自于我参与过的多个大型项目包括处理日均千万级请求的Web服务。日志记录不仅仅是简单的print语句替代品它是一个完整的系统需要考虑日志级别、格式、输出目标、性能影响等多个方面。合理的日志记录可以帮助你快速定位生产环境中的问题而不当的日志配置则可能导致磁盘爆满或关键信息丢失。2. 日志系统核心概念解析2.1 日志级别及其应用场景Python的logging模块定义了6个日志级别每个级别都有特定的使用场景DEBUG(10)最详细的日志信息通常只在开发阶段使用。比如记录函数的输入参数、中间计算结果等。INFO(20)确认程序按预期运行的信息性消息。比如用户登录成功、订单已创建等业务流程关键点。WARNING(30)表明可能有问题发生但程序还能继续运行。比如磁盘空间不足80%、使用默认配置等。ERROR(40)由于更严重的问题软件已不能执行某些功能。比如数据库连接失败、支付接口调用异常等。CRITICAL(50)严重的错误可能导致程序无法继续运行。比如内存耗尽、关键配置文件丢失等。NOTSET(0)所有消息都会被记录通常只在根logger上设置。提示在生产环境中通常将日志级别设置为INFO或WARNING在开发环境可以设置为DEBUG。避免在生产环境使用DEBUG级别因为它会产生大量日志并影响性能。2.2 Logger、Handler、Filter和FormatterPython的日志系统由四个主要组件组成Logger应用程序直接使用的接口负责产生日志记录。可以设置日志级别和传播规则。Handler决定日志记录的输出位置如控制台、文件、网络等。每个handler可以有自己的日志级别和格式。Filter提供了更细粒度的控制决定哪些日志记录会被输出。Formatter指定日志记录的最终输出格式。这种组件化的设计使得日志系统非常灵活你可以为不同的模块设置不同的日志级别或者将错误日志发送到邮件而将调试日志写入文件。3. 基础配置与使用3.1 快速开始基础日志配置最简单的日志配置方式是使用basicConfig()方法这适合小型应用或脚本import logging # 基本配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, filenameapp.log, filemodea ) # 使用日志 logging.info(程序启动) try: 1 / 0 except ZeroDivisionError: logging.error(除零错误发生, exc_infoTrue)这段代码配置了日志级别为INFO格式包含时间、logger名称、级别和消息输出到app.log文件。exc_infoTrue会在日志中记录完整的异常堆栈。3.2 高级配置字典配置与文件配置对于复杂的应用程序推荐使用字典或配置文件来配置日志系统。以下是使用字典配置的例子import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { standard: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s, datefmt: %Y-%m-%d %H:%M:%S }, }, handlers: { console: { class: logging.StreamHandler, level: DEBUG, formatter: standard, stream: ext://sys.stdout }, file: { class: logging.handlers.RotatingFileHandler, level: INFO, formatter: standard, filename: app.log, maxBytes: 10485760, # 10MB backupCount: 5, encoding: utf8 }, }, loggers: { : { # root logger handlers: [console, file], level: DEBUG, propagate: False }, my_module: { handlers: [file], level: INFO, propagate: False } } } logging.config.dictConfig(LOGGING_CONFIG) logger logging.getLogger(__name__)这种配置方式提供了更精细的控制包括不同的formatter用于不同的handlerRotatingFileHandler实现日志轮转防止单个文件过大为不同模块设置不同的日志级别和handler控制日志传播行为4. 生产环境最佳实践4.1 结构化日志记录在现代应用程序中特别是使用ELKElasticsearch, Logstash, Kibana等日志分析系统时结构化日志变得非常重要。Python的logging模块可以通过自定义Formatter实现JSON格式的日志import json import logging from datetime import datetime class JsonFormatter(logging.Formatter): def format(self, record): log_record { timestamp: datetime.utcnow().isoformat(), level: record.levelname, message: record.getMessage(), logger: record.name, module: record.module, function: record.funcName, line: record.lineno, } if record.exc_info: log_record[exception] self.formatException(record.exc_info) return json.dumps(log_record) # 配置使用JsonFormatter handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger logging.getLogger() logger.addHandler(handler) logger.setLevel(logging.INFO) # 使用示例 logger.info(用户登录成功, extra{user_id: 12345, ip: 192.168.1.1})结构化日志使得日志分析工具能够更容易地解析和索引日志内容实现更强大的查询和可视化功能。4.2 日志轮转与归档策略在生产环境中必须考虑日志文件的轮转和归档以防止磁盘空间被占满。Python提供了几种Handler来实现这一功能RotatingFileHandler基于文件大小的轮转from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes10*1024*1024, # 10MB backupCount5, encodingutf-8 )TimedRotatingFileHandler基于时间的轮转from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( app.log, whenmidnight, interval1, backupCount7, encodingutf-8 )注意在Windows系统上由于文件锁定机制可能需要先关闭现有的handler才能成功轮转文件。4.3 性能优化技巧日志记录虽然重要但不合理的实现可能成为性能瓶颈。以下是一些优化建议避免在热路径上进行字符串格式化# 不推荐 - 即使日志级别高于DEBUG也会执行字符串格式化 logger.debug(User %s logged in from %s, user, ip) # 推荐 - 先检查日志级别 if logger.isEnabledFor(logging.DEBUG): logger.debug(User %s logged in from %s, user, ip)使用适当的日志级别生产环境避免使用DEBUG级别它会生成大量日志并影响性能。异步日志记录对于高吞吐量应用考虑使用异步handler如concurrent-log-handler包提供的ConcurrentRotatingFileHandler。合理设置传播避免不必要的日志传播特别是当使用多个handler时。5. 高级应用场景5.1 多模块应用的日志管理在大型应用中通常需要为不同模块设置不同的日志级别。Python的logging模块支持层次化的logger命名# 主模块 import logging logging.basicConfig(levellogging.WARNING) logger logging.getLogger(__name__) # 子模块 module_logger logging.getLogger(main.submodule) module_logger.setLevel(logging.DEBUG) # 只为这个子模块设置DEBUG级别logger名称使用点号分隔形成层次结构例如app (主应用)app.models (模型层)app.views (视图层)父logger的设置可以被子logger继承除非子logger明确设置了不同的值。5.2 上下文感知的日志记录在Web应用或异步任务中通常需要在日志中包含请求ID、用户ID等上下文信息。有几种实现方式使用Filter添加上下文class ContextFilter(logging.Filter): def filter(self, record): record.request_id get_current_request_id() # 从线程局部存储等获取 return True logger.addFilter(ContextFilter()) formatter logging.Formatter(%(asctime)s [%(request_id)s] %(message)s)使用LoggerAdapterlogger logging.getLogger(__name__) adapter logging.LoggerAdapter(logger, {request_id: 12345}) adapter.info(Processing request)使用structlog等第三方库import structlog structlog.configure( processors[ structlog.processors.JSONRenderer() ] ) log structlog.get_logger() log.info(user_login, user_id123, ip192.168.1.1)5.3 分布式系统中的日志收集在微服务架构中日志分散在各个服务中需要集中收集和分析。常见的解决方案包括ELK Stack(Elasticsearch Logstash Kibana)FluentdSplunkAWS CloudWatch LogsPython应用可以通过配置适当的handler将日志发送到这些系统。例如使用python-logstash将日志发送到Logstashfrom logstash_async.handler import AsynchronousLogstashHandler logstash_handler AsynchronousLogstashHandler( hostlocalhost, port5959, database_pathlogstash.db ) logger.addHandler(logstash_handler)6. 常见问题与解决方案6.1 日志不显示或输出不正确问题现象配置了日志但看不到输出或者格式不符合预期。排查步骤检查logger的有效级别logger.getEffectiveLevel()确认handler是否添加到loggerlogger.handlers检查handler的级别是否高于logger的级别确认日志没有因为传播而被父logger过滤典型解决方案# 确保handler已添加 if not logger.handlers: logger.addHandler(logging.StreamHandler()) # 确保日志级别设置正确 logger.setLevel(logging.DEBUG) for handler in logger.handlers: handler.setLevel(logging.DEBUG)6.2 日志文件权限问题问题现象应用无法写入日志文件特别是在使用不同用户运行的应用中。解决方案确保运行用户对日志目录有写权限使用os.umask设置适当的文件创建权限考虑使用logging.handlers.WatchedFileHandler它会在文件被移动或删除后自动重新打开import os os.umask(0o002) # 设置文件创建时的权限掩码 from logging.handlers import WatchedFileHandler handler WatchedFileHandler(app.log)6.3 日志性能问题问题现象应用性能下降特别是在高负载时。优化建议使用异步日志handler减少不必要的日志记录特别是DEBUG级别避免在日志消息中进行复杂的计算或IO操作考虑使用更快的JSON序列化库如orjson代替标准库的json# 使用orjson加速JSON日志格式化 import orjson class FastJsonFormatter(logging.Formatter): def format(self, record): log_record {...} # 同上 return orjson.dumps(log_record).decode(utf-8)7. 日志安全注意事项日志中可能包含敏感信息如密码、API密钥、个人信息等必须小心处理避免记录敏感数据# 不推荐 logger.info(f用户登录密码: {password}) # 推荐 logger.info(用户登录, extra{user: username})实现敏感数据过滤class SensitiveDataFilter(logging.Filter): def filter(self, record): if hasattr(record, password): record.password ***REDACTED*** return True logger.addFilter(SensitiveDataFilter())控制日志访问权限确保日志文件只能被授权用户访问在生产环境中避免将日志写入web可访问的目录考虑使用加密的日志存储合规性考虑根据GDPR等法规要求可能需要定期清理包含个人数据的日志确保日志保留策略符合组织政策和法律法规8. 实战完整的日志配置示例下面是一个综合了上述最佳实践的完整日志配置示例适合大多数Python Web应用import logging import logging.config from logging.handlers import RotatingFileHandler import os import json from datetime import datetime def setup_logging(): LOG_DIR logs os.makedirs(LOG_DIR, exist_okTrue) class JsonFormatter(logging.Formatter): def format(self, record): log_record { timestamp: datetime.utcnow().isoformat(), level: record.levelname, logger: record.name, message: record.getMessage(), module: record.module, function: record.funcName, line: record.lineno, } if record.exc_info: log_record[exception] self.formatException(record.exc_info) return json.dumps(log_record, ensure_asciiFalse) # 基础配置 logging_config { version: 1, disable_existing_loggers: False, formatters: { json: { (): JsonFormatter, }, simple: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s, datefmt: %Y-%m-%d %H:%M:%S }, }, handlers: { console: { class: logging.StreamHandler, level: INFO, formatter: simple, stream: ext://sys.stdout }, file: { (): RotatingFileHandler, level: INFO, formatter: json, filename: os.path.join(LOG_DIR, app.log), maxBytes: 10 * 1024 * 1024, # 10MB backupCount: 5, encoding: utf8 }, error_file: { (): RotatingFileHandler, level: ERROR, formatter: json, filename: os.path.join(LOG_DIR, error.log), maxBytes: 10 * 1024 * 1024, backupCount: 5, encoding: utf8 }, }, loggers: { : { # root logger handlers: [console, file, error_file], level: INFO, }, app.db: { handlers: [file], level: DEBUG, propagate: False, }, app.api: { handlers: [console, file], level: INFO, propagate: False, }, }, } logging.config.dictConfig(logging_config) # 添加全局过滤器 class SensitiveDataFilter(logging.Filter): def filter(self, record): if hasattr(record, password): record.password ***REDACTED*** return True for handler in logging.root.handlers: handler.addFilter(SensitiveDataFilter()) # 初始化日志配置 setup_logging() # 获取logger实例 logger logging.getLogger(__name__) db_logger logging.getLogger(app.db) api_logger logging.getLogger(app.api)这个配置实现了多级别日志记录INFO到文件ERROR单独文件JSON格式的结构化日志日志轮转防止磁盘空间耗尽敏感数据过滤不同模块的不同日志级别配置控制台和文件双重输出9. 日志监控与告警仅仅记录日志是不够的还需要监控日志并设置适当的告警。常见的做法包括关键错误告警监控ERROR及以上级别的日志通过邮件、Slack等通知开发团队class AlertHandler(logging.Handler): def emit(self, record): if record.levelno logging.ERROR: send_alert(record.getMessage()) # 实现发送告警的逻辑 logger.addHandler(AlertHandler())异常模式检测使用ELK或Splunk等工具检测异常日志模式如突然增加的ERROR日志心跳监控确保应用定期产生心跳日志没有心跳则触发告警性能指标提取从日志中提取请求处理时间等指标监控性能变化日志采样在高负载情况下可以对DEBUG日志进行采样而不是记录全部以平衡详细度和性能10. 日志与应用程序健康检查日志可以作为应用程序健康检查的重要组成部分。一些实践建议启动日志应用启动时记录关键配置和环境信息logger.info(应用启动, extra{ config: sanitized_config, environment: os.environ.get(ENV, development) })健康检查端点在Web应用中添加/health端点检查依赖服务状态并记录app.route(/health) def health_check(): try: db_status check_database() cache_status check_cache() logger.info(健康检查, extra{ db_status: db_status, cache_status: cache_status }) return jsonify({status: healthy}), 200 except Exception as e: logger.error(健康检查失败, exc_infoTrue) return jsonify({status: unhealthy}), 500定期状态日志记录内存使用、线程数等系统指标import psutil import threading def log_system_stats(): logger.info(系统状态, extra{ memory: psutil.virtual_memory().percent, cpu: psutil.cpu_percent(), threads: threading.active_count() }) threading.Timer(300, log_system_stats).start() # 每5分钟记录一次 # 启动状态监控 log_system_stats()优雅关闭捕获终止信号并记录关闭原因import signal def handle_shutdown(signum, frame): logger.info(接收到关闭信号 %s, signum) # 执行清理操作 sys.exit(0) signal.signal(signal.SIGTERM, handle_shutdown) signal.signal(signal.SIGINT, handle_shutdown)11. 日志测试与验证确保日志系统正常工作同样需要测试。一些测试策略单元测试日志输出import unittest from io import StringIO class TestLogging(unittest.TestCase): def setUp(self): self.stream StringIO() handler logging.StreamHandler(self.stream) logger logging.getLogger(test) logger.setLevel(logging.INFO) logger.handlers [handler] self.logger logger def test_log_output(self): self.logger.info(测试消息) self.assertIn(测试消息, self.stream.getvalue())集成测试日志配置验证生产配置是否正确加载def test_log_config(): setup_logging() logger logging.getLogger(__name__) assert logger.level logging.INFO assert any(isinstance(h, RotatingFileHandler) for h in logger.handlers)日志内容断言在功能测试中验证特定日志是否产生def test_login_logging(): with self.assertLogs(app.auth, levelINFO) as cm: login_user(test, password) self.assertIn(用户test登录成功, cm.output[0])性能测试评估日志系统对应用性能的影响import timeit def test_log_performance(): def log_messages(): for i in range(1000): logger.info(测试消息 %d, i) time timeit.timeit(log_messages, number10) assert time 1.0 # 10000条日志应在1秒内完成12. 日志与审计跟踪在某些领域如金融、医疗日志不仅用于调试还用于合规审计。这时需要考虑不可篡改性确保日志一旦写入就不能被修改可以考虑将日志写入只读介质使用区块链技术存储日志哈希定期将日志归档到安全存储完整时间戳使用高精度、同步的时间源from datetime import datetime, timezone logger.info(关键操作, extra{ timestamp: datetime.now(timezone.utc).isoformat(), user: current_user.id, action: delete_record })操作关联使用唯一ID关联相关操作import uuid request_id str(uuid.uuid4()) def process_request(request): logger.info(开始处理请求, extra{request_id: request_id}) # ...处理逻辑... logger.info(请求处理完成, extra{request_id: request_id})长期保留根据法规要求保留日志足够长时间实现自动归档和过期策略考虑冷热存储分层近期日志在快速存储旧日志在廉价存储访问控制限制谁可以访问审计日志日志文件设置严格的文件权限查询接口实现认证和授权记录所有对审计日志的访问13. 日志与错误追踪系统集成现代错误追踪系统如Sentry、Rollbar可以与日志系统集成提供更强大的错误监控能力Sentry集成示例import sentry_sdk from sentry_sdk.integrations.logging import LoggingIntegration # 所有日志级别ERROR的事件都会发送到Sentry sentry_logging LoggingIntegration( levellogging.INFO, # 捕获INFO及以上作为breadcrumbs event_levellogging.ERROR # 发送ERROR及以上作为事件 ) sentry_sdk.init( dsnyour-dsn-here, integrations[sentry_logging], traces_sample_rate1.0 ) # 普通日志使用方式不变 logger.error(发生了严重错误, extra{data: {user_id: 123}})错误上下文增强try: risky_operation() except Exception as e: logger.error(操作失败, exc_infoTrue, extra{ context: { user: current_user.id, data: sanitized_data } }) raise性能监控集成# 在Sentry中跟踪事务 with sentry_sdk.start_transaction(optask, nameprocess_data): logger.info(开始处理数据) # ...处理逻辑... logger.info(数据处理完成)自定义指纹控制错误分组方式try: process_payment() except PaymentError as e: logger.error(支付处理失败, exc_infoTrue, extra{ fingerprint: [payment, str(e.code)] })14. 日志与指标监控系统集成日志可以与Prometheus、StatsD等指标监控系统结合提供更全面的可观测性从日志提取指标from prometheus_client import Counter LOG_ERRORS Counter(app_log_errors, Number of ERROR logs, [logger]) class MetricsFilter(logging.Filter): def filter(self, record): if record.levelno logging.ERROR: LOG_ERRORS.labels(loggerrecord.name).inc() return True logger.addFilter(MetricsFilter())记录性能指标import time from prometheus_client import Histogram REQUEST_TIME Histogram(request_processing_seconds, Time spent processing request) def process_request(request): start_time time.time() logger.info(开始处理请求) try: # ...处理逻辑... logger.info(请求处理成功) finally: duration time.time() - start_time REQUEST_TIME.observe(duration) logger.info(请求处理完成, extra{duration: duration})结构化日志与指标关联# 日志 logger.info(API响应, extra{ endpoint: /users, status_code: 200, response_time: 0.45 }) # 对应的Prometheus指标 API_RESPONSES Counter(api_responses, API responses, [endpoint, status_code]) API_DURATION Histogram(api_duration, API response time, [endpoint]) class ApiMetricsFilter(logging.Filter): def filter(self, record): if hasattr(record, endpoint): API_RESPONSES.labels( endpointrecord.endpoint, status_codegetattr(record, status_code, 0) ).inc() if hasattr(record, response_time): API_DURATION.labels( endpointrecord.endpoint ).observe(record.response_time) return True15. 日志与分布式追踪系统集成在微服务架构中分布式追踪系统如Jaeger、Zipkin可以帮助跟踪请求在多个服务间的流转传播追踪上下文from opentelemetry import trace from opentelemetry.propagate import inject, extract # 在发送请求前注入追踪上下文 headers {} inject(headers) logger.info(发送请求, extra{headers: headers}) # 在接收请求时提取追踪上下文 context extract(headers) tracer trace.get_tracer(__name__) with tracer.start_as_current_span(process, contextcontext): logger.info(处理请求)在日志中包含追踪IDclass TracingFilter(logging.Filter): def filter(self, record): span trace.get_current_span() if span.is_recording(): record.trace_id format_trace_id(span.get_span_context().trace_id) record.span_id format_span_id(span.get_span_context().span_id) return True logger.addFilter(TracingFilter()) formatter logging.Formatter(%(trace_id)s %(span_id)s %(message)s)关联日志与追踪with tracer.start_as_current_span(database.query): logger.info(执行数据库查询, extra{ query: sanitized_query, parameters: parameters }) # ...执行查询...采样决策传播def should_log(record): span trace.get_current_span() if span.is_recording(): # 如果span被采样则记录详细日志 return True # 否则只记录错误 return record.levelno logging.ERROR class SamplingFilter(logging.Filter): def filter(self, record): return should_log(record)16. 日志与机器学习日志数据可以用于训练机器学习模型用于异常检测、预测性维护等场景日志模式挖掘from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import KMeans # 从日志中提取文本特征 logs [log.message for log in query_logs()] vectorizer TfidfVectorizer(max_features1000) X vectorizer.fit_transform(logs) # 聚类分析日志模式 kmeans KMeans(n_clusters10) clusters kmeans.fit_predict(X) # 分析每个簇的特征 for i in range(10): cluster_logs [log for log, c in zip(logs, clusters) if c i] logger.info(f日志簇{i}示例, extra{examples: cluster_logs[:3]})异常检测from sklearn.ensemble import IsolationForest # 训练异常检测模型 model IsolationForest(contamination0.01) model.fit(X) # 预测新日志是否异常 new_logs [log.message for log in query_recent_logs()] X_new vectorizer.transform(new_logs) anomalies model.predict(X_new) -1 for log, is_anomaly in zip(new_logs, anomalies): if is_anomaly: logger.warning(异常日志检测, extra{log: log})预测性维护import pandas as pd from prophet import Prophet # 将错误日志按时间聚合 error_logs query_error_logs() df pd.DataFrame({ ds: [log.timestamp for log in error_logs], y: 1 }).groupby(ds).sum().reset_index() # 预测未来错误趋势 model Prophet() model.fit(df) future model.make_future_dataframe(periods30) forecast model.predict(future) # 如果预测错误率上升触发告警 if forecast.tail(30)[yhat].mean() df[y].mean() * 1.5: logger.error(预测错误率将显著上升)日志分类from sklearn.pipeline import Pipeline from sklearn.linear_model import LogisticRegression # 准备已标记的训练数据 X_train [log.message for log in training_logs] y_train [log.label for log in training_logs] # 训练分类器 text_clf Pipeline([ (tfidf, TfidfVectorizer()), (clf, LogisticRegression()) ]) text_clf.fit(X_train, y_train) # 分类新日志 new_logs [log.message for log in unlabeled_logs] predictions text_clf.predict(new_logs) for log, label in zip(new_logs, predictions): logger.info(自动分类日志, extra{ log: log, predicted_label: label })17. 日志与安全分析日志是安全分析的重要数据源可以用于检测入侵、异常行为等暴力破解检测from collections import defaultdict failed_logins defaultdict(int) class SecurityFilter(logging.Filter): def filter(self, record): if (record.getMessage().startswith(登录失败) and hasattr(record, ip)): failed_logins[record.ip] 1 if failed_logins[record.ip] 5: logger.warning(可能的暴力破解尝试, extra{ ip: record.ip, count: failed_logins[record.ip] }) return True logger.addFilter(SecurityFilter())敏感操作监控SENSITIVE_ACTIONS [delete, grant, revoke, admin] class SensitiveActionFilter(logging.Filter): def filter(self, record): if hasattr(record, action): if any(action in record.action for action in SENSITIVE_ACTIONS): logger.info(敏感操作记录, extra{ user: getattr(record, user, unknown), action: record.action, target: getattr(record, target, unknown) }) return True异常时间活动检测import datetime class UnusualTimeFilter(logging.Filter): def filter(self, record): if hasattr(record, timestamp): hour datetime.datetime.fromisoformat(record.timestamp).hour if hour 8 or hour 20: # 非工作时间 logger.warning(非工作时间活动, extra{ user: getattr(record, user, unknown), time: record.timestamp, action: getattr(record, action, unknown) }) return True数据泄露检测import re PATTERNS [ r\b\d{4}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b, # 信用卡 r\b\d{3}[ -]?\d{2}[ -]?\d{4}\b, # SSN # 其他敏感数据模式 ] class DataLeakFilter(logging.Filter): def filter(self, record): message record.getMessage() for pattern in PATTERNS: if re.search(pattern, message): logger.error(可能的敏感数据泄露, extra{ pattern: pattern, log: message }) # 可以进一步redact敏感数据 record.msg re.sub(pattern, ***REDACTED***, message) return True18. 日志与合规性在许多行业日志记录是合规性要求的一部分。需要考虑数据保留策略from datetime import datetime, timedelta def cleanup_old_logs(): cutoff datetime.now() - timedelta(days365) # 保留1年 old_logs query_logs(beforecutoff) for log in old_logs: if not is_required_for_compliance(log): delete_log(log)访问审计class AccessAuditFilter(logging.Filter): def filter(self, record): if hasattr(record, access_type): logger.info(访问审计, extra{ user: record.user, resource: record.resource, action: record.action, timestamp: datetime.now(timezone.utc).isoformat() }) return True不可否认性import hashlib def create_log_hash(record): log_data json.dumps({ timestamp: record.timestamp, message: record.message, user: record.user }, sort_keysTrue).encode(utf-8) return hashlib.sha256(log_data).hexdigest() class NonRepudiationFilter(logging.Filter): def filter(self, record): record.hash create_log_hash(record) return True合规性报告def generate_compliance_report(): report { period: {