做Python开发几年日志这块一直是容易被低估、又特别能体现工程水平的地方。刚起步时我也觉得日志嘛不就是print换成了logging能打出东西来就行。直到线上服务跑了一个月半夜被人叫起来翻一个几个G的日志文件grep半天卡死终端才意识到日志管理不做好迟早要还债。后来陆陆续续把日志按日期分割、自动清理、多进程写入这些机制都补齐了整个运维体验完全不一样。这篇就把我实际项目中用到的方案完整梳理一遍重点讲清楚怎么用Python实现日志的自动化按日期分割把背后的原理、踩过的坑和可以直接抄的代码都放出来。1. 日志管理方案的整体设计与思路拆解1.1 日志增长的现实困境先别急着上代码想清楚一个问题你为什么要按日期分割日志大多数Python应用跑起来之后日志文件会持续增长。开发环境无所谓删了重建就行但线上环境完全不是一回事。我就见过一个数据分析平台跑了一周没重启日志文件涨到6个多G最后日志所在磁盘分区直接满了程序反而因为写不进去日志而崩溃。这种事故说出来很丢人但确实是没做好日志管理的典型案例。不分割日志会带来三个连锁问题排查问题效率极差。日志文件大了以后用grep或者tail去检索响应会明显变慢。文件越大越慢到后期根本没法用。历史日志无法归档。日志不只有排查当下问题的作用还要做趋势分析、异常统计。按月或者按天归档是后面做分析的基础。磁盘空间不可控。一个文件无限制增长你永远不知道它什么时候会把磁盘写满而磁盘写满往往是线上事故的高发诱因。所以按日期分割日志的本质是让日志文件从无边界增长变成可预期、可管理、可清理。这个需求在运维层面几乎是刚需。1.2 按日期分割的两种主流实现路线在Python生态里做按日期分割日志有两条路线。第一条是直接用标准库logging自带的TimedRotatingFileHandler。这个handler可以按时间间隔切换日志文件比如按小时、天、周来滚动。但它有个很别扭的地方日志文件命名是固定的比如app.log滚动后变成app.log.2024-01-01。如果你希望文件名直接带日期比如app-2024-01-01.log它做不到至少不能优雅地做到。第二条是自定义Handler在每次写入日志时判断当前日期如果日期变了就关闭旧文件、打开新文件。这种方式实现起来也不复杂但灵活度和命名控制力是最强的。我实际项目中用的就是这种方案后面会给出完整代码。两条路线各有适用场景如果你不介意固定主文件名加日期后缀用TimedRotatingFileHandler最快如果你希望文件名干净、直观、方便归档程序直接按文件名匹配建议走自定义Handler这条路。2. 核心细节解析日志分割的触发机制与关键参数2.1 理解TimedRotatingFileHandler的工作机制TimedRotatingFileHandler是标准库提供的方案理解它的工作机制对你自己写分割逻辑也很有帮助。它内部维护一个下一次滚动时间rolloverAt到了时间点就执行一次doRollover()把当前文件改名备份再创建一个新的主文件继续写。时间间隔由when参数决定S是秒、M是分钟、H是小时、D是天、W是周。backupCount参数控制保留几个备份文件超出部分自动删除。这个机制看着简单实际用起来有几个问题容易踩。第一个是命名问题。比如你设置whenD日志主文件是app.log滚动后备份文件叫app.log.2024-01-01你要是想按文件名排序会发现字符串排序在跨年份的时候会有问题。第二个是跨天切换的粒度。TimedRotatingFileHandler默认是到点就切换但如果你用的是D级别它是根据当前时间推算下一个零点处理跨日没问题。第三个是它滚动的是主文件名和很多团队的归档习惯不一致。我自己的建议是如果只是个人项目、简单脚本用TimedRotatingFileHandler可以省事但如果是正式项目建议还是自己控制分割逻辑后面扩展压缩、清理策略都会直观很多。2.2 自定义按日期分割Handler的设计要点自己写一个按日期分割的Handler核心思路其实就一句话写入日志前先检查当前日期如果日期和当前日志文件的日期不一致就切换文件。但要把这个逻辑写好有几个关键设计点必须想清楚。第一个是文件名的格式。我建议统一用app-YYYY-MM-DD.log这种格式年-月-日用零填充。这样文件名的字典序就是时间顺序归档和清理脚本可以省掉很多解析工作。第二个是切换的时机判断。每次emit_log时都做一次判断如果是新的一天就关闭旧文件流打开新文件流。这个操作理论上每次写日志都有一次额外的日期比对但因为只是取当前时间和比较字符串性能开销可以忽略不计。第三个是初始化逻辑。程序启动时日志文件名应该以当前日期为准而不是固定写死一个文件名。很多初学方案都是写死文件名然后只在日期变化时切换这样当天第一次启动写的就是昨天的日期排查日志时很容易产生混乱。第四个是异常保护。文件写入过程中可能出现磁盘满、权限拒绝、文件被占用等情况Handler内部要做好异常捕获避免日志模块的异常把主业务逻辑打断。这三个设计点我后面都会在代码里体现。2.3 为什么推荐结合when和backupCount双保险分割日志只是第一步自动清理才是日志管理的闭环。如果你只分割不清理那么最终磁盘上的日志文件总数会越来越多等于把一个大文件拆成了无数个小文件磁盘迟早还是会被写满。所以我会在Handler里同时加入保留天数的控制。比如只保留最近30天的日志文件超过30天的直接删除。这样整个日志目录的大小就是可控的。具体实现上每次切换文件时扫描日志目录下的文件列表解析文件名中的日期把早于保留天数的文件删除。这个操作不用很频繁每天切换文件时做一次就够了。3. 实操过程与核心代码实现3.1 完整自定义Handler代码与配置示例直接上代码。这是我实际项目中一直在用的方案做得比较保守优先保证稳定性和可读性。import os import re import logging from logging import Handler from datetime import datetime, timedelta class DailyFileHandler(Handler): 按日期分割日志文件的Handler文件名格式app-YYYY-MM-DD.log def __init__(self, log_dir, prefixapp, encodingutf-8, retain_days30): super().__init__() self.log_dir log_dir self.prefix prefix self.encoding encoding self.retain_days retain_days os.makedirs(log_dir, exist_okTrue) self.current_date datetime.now().strftime(%Y-%m-%d) self.log_file self._build_filename(self.current_date) self._open_file_stream() def _build_filename(self, date_str): return os.path.join(self.log_dir, f{self.prefix}-{date_str}.log) def _open_file_stream(self): self.stream open(self.log_file, a, encodingself.encoding) def _close_file_stream(self): if self.stream: self.stream.flush() self.stream.close() self.stream None def _need_rollover(self): return datetime.now().strftime(%Y-%m-%d) ! self.current_date def _do_rollover(self): self._close_file_stream() self.current_date datetime.now().strftime(%Y-%m-%d) self.log_file self._build_filename(self.current_date) self._open_file_stream() self._cleanup_old_logs() def _cleanup_old_logs(self): if not self.retain_days or self.retain_days 0: return cutoff datetime.now() - timedelta(daysself.retain_days) pattern re.compile(rf^{self.prefix}-(\d{{4}}-\d{{2}}-\d{{2}})\.log$) try: for filename in os.listdir(self.log_dir): match pattern.match(filename) if not match: continue file_date datetime.strptime(match.group(1), %Y-%m-%d) if file_date cutoff: file_path os.path.join(self.log_dir, filename) try: os.remove(file_path) print(f[log cleanup] removed {filename}) except OSError as e: self.handleError(None) # 删除失败不影响主流程 except OSError: pass def emit(self, record): try: if self._need_rollover(): self._do_rollover() msg self.format(record) self.stream.write(msg \n) self.stream.flush() except Exception: self.handleError(record) def close(self): self._close_file_stream() super().close()配套的logging配置代码import logging def setup_logging(): handler DailyFileHandler( log_dir/var/log/myapp, prefixmyapp, encodingutf-8, retain_days30 ) formatter logging.Formatter( %(asctime)s [%(levelname)s] %(name)s: %(message)s ) handler.setFormatter(formatter) root_logger logging.getLogger() root_logger.setLevel(logging.INFO) # 避免重复添加handler if not any(isinstance(h, DailyFileHandler) for h in root_logger.handlers): root_logger.addHandler(handler) # 同时向控制台输出 console_handler logging.StreamHandler() console_handler.setLevel(logging.WARNING) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) return root_logger用起来就是最标准的logging用法logger logging.getLogger(myapp.module) logger.info(服务已启动) logger.error(连接数据库失败, exc_infoTrue)这个方案我用了一年多稳定性和可维护性都验证过。核心逻辑很直白没有用到任何hack手段即使后面出了新需求上手改也很容易。3.2 配置演进用DictConfig管理日志参数如果项目里logger特别多不同模块日志级别不一样或者你有按模块拆分日志文件的需求建议把上面的setup_logging改成基于DictConfig的配置。Python 3.8以上都推荐用这种方式把日志配置独立出来避免在代码里到处设置。import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { default: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s } }, handlers: { daily_file: { class: myapp.log_handler.DailyFileHandler, log_dir: /var/log/myapp, prefix: myapp, retain_days: 30, formatter: default } }, root: { level: INFO, handlers: [daily_file] } } logging.config.dictConfig(LOGGING_CONFIG)这样做的好处是配置集中、可读性强换环境时也不用改代码改配置就行。尤其是多环境部署时dev环境的retain_days可以给短一点生产环境给长一点全部通过配置控制。3.3 多进程场景下的日志写入安全Python的logging在多线程下是安全的但多进程下写入同一个日志文件如果不做处理日志可能会交错错乱。原因是多个进程同时打开同一个文件句柄写入写的时候互相覆盖。这个问题在Gunicorn多worker、Celery多进程下特别普遍。我在实际项目中因为用到了Gunicorn多worker所以专门处理过。最简单的方案有两种。一种是ConcurrentRotatingFileHandler但它主要解决的是多进程按大小分割的场景按日期分割还是要自己兜底。另一种是把日志写到一个中间管道由单独进程统一处理。这个方案思路清晰但实现重。更轻量级的方案是在Handler里加一个进程级别的文件锁。虽然每个进程还是各自打文件但因为有了锁写入时不会互相穿插。缺点是锁会带来一些性能损耗但对绝大多数应用来说可以忽略。import fcntl import threading import time class SafeDailyFileHandler(DailyFileHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._lock threading.Lock() def emit(self, record): with self._lock: try: fcntl.flock(self.stream.fileno(), fcntl.LOCK_EX) super().emit(record) finally: fcntl.flock(self.stream.fileno(), fcntl.LOCK_UN)这个改进只在单机多进程场景下有效如果日志要跨机器汇总那就需要走日志采集链路了。不过作为应用层面的日志管理写锁已经能解决大部分实际问题。3.4 快捷方案loguru库的一行式分割如果你不想折腾标准库loguru库也是个很好的选择。它封装了很多细节按日期分割和日志清理都做得非常优雅。安装很简单pip install loguru然后配置也极其简洁from loguru import logger logger.add( /var/log/myapp/myapp-{time:YYYY-MM-DD}.log, rotation00:00, # 每天零点切换日志文件 retention30 days, # 保留最近30天日志 encodingutf-8, enqueueTrue, # 多进程安全写入 format{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}: {message} )就这几行按日期分割和自动清理都搞定了。loguru底层用的是标准库的logging机制但它在handler层面帮你把切换、清理、线程安全都处理好了。对于中小型项目来说切换到loguru的成本非常低收益却非常明显。我自己在个人项目和快速原型里常会直接用loguru在生产环境则会用自定义Handler因为生产环境的日志链路往往要对接已有的日志采集和分析系统自定义Handler更容易控制输出格式、路径和归档策略。4. 常见问题与排查技巧实录4.1 跨日切换延迟与今天写昨天文件的坑我自己遇到过最尴尬的bug日志每天切换文件后新文件里总是先出现几条昨天的日志。排查下来发现是应用里有一些长驻任务跨零点前后打印日志时时间戳取的是打印语句执行那一刻但文件流还停留在旧文件的buffer中。解决方案就是Handler里每次写入都主动flush。上面代码里的stream.flush()不是多余的它保证了每条日志都真正落盘。如果你因为性能考虑想批量flush也至少要保证切换文件前做一次flush。另一个经验是切换文件之后最好顺手关闭旧文件的fd。有些人直接替换文件名而不关闭旧连接导致文件句柄泄漏。时间长了你会看到Too many open files的错误。我在_handler里每次切换都会先_close_filestream实际效果非常稳定。4.2 文件权限与日志丢失的坑日志目录权限不对会导致写入失败而最坑的是logging默认会静默吞掉异常。你在代码里看到logger.info打印成功但文件里什么都没有排查半天发现是目录没创建成功。我的建议是在初始化时调用os.makedirs并检查目录是否可写宁可启动时就报错也不要等运行到某个业务逻辑时才发现日志写不进去。上面代码里已经做了exist_okTrue但还可以更进一步显式做一次文件打开测试。另外要注意临时目录和长期目录的区别。日志目录不要放到/tmp下避免被系统清理策略删掉。生产环境建议放在/var/log下并单独用logrotate做一层额外保障。4.3 时区与零点切换的歧义按日期分割有一个隐含前提你的当天用的是哪个时区。如果服务器是UTC业务是北京时间日志文件命名就很容易混乱。我见过一台美国服务器配了UTC时间然后业务日志按UTC切分排查问题时经常出现日志时间对不上业务时间的情况。解决思路是在初始化时明确指定时区。Python的datetime.now()默认用系统时区建议通过环境变量TZ显式控制或者在Handler里传入时区参数内部用pytz或者zoneinfo处理。from datetime import datetime from zoneinfo import ZoneInfo class DailyFileHandler(Handler): def __init__(self, log_dir, prefixapp, timezoneAsia/Shanghai, *args, **kwargs): super().__init__(*args, **kwargs) self.tz ZoneInfo(timezone) if timezone else None def _now(self): return datetime.now(self.tz) if self.tz else datetime.now()这样切分的文件名和内部记录的时间戳就完全一致了。4.4 文件清理与磁盘占用排查日志文件清理策略要和监控打通。即使有了retain_days自动清理也建议在日志目录的上一级做一层磁盘配额或监控报警。我经历过一次教训日志分割和清理都正常但某个突发流量高单日日志量暴涨一天写了20G虽然旧的清理了但磁盘还是在短时间内被占满了。从那以后我会在部署脚本里额外加一个crontab任务每日定时统计日志目录占用超过阈值就告警。同时日志目录单独挂载一个分区也可以这样即使日志磁盘满了也不会拖垮主业务程序所在的磁盘。5. 日志管理的进阶思路与实践总结5.1 从写日志到用日志按日期分割日志只是第一步。日志真正产生价值在于事后怎么查、怎么用。我建议在项目里顺手做三件事一是统一日志格式让每条日志都包含时间、级别、模块、请求IDtrace_id二是加结构化字段比如用户ID、订单号方便用grep或者日志分析平台快速定位三是把ERROR以上级别的日志单独输出一个文件平时排查重点问题不用翻全量日志。这些需求可以在Handler基础上再包装一层。比如在emit里额外判断record.levelno把ERROR级别写入单独的error-YYYY-MM-DD.log。这样的好处是排查问题时一开始就有明确的目标而不是在海量日志里捞针。5.2 日志文件对接采集与检索系统如果项目规模变大日志分析会从手工grep升级到采集检索。常见做法是把日志目录挂给Filebeat、Fluentd这类采集器把它吐到ES或ClickHouse里。此时你按日期分割的日志文件结构会带来很自然的采集优势采集器可以直接按文件名做数据分区或者用日志里的时间字段做事件时间。我踩过的坑是采集器只能识别某个固定文件名模式所以文件命名规则要提前定好越早统一越好。如果中途改命名规则旧日志和新日志的采集链路会断开历史排查很痛苦。5.3 最后再分享一个提升效率的小技巧大日志文件虽然按天分割后变小了但单日日志量仍然可能几百MB用grep直接搜效率还是低。我个人的习惯是每天在日志切割后自动压缩当天凌晨前的日志文件用gzip或者zstandard压缩。gzip压缩率对于文本日志可以达到80%以上300MB能压到40MB左右。需要排查时先把对应日期的压缩文件解压出来再搜磁盘占用和维护成本都降下来了。实现上可以加一个crontab任务每天凌晨3点执行一次find /var/log/myapp -name *.log -mtime 1 -exec gzip {} \;或者直接在Handler的_do_rollover里顺手把昨天的文件用gzip压缩。让处理流程自动衔接比事后手工操作靠谱得多。6. 写在项目之外我的一点实际体会日志管理的核心不在于用了多复杂的库或框架而在于把日志文件的生命周期想清楚。从创建、写入、切分、压缩、清理到最终被采集分析每个环节都要可控。我在实际项目中体会最深的一点是日志方案要尽早统一越晚改成本越高。很多团队一开始图省事直接用一个简单的FileHandler日志全部写在一个文件里。等线上出问题、发现日志查不动的时候再做按日期分割就得面对历史日志迁移、线上版本切换、团队成员适应等一系列连锁问题改动成本非常高。所以我特别建议新项目从第一天起就把日志分割、清理、命名规则定好哪怕初期日志量不大也要把机制建起来。这个基建投入换来的回报会在某个深夜排查问题时体现得淋漓尽致。
Python日志按日期分割实战：从TimedRotatingFileHandler到自定义Handler
做Python开发几年日志这块一直是容易被低估、又特别能体现工程水平的地方。刚起步时我也觉得日志嘛不就是print换成了logging能打出东西来就行。直到线上服务跑了一个月半夜被人叫起来翻一个几个G的日志文件grep半天卡死终端才意识到日志管理不做好迟早要还债。后来陆陆续续把日志按日期分割、自动清理、多进程写入这些机制都补齐了整个运维体验完全不一样。这篇就把我实际项目中用到的方案完整梳理一遍重点讲清楚怎么用Python实现日志的自动化按日期分割把背后的原理、踩过的坑和可以直接抄的代码都放出来。1. 日志管理方案的整体设计与思路拆解1.1 日志增长的现实困境先别急着上代码想清楚一个问题你为什么要按日期分割日志大多数Python应用跑起来之后日志文件会持续增长。开发环境无所谓删了重建就行但线上环境完全不是一回事。我就见过一个数据分析平台跑了一周没重启日志文件涨到6个多G最后日志所在磁盘分区直接满了程序反而因为写不进去日志而崩溃。这种事故说出来很丢人但确实是没做好日志管理的典型案例。不分割日志会带来三个连锁问题排查问题效率极差。日志文件大了以后用grep或者tail去检索响应会明显变慢。文件越大越慢到后期根本没法用。历史日志无法归档。日志不只有排查当下问题的作用还要做趋势分析、异常统计。按月或者按天归档是后面做分析的基础。磁盘空间不可控。一个文件无限制增长你永远不知道它什么时候会把磁盘写满而磁盘写满往往是线上事故的高发诱因。所以按日期分割日志的本质是让日志文件从无边界增长变成可预期、可管理、可清理。这个需求在运维层面几乎是刚需。1.2 按日期分割的两种主流实现路线在Python生态里做按日期分割日志有两条路线。第一条是直接用标准库logging自带的TimedRotatingFileHandler。这个handler可以按时间间隔切换日志文件比如按小时、天、周来滚动。但它有个很别扭的地方日志文件命名是固定的比如app.log滚动后变成app.log.2024-01-01。如果你希望文件名直接带日期比如app-2024-01-01.log它做不到至少不能优雅地做到。第二条是自定义Handler在每次写入日志时判断当前日期如果日期变了就关闭旧文件、打开新文件。这种方式实现起来也不复杂但灵活度和命名控制力是最强的。我实际项目中用的就是这种方案后面会给出完整代码。两条路线各有适用场景如果你不介意固定主文件名加日期后缀用TimedRotatingFileHandler最快如果你希望文件名干净、直观、方便归档程序直接按文件名匹配建议走自定义Handler这条路。2. 核心细节解析日志分割的触发机制与关键参数2.1 理解TimedRotatingFileHandler的工作机制TimedRotatingFileHandler是标准库提供的方案理解它的工作机制对你自己写分割逻辑也很有帮助。它内部维护一个下一次滚动时间rolloverAt到了时间点就执行一次doRollover()把当前文件改名备份再创建一个新的主文件继续写。时间间隔由when参数决定S是秒、M是分钟、H是小时、D是天、W是周。backupCount参数控制保留几个备份文件超出部分自动删除。这个机制看着简单实际用起来有几个问题容易踩。第一个是命名问题。比如你设置whenD日志主文件是app.log滚动后备份文件叫app.log.2024-01-01你要是想按文件名排序会发现字符串排序在跨年份的时候会有问题。第二个是跨天切换的粒度。TimedRotatingFileHandler默认是到点就切换但如果你用的是D级别它是根据当前时间推算下一个零点处理跨日没问题。第三个是它滚动的是主文件名和很多团队的归档习惯不一致。我自己的建议是如果只是个人项目、简单脚本用TimedRotatingFileHandler可以省事但如果是正式项目建议还是自己控制分割逻辑后面扩展压缩、清理策略都会直观很多。2.2 自定义按日期分割Handler的设计要点自己写一个按日期分割的Handler核心思路其实就一句话写入日志前先检查当前日期如果日期和当前日志文件的日期不一致就切换文件。但要把这个逻辑写好有几个关键设计点必须想清楚。第一个是文件名的格式。我建议统一用app-YYYY-MM-DD.log这种格式年-月-日用零填充。这样文件名的字典序就是时间顺序归档和清理脚本可以省掉很多解析工作。第二个是切换的时机判断。每次emit_log时都做一次判断如果是新的一天就关闭旧文件流打开新文件流。这个操作理论上每次写日志都有一次额外的日期比对但因为只是取当前时间和比较字符串性能开销可以忽略不计。第三个是初始化逻辑。程序启动时日志文件名应该以当前日期为准而不是固定写死一个文件名。很多初学方案都是写死文件名然后只在日期变化时切换这样当天第一次启动写的就是昨天的日期排查日志时很容易产生混乱。第四个是异常保护。文件写入过程中可能出现磁盘满、权限拒绝、文件被占用等情况Handler内部要做好异常捕获避免日志模块的异常把主业务逻辑打断。这三个设计点我后面都会在代码里体现。2.3 为什么推荐结合when和backupCount双保险分割日志只是第一步自动清理才是日志管理的闭环。如果你只分割不清理那么最终磁盘上的日志文件总数会越来越多等于把一个大文件拆成了无数个小文件磁盘迟早还是会被写满。所以我会在Handler里同时加入保留天数的控制。比如只保留最近30天的日志文件超过30天的直接删除。这样整个日志目录的大小就是可控的。具体实现上每次切换文件时扫描日志目录下的文件列表解析文件名中的日期把早于保留天数的文件删除。这个操作不用很频繁每天切换文件时做一次就够了。3. 实操过程与核心代码实现3.1 完整自定义Handler代码与配置示例直接上代码。这是我实际项目中一直在用的方案做得比较保守优先保证稳定性和可读性。import os import re import logging from logging import Handler from datetime import datetime, timedelta class DailyFileHandler(Handler): 按日期分割日志文件的Handler文件名格式app-YYYY-MM-DD.log def __init__(self, log_dir, prefixapp, encodingutf-8, retain_days30): super().__init__() self.log_dir log_dir self.prefix prefix self.encoding encoding self.retain_days retain_days os.makedirs(log_dir, exist_okTrue) self.current_date datetime.now().strftime(%Y-%m-%d) self.log_file self._build_filename(self.current_date) self._open_file_stream() def _build_filename(self, date_str): return os.path.join(self.log_dir, f{self.prefix}-{date_str}.log) def _open_file_stream(self): self.stream open(self.log_file, a, encodingself.encoding) def _close_file_stream(self): if self.stream: self.stream.flush() self.stream.close() self.stream None def _need_rollover(self): return datetime.now().strftime(%Y-%m-%d) ! self.current_date def _do_rollover(self): self._close_file_stream() self.current_date datetime.now().strftime(%Y-%m-%d) self.log_file self._build_filename(self.current_date) self._open_file_stream() self._cleanup_old_logs() def _cleanup_old_logs(self): if not self.retain_days or self.retain_days 0: return cutoff datetime.now() - timedelta(daysself.retain_days) pattern re.compile(rf^{self.prefix}-(\d{{4}}-\d{{2}}-\d{{2}})\.log$) try: for filename in os.listdir(self.log_dir): match pattern.match(filename) if not match: continue file_date datetime.strptime(match.group(1), %Y-%m-%d) if file_date cutoff: file_path os.path.join(self.log_dir, filename) try: os.remove(file_path) print(f[log cleanup] removed {filename}) except OSError as e: self.handleError(None) # 删除失败不影响主流程 except OSError: pass def emit(self, record): try: if self._need_rollover(): self._do_rollover() msg self.format(record) self.stream.write(msg \n) self.stream.flush() except Exception: self.handleError(record) def close(self): self._close_file_stream() super().close()配套的logging配置代码import logging def setup_logging(): handler DailyFileHandler( log_dir/var/log/myapp, prefixmyapp, encodingutf-8, retain_days30 ) formatter logging.Formatter( %(asctime)s [%(levelname)s] %(name)s: %(message)s ) handler.setFormatter(formatter) root_logger logging.getLogger() root_logger.setLevel(logging.INFO) # 避免重复添加handler if not any(isinstance(h, DailyFileHandler) for h in root_logger.handlers): root_logger.addHandler(handler) # 同时向控制台输出 console_handler logging.StreamHandler() console_handler.setLevel(logging.WARNING) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) return root_logger用起来就是最标准的logging用法logger logging.getLogger(myapp.module) logger.info(服务已启动) logger.error(连接数据库失败, exc_infoTrue)这个方案我用了一年多稳定性和可维护性都验证过。核心逻辑很直白没有用到任何hack手段即使后面出了新需求上手改也很容易。3.2 配置演进用DictConfig管理日志参数如果项目里logger特别多不同模块日志级别不一样或者你有按模块拆分日志文件的需求建议把上面的setup_logging改成基于DictConfig的配置。Python 3.8以上都推荐用这种方式把日志配置独立出来避免在代码里到处设置。import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { default: { format: %(asctime)s [%(levelname)s] %(name)s: %(message)s } }, handlers: { daily_file: { class: myapp.log_handler.DailyFileHandler, log_dir: /var/log/myapp, prefix: myapp, retain_days: 30, formatter: default } }, root: { level: INFO, handlers: [daily_file] } } logging.config.dictConfig(LOGGING_CONFIG)这样做的好处是配置集中、可读性强换环境时也不用改代码改配置就行。尤其是多环境部署时dev环境的retain_days可以给短一点生产环境给长一点全部通过配置控制。3.3 多进程场景下的日志写入安全Python的logging在多线程下是安全的但多进程下写入同一个日志文件如果不做处理日志可能会交错错乱。原因是多个进程同时打开同一个文件句柄写入写的时候互相覆盖。这个问题在Gunicorn多worker、Celery多进程下特别普遍。我在实际项目中因为用到了Gunicorn多worker所以专门处理过。最简单的方案有两种。一种是ConcurrentRotatingFileHandler但它主要解决的是多进程按大小分割的场景按日期分割还是要自己兜底。另一种是把日志写到一个中间管道由单独进程统一处理。这个方案思路清晰但实现重。更轻量级的方案是在Handler里加一个进程级别的文件锁。虽然每个进程还是各自打文件但因为有了锁写入时不会互相穿插。缺点是锁会带来一些性能损耗但对绝大多数应用来说可以忽略。import fcntl import threading import time class SafeDailyFileHandler(DailyFileHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._lock threading.Lock() def emit(self, record): with self._lock: try: fcntl.flock(self.stream.fileno(), fcntl.LOCK_EX) super().emit(record) finally: fcntl.flock(self.stream.fileno(), fcntl.LOCK_UN)这个改进只在单机多进程场景下有效如果日志要跨机器汇总那就需要走日志采集链路了。不过作为应用层面的日志管理写锁已经能解决大部分实际问题。3.4 快捷方案loguru库的一行式分割如果你不想折腾标准库loguru库也是个很好的选择。它封装了很多细节按日期分割和日志清理都做得非常优雅。安装很简单pip install loguru然后配置也极其简洁from loguru import logger logger.add( /var/log/myapp/myapp-{time:YYYY-MM-DD}.log, rotation00:00, # 每天零点切换日志文件 retention30 days, # 保留最近30天日志 encodingutf-8, enqueueTrue, # 多进程安全写入 format{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}: {message} )就这几行按日期分割和自动清理都搞定了。loguru底层用的是标准库的logging机制但它在handler层面帮你把切换、清理、线程安全都处理好了。对于中小型项目来说切换到loguru的成本非常低收益却非常明显。我自己在个人项目和快速原型里常会直接用loguru在生产环境则会用自定义Handler因为生产环境的日志链路往往要对接已有的日志采集和分析系统自定义Handler更容易控制输出格式、路径和归档策略。4. 常见问题与排查技巧实录4.1 跨日切换延迟与今天写昨天文件的坑我自己遇到过最尴尬的bug日志每天切换文件后新文件里总是先出现几条昨天的日志。排查下来发现是应用里有一些长驻任务跨零点前后打印日志时时间戳取的是打印语句执行那一刻但文件流还停留在旧文件的buffer中。解决方案就是Handler里每次写入都主动flush。上面代码里的stream.flush()不是多余的它保证了每条日志都真正落盘。如果你因为性能考虑想批量flush也至少要保证切换文件前做一次flush。另一个经验是切换文件之后最好顺手关闭旧文件的fd。有些人直接替换文件名而不关闭旧连接导致文件句柄泄漏。时间长了你会看到Too many open files的错误。我在_handler里每次切换都会先_close_filestream实际效果非常稳定。4.2 文件权限与日志丢失的坑日志目录权限不对会导致写入失败而最坑的是logging默认会静默吞掉异常。你在代码里看到logger.info打印成功但文件里什么都没有排查半天发现是目录没创建成功。我的建议是在初始化时调用os.makedirs并检查目录是否可写宁可启动时就报错也不要等运行到某个业务逻辑时才发现日志写不进去。上面代码里已经做了exist_okTrue但还可以更进一步显式做一次文件打开测试。另外要注意临时目录和长期目录的区别。日志目录不要放到/tmp下避免被系统清理策略删掉。生产环境建议放在/var/log下并单独用logrotate做一层额外保障。4.3 时区与零点切换的歧义按日期分割有一个隐含前提你的当天用的是哪个时区。如果服务器是UTC业务是北京时间日志文件命名就很容易混乱。我见过一台美国服务器配了UTC时间然后业务日志按UTC切分排查问题时经常出现日志时间对不上业务时间的情况。解决思路是在初始化时明确指定时区。Python的datetime.now()默认用系统时区建议通过环境变量TZ显式控制或者在Handler里传入时区参数内部用pytz或者zoneinfo处理。from datetime import datetime from zoneinfo import ZoneInfo class DailyFileHandler(Handler): def __init__(self, log_dir, prefixapp, timezoneAsia/Shanghai, *args, **kwargs): super().__init__(*args, **kwargs) self.tz ZoneInfo(timezone) if timezone else None def _now(self): return datetime.now(self.tz) if self.tz else datetime.now()这样切分的文件名和内部记录的时间戳就完全一致了。4.4 文件清理与磁盘占用排查日志文件清理策略要和监控打通。即使有了retain_days自动清理也建议在日志目录的上一级做一层磁盘配额或监控报警。我经历过一次教训日志分割和清理都正常但某个突发流量高单日日志量暴涨一天写了20G虽然旧的清理了但磁盘还是在短时间内被占满了。从那以后我会在部署脚本里额外加一个crontab任务每日定时统计日志目录占用超过阈值就告警。同时日志目录单独挂载一个分区也可以这样即使日志磁盘满了也不会拖垮主业务程序所在的磁盘。5. 日志管理的进阶思路与实践总结5.1 从写日志到用日志按日期分割日志只是第一步。日志真正产生价值在于事后怎么查、怎么用。我建议在项目里顺手做三件事一是统一日志格式让每条日志都包含时间、级别、模块、请求IDtrace_id二是加结构化字段比如用户ID、订单号方便用grep或者日志分析平台快速定位三是把ERROR以上级别的日志单独输出一个文件平时排查重点问题不用翻全量日志。这些需求可以在Handler基础上再包装一层。比如在emit里额外判断record.levelno把ERROR级别写入单独的error-YYYY-MM-DD.log。这样的好处是排查问题时一开始就有明确的目标而不是在海量日志里捞针。5.2 日志文件对接采集与检索系统如果项目规模变大日志分析会从手工grep升级到采集检索。常见做法是把日志目录挂给Filebeat、Fluentd这类采集器把它吐到ES或ClickHouse里。此时你按日期分割的日志文件结构会带来很自然的采集优势采集器可以直接按文件名做数据分区或者用日志里的时间字段做事件时间。我踩过的坑是采集器只能识别某个固定文件名模式所以文件命名规则要提前定好越早统一越好。如果中途改命名规则旧日志和新日志的采集链路会断开历史排查很痛苦。5.3 最后再分享一个提升效率的小技巧大日志文件虽然按天分割后变小了但单日日志量仍然可能几百MB用grep直接搜效率还是低。我个人的习惯是每天在日志切割后自动压缩当天凌晨前的日志文件用gzip或者zstandard压缩。gzip压缩率对于文本日志可以达到80%以上300MB能压到40MB左右。需要排查时先把对应日期的压缩文件解压出来再搜磁盘占用和维护成本都降下来了。实现上可以加一个crontab任务每天凌晨3点执行一次find /var/log/myapp -name *.log -mtime 1 -exec gzip {} \;或者直接在Handler的_do_rollover里顺手把昨天的文件用gzip压缩。让处理流程自动衔接比事后手工操作靠谱得多。6. 写在项目之外我的一点实际体会日志管理的核心不在于用了多复杂的库或框架而在于把日志文件的生命周期想清楚。从创建、写入、切分、压缩、清理到最终被采集分析每个环节都要可控。我在实际项目中体会最深的一点是日志方案要尽早统一越晚改成本越高。很多团队一开始图省事直接用一个简单的FileHandler日志全部写在一个文件里。等线上出问题、发现日志查不动的时候再做按日期分割就得面对历史日志迁移、线上版本切换、团队成员适应等一系列连锁问题改动成本非常高。所以我特别建议新项目从第一天起就把日志分割、清理、命名规则定好哪怕初期日志量不大也要把机制建起来。这个基建投入换来的回报会在某个深夜排查问题时体现得淋漓尽致。
