1. 为什么放着现成的监控平台不用还要自己写脚本先交代一下背景。很多人一看标题就会问Zabbix、Prometheus、Graylog、夜莺现成的监控平台一大堆日志采集、告警通知、可视化面板全都给你配好了为什么还要用Python自己写一个日志监控脚本这不是重复造轮子吗这个质疑放在大规模生产环境里是成立的。几十台、上百台服务器的场景下你确实应该老老实实部署一套Prometheus加Grafana或者用Zabbix做集中告警用Graylog做日志聚合。这些东西经过大量生产环境验证有完整的权限体系、告警路由、历史数据存储不是个人脚本能比的。但我写这个脚本的出发点恰恰相反——我遇到过太多“杀鸡用牛刀”的尴尬场面。一次是在一个只有3台服务器的中小型项目里团队花了一周时间部署了一套完整的监控平台结果后续没人愿意维护告警规则写得乱七八糟最后大家直接不看监控面板了。另一次是在一个边缘计算节点上环境非常受限压根不可能跑Java写的采集组件只有Python环境是现成的。还有一类场景更常见你手上有一个老旧系统的日志文件格式极其特殊既不是标准JSON也不是常见分隔符现成监控平台的日志解析规则根本匹配不了。你总不能为了几行特殊日志去改平台源码吧这时候用Python写一个针对性极强的监控脚本反而是效率最高的方案。所以我把这个项目的适用场景说清楚你可以自己对号入座服务器数量不多几台到十几台不想为了监控而专门维护一套平台日志格式非常特殊现成工具解析困难或者解析规则写起来比写脚本还麻烦需要快速响应今天提需求明天就想要效果没时间走平台配置流程运行环境资源有限跑不起重量级监控组件想把日志监控的逻辑嵌入到你已有的Python业务系统里当然这也不是说脚本方案就没有短板。比如缺少历史趋势图、没有统一的告警聚合页面、多节点日志集中管理困难这些确实是它的先天不足。理解了这些边界你才知道什么时候该用脚本什么时候该上平台。这个项目的核心思路其实不复杂用Python读取目标日志文件的新增内容通过正则表达式匹配预设的异常模式一旦命中就把告警消息通过邮件、IM机器人等方式推送出去。整个流程里最难的不是写代码本身而是把“怎么读文件”“怎么判异常”“怎么发通知”“怎么防止告警轰炸”这几件事做扎实。2. 开工之前先把需求搞清楚到底要监控什么、报什么警我在网上看到很多人写日志监控脚本一上来就贴代码好像只要把读文件和发邮件的代码拼在一起就完事了。实际上动手写代码之前如果没把需求理清楚后面返工的概率极高。我自己做这个项目时先花了半天时间梳理需求核心问题就三个监控对象、异常定义、通知策略。2.1 明确监控对象到底有哪些日志要看系统里常见的日志文件有这些应用日志比如Nginx的access.log和error.log、Tomcat的catalina.out、Java应用自己写的log文件系统日志Linux下的/var/log/messages、/var/log/syslog、/var/log/secure登录安全日志业务日志订单系统、支付系统自己输出的业务日志通常包含关键业务状态第三方组件日志Redis、MySQL、Kafka等中间件写的日志不是所有日志都需要监控也没必要把每一个文件都纳入进来。我的建议是先梳理出直接影响线上稳定性的核心日志。判断标准很简单——如果这个日志文件里的某类报错你24小时内没发现会不会出大事会就纳入监控不会先放一边。以我当时的项目为例最终纳入了三个文件日志文件原因关注重点/var/log/nginx/error.logNginx是流量入口挂了或报错直接影响线上连接超时、worker进程崩溃、上游不可达/var/log/messages系统级错误和硬件状态OOM、磁盘I/O错误、内核报错/app/logs/order.log核心业务日志下单失败、支付回调异常、数据库连接失败这个清单不是一成不变的。过程中我发现order.log里大量出现某个第三方接口的报错但那个接口本身就不稳定属于“已知问题”我就把这条规则临时去掉了否则它会一直在告警列表里刷屏反而掩盖了真正需要关注的问题。2.2 异常怎么定义正则表达式的颗粒度问题这是整个项目里最需要仔细琢磨的部分。异常定义得太宽天天误报定义得太窄漏报。这个度怎么把握我的经验是不要试图匹配“所有可能的异常”而是精准匹配“必须马上处理的异常”。比如Nginx的error.log里下面几种情况性质完全不同connect() failed (111: Connection refused)——上游服务挂了必须马上报警worker_connections are not enough——连接数配置不足需要扩容或调参也要报警client closed connection——客户端主动断开大部分时候不用管可能只是用户关掉了页面各种级别的debug信息——完全不用管如果你用一句宽泛的error去匹配结果就是把所有噪音全部拉进来。正确做法是维护一个规则列表每个规则包含“匹配模式”和“告警级别”。正则表达式的写法也有讲究。我用的模式大致长这样patterns [ { name: upstream_connection_refused, pattern: rconnect\(\) failed.*Connection refused, level: critical, description: 上游服务连接被拒绝服务可能已宕机 }, { name: worker_connections_not_enough, pattern: rworker_connections are not enough, level: warning, description: Nginx worker连接数不足 }, { name: out_of_memory, pattern: rOut of memory|Killed process, level: critical, description: 系统发生OOM } ]几个写正则时的经验能不匹配行首行尾就不要加^和$因为日志行往往有前缀时间戳、级别、进程号用.*做模糊匹配时注意颗粒度太贪婪会误匹配到不相关的行模式要覆盖同一种异常的不同写法。比如“Out of memory”和“Killed process”在OOM时都可能出现放同一个规则里用|连接每一条规则必须能在真实日志里找到对应样例去验证凭空写的正则基本都要返工2.3 通知策略什么级别发什么渠道如果只有一种通知渠道策略就很简单——匹配到就发。但一旦你有邮件、短信、IM机器人等多个渠道就得想清楚什么情况发邮件什么情况直接打手机我的策略是分级处理critical级别严重立即推送邮件、IM机器人一起发短信也发warning级别警告只发IM机器人不惊动手机info级别信息只记录到本地不发任何通知这个分级非常重要否则你会被告警搞得精神衰弱。我在项目初期没做分级所有异常都发邮件加IM结果第三天大家就开始“选择性忽略”告警了。这个问题的根源不在大家不负责而在于告警系统没有帮人分清轻重缓急。还有一个容易被忽略的点告警内容必须包含“足够上下文”。光是“order.log出现异常”这种消息等于没发。至少要有日志文件路径、具体匹配的规则、原始日志内容完整那一行、发生时间、服务器IP。如果能加上日志前后的几行上下文排查效率会高很多。3. 核心设计一个可扩展的多文件日志监控框架需求梳理清楚后技术方案的选型就顺理成章了。这个项目里我采用的方案是Python 3.6以上版本、watchdog做文件变更监听、正则表达式做模式匹配、smtplib加requests库做多渠道通知。这一节我把整个框架的架构和核心模块拆开讲后面附上可以直接用的代码。3.1 技术选型为什么用watchdog而不是自己写死循环读日志文件最朴素的做法是死循环里用open()打开文件一行一行读读完sleep一下继续读。这个方案能跑但有两个问题一是日志文件被轮转logrotate时你打开的文件句柄还指向旧文件二是轮询间隔不好把握太频繁浪费CPU太久又影响实时性。这时候有两条路自己实现一个像tail -f那样的文件跟踪器或者直接用watchdog库监听文件系统事件。我最终选了watchdog原因有三watchdog不是简单轮询它底层通过inotifyLinux或ReadDirectoryChangesWWindows接收文件系统的真实变更事件实时性比轮询好得多代码复杂度低不用自己处理各种文件状态机跨平台同一套代码在Linux和Windows上都能跑当然用watchdog也有一个要处理的问题它监听的是“文件系统事件”文件被轮转时你会收到on_created、on_deleted、on_moved一系列事件得自己判断应该重新打开哪个文件。不过这里我想多说一句不要把watchdog神化。它并不是日志监控的银弹。如果你的日志文件写入量极大每秒几万行watchdog事件会非常密集反而导致性能问题。这种情况下直接用tail -f做流式读取或者按偏移量轮询反而更可控。中小规模的日志量watchdog完全够用。3.2 整体架构四个模块各司其职我把整个脚本拆成四个模块每个模块只做一件事方便后续扩展log_monitor/ ├── monitor.py # 主程序调度逻辑进程入口 ├── file_watcher.py # 文件监听模块封装watchdog输出新增内容 ├── rule_engine.py # 规则引擎对新增内容做正则匹配 └── notification.py # 通知模块邮件、IM机器人、本地记录模块化设计的好处我在实际开发中感受很深。最典型的例子是项目上线后没几天业务方提了一个新需求——希望某些特定类型的日志不仅发通知还要写入一个单独的明细文件用来做周报统计。这时候我只需要在rule_engine.py里加一个“命中后同时写入明细文件”的处理分支其他模块完全不用动。如果当初把所有逻辑写在一个脚本文件里这个需求至少得动两三个地方改起来风险大得多。所以哪怕你的项目再小我也建议按职责拆分不要图省事写一个五六百行的单文件脚本。3.3 文件监听模块核心代码详解先说一个关键点监控日志文件不是简单“读到文件内容”就完事而是要“持续跟踪文件的新增内容”并且要正确处理文件轮转、文件被截断、文件被删除重建这些边界情况。import time import os from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogFileHandler(FileSystemEventHandler): 监听日志文件目录当目标文件被写入时触发处理 def __init__(self, file_path, callback, position_holderNone): self.file_path os.path.abspath(file_path) self.callback callback # position_holder 用于记录每个文件当前读到的偏移量 self.positions position_holder if position_holder is not None else {} # 打开文件并定位到文件末尾效果等同于 tail -f self._open_file() self._seek_to_end() def _open_file(self): try: self._file open(self.file_path, r, encodingutf-8, errorsignore) except FileNotFoundError: # 文件可能还没创建稍后由事件触发时再尝试打开 self._file None def _seek_to_end(self): 启动时定位到文件末尾避免重复处理历史日志 if self._file: self._file.seek(0, os.SEEK_END) self.positions[self.file_path] self._file.tell() def on_modified(self, event): if event.src_path ! self.file_path: return if self._file is None: self._open_file() if self._file is None: return self.positions[self.file_path] 0 self._read_new_lines() def _read_new_lines(self): 读取从上次位置到文件末尾的新内容 while True: line self._file.readline() if not line: break self.callback(line.rstrip(\n)) self.positions[self.file_path] self._file.tell() def stop(self): if self._file: self._file.close()这个实现里有几个细节值得注意。第一个编码问题。日志文件的编码格式千奇百怪UTF-8是常见情况但有些老系统用的是GBK或者其它编码。这里errorsignore的作用是防止因为个别字节解析失败导致整个读取流程崩溃。代价是这部分内容会丢失但相比监控程序挂掉丢几行乱码日志是可以接受的。如果业务上要求不能丢可以改为errorsreplace用替代字符保住整行内容。第二个定位到文件末尾。启动脚本时默认不处理历史日志避免告警轰炸。但如果你第一次部署时想先跑一遍历史日志检查有没有存量问题可以把_seek_to_end改成可选参数默认跳到最后指定参数时从头开始。第三个文件轮转的问题。日志文件被logrotate重命名后watchdog会触发一系列事件但on_modified里我们只检查event.src_path是否等于目标路径。轮转后原路径指向的是新创建的文件由于我们的_file对象还指向旧文件会一直读不到新内容。这是新手最容易踩的坑。解决思路有两种一种是用on_created事件去判断文件是否被重建重建了就重新open另一种是每次on_modified触发时都检查一下当前文件的inode是否变了。我采用的方案是在模块里加一个简单的轮转检测def on_modified(self, event): if event.src_path ! self.file_path: return # 检查当前文件是否已经被替换轮转场景 if self._file: current_inode os.stat(self.file_path).st_ino old_inode os.fstat(self._file.fileno()).st_ino if current_inode ! old_inode: self._file.close() self._open_file() if self._file: # 新文件从开头读否则会漏掉轮转期间写入的内容 self.positions[self.file_path] 0 self._read_new_lines()3.4 规则引擎正则匹配与多规则处理逻辑文件监听模块把新增日志行传给回调函数后规则引擎开始工作。规则引擎的核心逻辑是遍历规则列表逐条匹配当前行命中了就触发告警。这里有一个性能优化的细节规则多了之后逐条正则匹配是有开销的。如果日志量很大正则引擎会变成瓶颈。我的优化思路是“预过滤加精匹配”import re class RuleEngine: def __init__(self, rules): self.rules [] for rule in rules: compiled { name: rule[name], level: rule.get(level, warning), description: rule.get(description, ), regex: re.compile(rule[pattern]), } self.rules.append(compiled) def match(self, line): 对单行日志做规则匹配返回命中的规则列表 hits [] # 预过滤先做一次普通字符串包含判断能过滤掉大部分无关行 # 注意这里只是为了性能优化不能用它替代正式正则 for rule in self.rules: if rule[regex].search(line): hits.append(rule) return hits你可能注意到我特意写了预过滤的注释。这个技巧的收益取决于规则模式的设计。如果一个规则本身就是简单子串匹配比如Connection refused那么正则和执行if Connection refused in line开销差别不大。但规则一多、日志量大时预过滤能省下不少CPU。再往深一层说规则引擎里还需要考虑“同一个异常连续出现多次”的问题。比如上游服务挂了Nginx的error.log会在短时间内刷出几百行一模一样的错误。如果你每一行都发一条告警告警平台直接被刷爆。这个问题的处理我会在“防告警轰炸”那一节专门展开这里先留个钩子。3.5 主程序把所有模块串起来主程序的职责很简单初始化各个模块启动监听挂载规则引擎和通知模块。它的代码量不大但整个系统是否能优雅退出、是否能处理意外异常都看这一层。import time import logging from file_watcher import LogFileHandler from rule_engine import RuleEngine from notification import Notifier from watchdog.observers import Observer logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.FileHandler(log_monitor.log), logging.StreamHandler(), ], ) logger logging.getLogger(log_monitor) def main(): config load_config() # 初始化通知模块邮件 企业微信机器人 notifier Notifier( email_configconfig[email], webhook_configconfig[webhook], ) # 初始化规则引擎 engine RuleEngine(config[rules]) # 文件监听和事件回调 observer Observer() watcher LogFileHandler( file_pathconfig[log_path], callbacklambda line: handle_line(line, engine, notifier, config), ) # 注意watchdog的schedule是针对目录的不是针对单个文件 # 这里要监听文件所在目录然后在事件处理器里过滤文件名 observer.schedule( watcher, pathconfig[log_dir], recursiveFalse, ) observer.start() logger.info(f日志监控已启动监听文件: {config[log_path]}) try: while True: time.sleep(2) except KeyboardInterrupt: observer.stop() observer.join() def handle_line(line, engine, notifier, config): # 简单地按行处理先不做聚合 hits engine.match(line) for hit in hits: notifier.send( levelhit[level], titlef[日志告警] {hit[name]}, contentline, ) def load_config(): # 实际项目中建议使用YAML或环境变量管理配置 return { log_path: /var/log/nginx/error.log, log_dir: /var/log/nginx, rules: [ { name: upstream_connection_refused, pattern: rconnect\(\) failed.*Connection refused, level: critical, }, ], email: { smtp_server: smtp.example.com, smtp_port: 465, username: monitorexample.com, password: your_password, to: [opsexample.com], }, webhook: { url: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx, }, } if __name__ __main__: main()这里有一个非常容易踩的坑observer.schedule接收的是一个目录路径而LogFileHandler监听的是单个文件。如果你把/var/log/nginx/error.log直接传给schedule代码不会报错但根本不会生效。watchdog对单文件的监听需要你自己在事件处理器里过滤也就是我在代码里写的“监听目录、在on_modified里比对event.src_path”的思路。4. 踩坑最多的三个地方logrotate、编码和正则误报代码能跑通只是第一步真正让日志监控脚本稳定运行需要跨过好几个坑。这三个坑我全部踩过每一个都让我排查了半天时间写出来给大家避避雷。4.1 logrotate轮转导致监控“静默失明”日志轮转是Linux系统管理里的标准操作日志文件按天或按大小切分旧的压缩归档。但对我的监控脚本来说轮转就是一次“换血”文件路径没变但inode变了文件内容被截断或新建。我第一次跑监控脚本时测试阶段一切正常结果第二天早上起来发现昨晚一条告警都没收到。排查了半天最后发现是凌晨logrotate执行了日志轮转监控脚本的文件句柄还指向旧文件新日志全部写到了新文件里而我的脚本一无所知。解决思路就是我上面说的inode比对。每次on_modified触发时检查一下当前路径文件的inode和已打开文件的inode是否一致不一致说明文件被替换了重新打开新文件。这里要特别注意一点轮转后新文件通常是空的如果你打开后从第0行开始读就需要自己处理“文件被重建后要不要读已有内容”的问题。我的处理是直接把偏移量归零再从新文件开头读因为轮转瞬间写入的内容很少重新读一遍最多产生一两条重复告警这个代价可以接受。4.2 编码问题比想象中更恶心很多日志监控脚本的处理逻辑里编码是被忽略的一个问题。直到你第一次遇到UnicodeDecodeError。我的情况是这样的一个Java应用输出的日志里混入了GBK编码的中文而监控脚本默认用UTF-8读取结果某一行解析直接报错。更恶心的是Python的readline()读到一半抛异常后文件指针的位置是不确定的可能导致后续内容全部读取错乱。第一版代码我没加errorsignore结果脚本在运行第6个小时就挂掉了。加上errorsignore之后问题解决但丢失了部分内容。后面对编码问题敏感的场景特别是日志里包含业务数据、中文编码不一致等我改成了更稳妥的方案try: line self._file.readline() if not line: break except UnicodeDecodeError: # 单行解码失败跳过这行但保证文件指针继续前进 # 这里用readline读入二进制再手动decode避免指针错乱 raw_line self._file.buffer.readline() self.callback(f[编码错误] 日志行无法解码原始字节长度: {len(raw_line)}) continue这个方案的逻辑是遇到解码错误时不去尝试修复那一行而是用二进制方式把这一整行读掉保证指针不会卡死然后记录一条“该行无法解码”的信息。这样既不会导致整个监控挂掉也不会丢失大量后续内容。实际部署中我推荐在启动脚本时先对目标日志文件做一次编码探测确认主要的编码格式再针对性地设置解码方案。Python的charset-normalizer库chardet的替代品可以自动检测文件编码但注意检测大文件时速度较慢建议只对样本做检测。4.3 正则规则写得不准告警内容反而误导人正则误报的问题在日志监控里很常见。经典案例你想匹配error结果日志里有一条“Error: something went wrong in the error handler”正好你的正则和业务上下文不匹配就会误报。我的做法是每条规则必须同时提供“必须匹配”和“必须不匹配”的上下文条件。规则引擎写成这样{ name: db_connection_failed, pattern: rdatabase connection (failed|error|refused), exclude_pattern: rretry|timeout waiting for idle connection, level: critical, }排除模式的含义是即使主模式匹配了但如果当前行还包含retry或timeout waiting for idle connection这类表示“正在自动恢复”的内容就不发告警。这个设计来源于一次真实事故数据库连接池在浪涌时会出现大量“connection error”日志但很多是重试中的短暂错误并不代表数据库真挂了。如果全部告警值班同学会收到大量无效消息真正的集群故障反而不容易被注意到。正则的精细化需要不断迭代。我建议上线初期设置一个“调试模式”把正则命中结果和日志原文记录下来连续观察几天不断修正规则确认稳定后再把调试模式关掉。不要指望第一版正则就完美这不现实。5. 警报通知的多种姿势邮件、IM机器人、本地弹窗规则引擎匹配到异常日志后接下来就是通知模块的活了。通知方式的选择直接决定了告警能不能被及时看到、会不会被刷屏反感。这一节我把邮件、企业微信/钉钉机器人、本地桌面通知三种实现方式都讲一遍你可以按需选择组合。5.1 邮件通知经典但不该是唯一渠道用smtplib发邮件是最经典的通知方式好处是部署简单、几乎所有场景都支持而且邮件本身就是正式沟通渠道适合做记录留痕。但它的致命弱点是时效性差。一封告警邮件发出去值班同学不一定会立刻看见尤其是深夜或者在忙别的事情时。所以我把邮件定义为“事后追溯”的告警渠道而不是“第一响应”的渠道。核心系统故障时优先靠IM机器人和短信把人叫醒邮件只是补充。邮件通知的关键代码import smtplib from email.mime.text import MIMEText from email.header import Header def send_email(config, subject, content): msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] config[username] msg[To] ,.join(config[to]) try: if config.get(smtp_port) 465: server smtplib.SMTP_SSL(config[smtp_server], config[smtp_port]) else: server smtplib.SMTP(config[smtp_server], config[smtp_port]) server.starttls() server.login(config[username], config[password]) server.sendmail( config[username], config[to], msg.as_string(), ) server.quit() except Exception as e: # 发邮件失败不能导致主流程崩溃要记录日志 logger.error(f邮件发送失败: {e})两个容易出问题的点一是很多企业邮箱的SMTP端口在25、465、587之间需要对齐465用SSL587用STARTTLS25通常明文。我在代码里做了一个简单判断实际使用时要根据你的邮件服务商来配置。二是邮箱的授权码问题很多邮箱用密码登不上SMTP必须用客户端专用授权码。5.2 IM机器人群消息才是最适合值班的告警方式国内业务团队基本都在用企业微信或钉钉这两者都提供了群机器人Webhook接口发送告警消息到群里。这也是我实际项目里用的主力通知渠道。企业微信机器人的用法是在群里添加一个自定义机器人得到一个Webhook URL用POST方式向这个URL发送一个JSON消息体机器人就会在群里说话。具体实现如下import requests import json def send_webhook(url, content): headers {Content-Type: application/json} payload { msgtype: text, text: { content: content, }, } try: resp requests.post(url, jsonpayload, headersheaders, timeout5) resp.raise_for_status() except Exception as e: logger.error(fWebhook发送失败: {e})企业微信机器人还支持markdown类型的消息体可以做得更精致一点比如用不同颜色标注告警级别、添加超链接跳转到日志平台。钉钉机器人的消息格式略有不同但思路一样。我踩过的一个坑是Webhook URL里有一个key参数这个参数必须被完整保留否则机器人会返回invalid webhook。如果配置在YAML文件里记得给URL加上引号避免特殊字符被解析掉。另一个坑是限流问题。企业微信机器人有频率限制每分钟最多20条。告警风暴时你一秒发几十条消息会触发限流后面的全被丢弃。所以通知模块里我加了一个简单的队列和节流机制import queue import threading import time class ThrottledNotifier: 带节流控制的通知器防止触发IM平台限流 def __init__(self, min_interval3): self.min_interval min_interval self.last_send_time 0 self.lock threading.Lock() self.pending [] def send(self, message): with self.lock: self.pending.append(message) now time.time() if now - self.last_send_time self.min_interval: self._flush() def _flush(self): if not self.pending: return messages self.pending.copy() self.pending.clear() # 多个消息合并成一条发送降低触发频率 combined \n.join(messages) send_webhook(WEBHOOK_URL, combined) self.last_send_time time.time()这个设计的核心是把短时间内多条告警合并成一条消息发送既避免限流也减少群里刷屏。合并消息对值班同学来说也更友好他们只需要看一条汇总而不是把几十条单条消息看完。5.3 本地通知开发调试时的好帮手本地桌面通知适合在开发调试时使用。你把脚本集成到自己的开发环境里日志一有异常就弹一个系统通知立刻看到效果。跨平台的方案是用plyer库from plyer import notification def send_local_notification(title, message): notification.notify( titletitle, messagemessage, timeout10, )这个方案在Windows和macOS上都能用但要注意它依赖系统通知服务在一些精简版Linux桌面环境上可能不工作。服务器上是没有桌面的所以本地通知只用于开发机调试不要部署到生产服务器。5.4 多渠道怎么协同主备关系和降级策略通知模块的真正难点不在于“能发”而在于“发不出去的时候怎么办”。我部署的监控脚本把通知模块做成了主备模式优先IM机器人如果IM发送失败比如网络抖动、机器人被移出群自动降级发邮件如果邮件也失败就把告警写入本地文件。这个降级策略看起来简单但保证了告警不丢失。写入本地文件这个兜底方案虽然不能主动触达值班同学但至少保留了证据事后排查时可以回溯。如果你对告警即时性要求特别高还可以再加上短信或电话告警的接口但需要额外接入短信服务商的API一般场景用不上。6. 防止告警轰炸去重、聚合与升级机制不能少任何告警系统在上线运行一段时间后都会遇到同一个问题——告警疲劳。第一个月还认真看第三个月基本就“又是它不用管”。这个问题的根源通常不是值班同学态度有问题而是告警系统没有做好去重、聚合和升级。6.1 相同告警合并只报一次但持续提醒急性告警场景里同一个异常可能在一分钟内出现几十次。如果每次都要发群消息瞬间爆炸。我的做法是引入“窗口期”概念同一规则或者按规则加内容指纹在窗口期内最多只发一次。实现方式很直接用一个字典记录每个规则的最近告警时间class Deduplicator: def __init__(self, window_seconds300): self.window_seconds window_seconds self.last_alert_time {} def should_alert(self, rule_name): now time.time() last self.last_alert_time.get(rule_name, 0) if now - last self.window_seconds: self.last_alert_time[rule_name] now return True return False但“只报一次”也有问题如果这个异常持续了30分钟值班同学看完第一次告警后可能就去处理了30分钟内没有再收到提醒直到问题升级成事故。所以更合理的方案是在窗口期内不重复发但窗口期结束后如果异常还在再发一次并且告警内容里附上“从第一次出现到现在持续了多久”的信息。这会大幅提升值班效率。6.2 相似告警聚合把“一堆碎消息”变成“一条汇总”比去重更进一步的是聚合。假设Nginx的上游服务挂了error.log里每分钟会产生“connect() failed”和“worker_connections are not enough”两种不同类型的错误。如果分开发就是两条告警。但如果在一个时间窗口内这两种错误都指向同一个根因上游服务宕机导致连接堆积那它们应该合并成一条包含两个维度的告警消息。聚合的一个简单实现是在固定时间窗口比如60秒内把所有命中的告警规则和它们出现的次数统计出来窗口结束时统一发送一次。class Aggregator: def __init__(self, window_seconds60): self.window_seconds window_seconds self.events {} self.lock threading.Lock() def add_event(self, rule, line): with self.lock: if rule not in self.events: self.events[rule] {count: 0, sample_lines: [], first_seen: time.time()} self.events[rule][count] 1 if len(self.events[rule][sample_lines]) 3: self.events[rule][sample_lines].append(line) def flush(self): with self.lock: events self.events self.events {} return events聚合后的告警内容大致长这样[汇总告警] 60秒内共捕获3类异常总计42次 1. upstream_connection_refused | 发生次数: 30 示例: 2026/05/14 10:00:03 [error] connect() failed (111: Connection refused) 2. worker_connections_not_enough | 发生次数: 10 示例: ... 3. upstream_response_timeout | 发生次数: 2 示例: ...值班同学看到这样一条汇总比看到42条零散消息要舒服得多。而且汇总里给了示例日志可以直接去日志平台搜索定位。6.3 告警升级一直没人处理就一直升级去重和聚合解决的是“告警太多”的问题告警升级解决的是“告警被忽略”的问题。我把告警升级分成两个维度第一个维度是“时间升级”一条critical级别的告警发出后如果30分钟内没有被确认ack升级为更高优先级通知同时抄送团队leader。第二个维度是“重复升级”同一规则在1小时内多次触发说明问题没有被解决自动提高告警级别。实现时间升级需要一个“告警确认”机制。你可以在通知消息里附带一个确认链接比如点击后调用Webhook标记已确认但这个会让系统复杂度上一个台阶。如果你不想做得太重退而求其次的方案是定期扫描“未处理告警列表”对超过指定时间的告警发一条“您有一条告警尚未确认”的提醒。这个逻辑用定时任务实现就行不需要引入复杂的状态机。我在实际项目里做了一版简化实现告警按5分钟窗口聚合并发送如果某条规则在连续3个窗口里都命中则触发“持续告警”通知内容明确说明“已持续15分钟未恢复”。这样不用值班同学手动确认系统也能起到升级提醒的作用。6.4 静默与白名单有些告警应该允许被“关掉”体系设计得再完善也总有需要临时调整的时刻。比如某个第三方接口在做计划内变更会短暂出现一批报错日志或者某个已知问题临时无法解决会周期性告警。这时候如果没有“静默”机制告警轰炸只会让人对系统失去信任。我的做法是维护一个静默规则列表支持两种匹配方式按规则名静默、按日志内容关键词静默。{ silent_rules: [ {rule_name: third_party_api_timeout, expire_at: 2026-05-20 18:00:00}, ], silent_keywords: [ {keyword: planned maintenance, expire_at: 2026-05-15 23:59:59}, ], }静默规则必须设置过期时间不允许永久静默。忘记关闭静默是运维事故的常见诱因设置过期时间能强制你重新审视这条规则到底还需不需要。7. 工程化落地从脚本到可维护的小系统前面六节的内容集中在“核心逻辑”上但一个脚本从“我本地能跑”变成“生产环境稳定运行”中间还差着很多工程化的功夫。这一节把部署上线过程中必须考虑的几个问题讲完。7.1 配置管理把“改代码”变成“改配置”第一版脚本我把规则、邮箱账号、Webhook地址全部硬编码在代码里。结果就是每次调整规则都要改代码、重启进程非常痛苦而且容易改错。后来我把配置全部抽离到config.yaml代码逻辑完全不感知配置细节。规则变更、通知渠道变更、静默策略调整全部通过改配置文件完成不需要动代码。# config.yaml monitor: log_path: /var/log/nginx/error.log log_dir: /var/log/nginx rules: - name: upstream_connection_refused pattern: connect\(\) failed.*Connection refused exclude_pattern: level: critical description: 上游服务连接被拒绝服务可能已宕机 alert_window_seconds: 300 - name: worker_connections_not_enough pattern: worker_connections are not enough exclude_pattern: level: warning description: Nginx worker连接数不足 alert_window_seconds: 600 notification: email: smtp_server: smtp.example.com smtp_port: 465 username: monitorexample.com password: your_password to: [opsexample.com] webhook: url: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx silent: - rule_name: third_party_api_timeout expire_at: 2026-05-20 18:00:00用YAML管理配置有个隐藏好处配置文件本身就是规则文档。以后新同学接手这个监控系统翻一翻配置文件就知道目前监控了哪些日志、关注哪些异常、告警发到哪完全不需要去代码里猜。7.2 运行状态可视化一个简单的状态文件脚本运行得好不好不能靠“感觉”。我用一个状态文件记录最近一次启动时间、最近一次读到日志偏移量、最近一次命中规则、最近一次成功发送通知。脚本每处理完一批日志就更新一次状态文件。{ start_time: 2026-05-14 10:00:00, last_read_position: 123456789, last_match_time: 2026-05-14 10:05:30, last_match_rule: upstream_connection_refused, last_notify_time: 2026-05-14 10:05:31, last_notify_success: true, total_lines_processed: 12345, }这个状态文件的核心价值在于当你怀疑监控系统已经“静默失效”时看一眼状态文件里last_read_position是否持续增长就能判断脚本到底是在正常消费日志还是已经卡住。我自己排查问题时这个状态文件帮了大忙。7.3 systemd守护别用nohup后台运行真的把Python脚本部署到Linux服务器上最忌讳直接用nohup python monitor.py 跑。这种方式进程管理能力为零进程挂了没人知道、开机不会自动启动、日志管理混乱。正确做法是写一个systemd service文件[Unit] DescriptionPython Log Monitor Service Afternetwork.target [Service] Typesimple WorkingDirectory/opt/log_monitor ExecStart/usr/bin/python3 /opt/log_monitor/monitor.py Restartalways RestartSec5 Userroot [Install] WantedBymulti-user.target配置好后sudo cp log_monitor.service /etc/systemd/system/ sudo systemctl daemon-reload sudo systemctl enable log_monitor sudo systemctl start log_monitorsystemd会负责守护进程异常退出时自动重启Restartalways服务器重启后自动拉起enable。日常维护只需要systemctl status log_monitor、systemctl restart log_monitor就够了。这一步带来的稳定性提升是任何代码层面的优化都比不了的。7.4 自监控监控脚本自己也会出问题聊一个运维圈经常提到的问题谁来监控监控系统我已经在项目里踩过这个坑——监控脚本因为未捕获的异常挂掉后我整整两天没收到任何告警最后还是看业务同学反馈才发现监控已经失效。从那以后我给自己定了两条规矩。第一条主循环里任何单条日志处理异常都不能导致整个进程退出。异常捕获到并记录到日志后继续处理下一条不能因为一条异常日志就让整个监控停摆。第二条写一个简单的定时任务比如cron每5分钟跑一次检查监控进程是否存活如果不存活就发一条告警#!/bin/bash # /opt/log_monitor/check_monitor.sh if ! systemctl is-active --quiet log_monitor; then curl -s -X POST https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx \ -H Content-Type: application/json \ -d {msgtype:text,text:{content:【监控告警】日志监控服务未运行请立即检查}} /dev/null fi把这个脚本加进crontab每5分钟执行一次。这样即使监控脚本挂了告警链路也不会断。7.5 性能实测与调优这个脚本能扛多大流量最后说说性能表现。我部署的这套脚本实际监控三个日志文件峰值时的日志写入速度大约每秒2000行。整个脚本的CPU占用率稳定在3%以内内存占用约30MB。对于常规的中小规模应用这个指标完全够用。但如果你的日志量达到每秒几万行甚至百万行级别Python脚本就力不从心了。这时需要换链路用Filebeat做日志采集、Kafka做消息队列、Elasticsearch做存储、Kibana做可视化再用ElastAlert做告警规则。整个技术栈就复杂多了那已经不属于“脚本”的范畴而是一个完整的日志系统。8. 给这段实战经验补个尾巴几个容易被忽视的小细节项目做到这里整个监控系统已经能稳定运行了。但每次回顾这个项目总还能想起一些当时容易被忽略的小细节。这些细节不写进正式文档却实实在在影响使用体验和维护成本。第一个细节是日志时间戳的时区问题。日志文件里的时间往往是服务器本地时间而告警消息发出的时间是你脚本所在机器的时间。如果两者时区不一致排查问题时会出现“告警说10点05分发生日志里却是9点05分”的混乱。我在加载配置时加上了一个timezone参数发告警时把时间统一转换能少掉很多沟通成本。更简单的方法是直接在告警内容里同时附上服务器时间和监控端时间让值班同学自己对照。第二个细节是告警消息里要附上“日志文件的完整路径”加上“日志行号”。这条在排查时非常有用——你直接复制路径到服务器上看文件就行不用再去猜是哪台机器哪个文件。第三个细节是配置文件里的密码和Webhook密钥。这些敏感信息直接写在YAML里如果代码仓库是共享的等于把钥匙交给了所有人。更安全的方式是用环境变量或者专门的密钥管理工具比如Vault代码运行的时候动态获取。如果项目复杂度不高至少也要确保配置文件不要提交到git仓库里。第四个细节是日志监控脚本本身运行时的日志。很多人搭好了监控却忘了监控脚本自己也会产生日志我用的是log_monitor.log。不要小看这个日志文件等你排查“为什么没告警”的时候它就是第一手线索。日志文件也要注意定期清理避免长时间运行后把磁盘塞满。最简单的方式是用crontab定期执行logrotate或者干脆用系统的logrotate配置管理它。最后一个建议如果你第一次做这类日志监控项目不要一上来就贪大求全。先选一个日志文件、配三条规则、用一个通知渠道跑通全流程体验一遍“日志写入 - 脚本捕获 - 匹配规则 - 发出告警”的完整链路。确认一切正常后再逐步扩展规则、增加被监控的日志文件池、接入更多通知渠道。一口吃不成胖子监控系统的迭代和线上系统的迭代一样都是“小步快跑”最稳妥。我这个项目从第一行代码到稳定运行大概用了一周多中间大部分时间不是在写代码而是在打磨规则、调试边界条件、处理各种“意外”。但这一遍下来后续所有类似需求我都有了完整的解决框架和避坑清单这个积累的收益比脚本本身大得多。如果你也要在项目里做日志监控希望这篇文字能帮你少走几段弯路。 Python日志监控脚本实战：文件监听、正则匹配与告警推送
1. 为什么放着现成的监控平台不用还要自己写脚本先交代一下背景。很多人一看标题就会问Zabbix、Prometheus、Graylog、夜莺现成的监控平台一大堆日志采集、告警通知、可视化面板全都给你配好了为什么还要用Python自己写一个日志监控脚本这不是重复造轮子吗这个质疑放在大规模生产环境里是成立的。几十台、上百台服务器的场景下你确实应该老老实实部署一套Prometheus加Grafana或者用Zabbix做集中告警用Graylog做日志聚合。这些东西经过大量生产环境验证有完整的权限体系、告警路由、历史数据存储不是个人脚本能比的。但我写这个脚本的出发点恰恰相反——我遇到过太多“杀鸡用牛刀”的尴尬场面。一次是在一个只有3台服务器的中小型项目里团队花了一周时间部署了一套完整的监控平台结果后续没人愿意维护告警规则写得乱七八糟最后大家直接不看监控面板了。另一次是在一个边缘计算节点上环境非常受限压根不可能跑Java写的采集组件只有Python环境是现成的。还有一类场景更常见你手上有一个老旧系统的日志文件格式极其特殊既不是标准JSON也不是常见分隔符现成监控平台的日志解析规则根本匹配不了。你总不能为了几行特殊日志去改平台源码吧这时候用Python写一个针对性极强的监控脚本反而是效率最高的方案。所以我把这个项目的适用场景说清楚你可以自己对号入座服务器数量不多几台到十几台不想为了监控而专门维护一套平台日志格式非常特殊现成工具解析困难或者解析规则写起来比写脚本还麻烦需要快速响应今天提需求明天就想要效果没时间走平台配置流程运行环境资源有限跑不起重量级监控组件想把日志监控的逻辑嵌入到你已有的Python业务系统里当然这也不是说脚本方案就没有短板。比如缺少历史趋势图、没有统一的告警聚合页面、多节点日志集中管理困难这些确实是它的先天不足。理解了这些边界你才知道什么时候该用脚本什么时候该上平台。这个项目的核心思路其实不复杂用Python读取目标日志文件的新增内容通过正则表达式匹配预设的异常模式一旦命中就把告警消息通过邮件、IM机器人等方式推送出去。整个流程里最难的不是写代码本身而是把“怎么读文件”“怎么判异常”“怎么发通知”“怎么防止告警轰炸”这几件事做扎实。2. 开工之前先把需求搞清楚到底要监控什么、报什么警我在网上看到很多人写日志监控脚本一上来就贴代码好像只要把读文件和发邮件的代码拼在一起就完事了。实际上动手写代码之前如果没把需求理清楚后面返工的概率极高。我自己做这个项目时先花了半天时间梳理需求核心问题就三个监控对象、异常定义、通知策略。2.1 明确监控对象到底有哪些日志要看系统里常见的日志文件有这些应用日志比如Nginx的access.log和error.log、Tomcat的catalina.out、Java应用自己写的log文件系统日志Linux下的/var/log/messages、/var/log/syslog、/var/log/secure登录安全日志业务日志订单系统、支付系统自己输出的业务日志通常包含关键业务状态第三方组件日志Redis、MySQL、Kafka等中间件写的日志不是所有日志都需要监控也没必要把每一个文件都纳入进来。我的建议是先梳理出直接影响线上稳定性的核心日志。判断标准很简单——如果这个日志文件里的某类报错你24小时内没发现会不会出大事会就纳入监控不会先放一边。以我当时的项目为例最终纳入了三个文件日志文件原因关注重点/var/log/nginx/error.logNginx是流量入口挂了或报错直接影响线上连接超时、worker进程崩溃、上游不可达/var/log/messages系统级错误和硬件状态OOM、磁盘I/O错误、内核报错/app/logs/order.log核心业务日志下单失败、支付回调异常、数据库连接失败这个清单不是一成不变的。过程中我发现order.log里大量出现某个第三方接口的报错但那个接口本身就不稳定属于“已知问题”我就把这条规则临时去掉了否则它会一直在告警列表里刷屏反而掩盖了真正需要关注的问题。2.2 异常怎么定义正则表达式的颗粒度问题这是整个项目里最需要仔细琢磨的部分。异常定义得太宽天天误报定义得太窄漏报。这个度怎么把握我的经验是不要试图匹配“所有可能的异常”而是精准匹配“必须马上处理的异常”。比如Nginx的error.log里下面几种情况性质完全不同connect() failed (111: Connection refused)——上游服务挂了必须马上报警worker_connections are not enough——连接数配置不足需要扩容或调参也要报警client closed connection——客户端主动断开大部分时候不用管可能只是用户关掉了页面各种级别的debug信息——完全不用管如果你用一句宽泛的error去匹配结果就是把所有噪音全部拉进来。正确做法是维护一个规则列表每个规则包含“匹配模式”和“告警级别”。正则表达式的写法也有讲究。我用的模式大致长这样patterns [ { name: upstream_connection_refused, pattern: rconnect\(\) failed.*Connection refused, level: critical, description: 上游服务连接被拒绝服务可能已宕机 }, { name: worker_connections_not_enough, pattern: rworker_connections are not enough, level: warning, description: Nginx worker连接数不足 }, { name: out_of_memory, pattern: rOut of memory|Killed process, level: critical, description: 系统发生OOM } ]几个写正则时的经验能不匹配行首行尾就不要加^和$因为日志行往往有前缀时间戳、级别、进程号用.*做模糊匹配时注意颗粒度太贪婪会误匹配到不相关的行模式要覆盖同一种异常的不同写法。比如“Out of memory”和“Killed process”在OOM时都可能出现放同一个规则里用|连接每一条规则必须能在真实日志里找到对应样例去验证凭空写的正则基本都要返工2.3 通知策略什么级别发什么渠道如果只有一种通知渠道策略就很简单——匹配到就发。但一旦你有邮件、短信、IM机器人等多个渠道就得想清楚什么情况发邮件什么情况直接打手机我的策略是分级处理critical级别严重立即推送邮件、IM机器人一起发短信也发warning级别警告只发IM机器人不惊动手机info级别信息只记录到本地不发任何通知这个分级非常重要否则你会被告警搞得精神衰弱。我在项目初期没做分级所有异常都发邮件加IM结果第三天大家就开始“选择性忽略”告警了。这个问题的根源不在大家不负责而在于告警系统没有帮人分清轻重缓急。还有一个容易被忽略的点告警内容必须包含“足够上下文”。光是“order.log出现异常”这种消息等于没发。至少要有日志文件路径、具体匹配的规则、原始日志内容完整那一行、发生时间、服务器IP。如果能加上日志前后的几行上下文排查效率会高很多。3. 核心设计一个可扩展的多文件日志监控框架需求梳理清楚后技术方案的选型就顺理成章了。这个项目里我采用的方案是Python 3.6以上版本、watchdog做文件变更监听、正则表达式做模式匹配、smtplib加requests库做多渠道通知。这一节我把整个框架的架构和核心模块拆开讲后面附上可以直接用的代码。3.1 技术选型为什么用watchdog而不是自己写死循环读日志文件最朴素的做法是死循环里用open()打开文件一行一行读读完sleep一下继续读。这个方案能跑但有两个问题一是日志文件被轮转logrotate时你打开的文件句柄还指向旧文件二是轮询间隔不好把握太频繁浪费CPU太久又影响实时性。这时候有两条路自己实现一个像tail -f那样的文件跟踪器或者直接用watchdog库监听文件系统事件。我最终选了watchdog原因有三watchdog不是简单轮询它底层通过inotifyLinux或ReadDirectoryChangesWWindows接收文件系统的真实变更事件实时性比轮询好得多代码复杂度低不用自己处理各种文件状态机跨平台同一套代码在Linux和Windows上都能跑当然用watchdog也有一个要处理的问题它监听的是“文件系统事件”文件被轮转时你会收到on_created、on_deleted、on_moved一系列事件得自己判断应该重新打开哪个文件。不过这里我想多说一句不要把watchdog神化。它并不是日志监控的银弹。如果你的日志文件写入量极大每秒几万行watchdog事件会非常密集反而导致性能问题。这种情况下直接用tail -f做流式读取或者按偏移量轮询反而更可控。中小规模的日志量watchdog完全够用。3.2 整体架构四个模块各司其职我把整个脚本拆成四个模块每个模块只做一件事方便后续扩展log_monitor/ ├── monitor.py # 主程序调度逻辑进程入口 ├── file_watcher.py # 文件监听模块封装watchdog输出新增内容 ├── rule_engine.py # 规则引擎对新增内容做正则匹配 └── notification.py # 通知模块邮件、IM机器人、本地记录模块化设计的好处我在实际开发中感受很深。最典型的例子是项目上线后没几天业务方提了一个新需求——希望某些特定类型的日志不仅发通知还要写入一个单独的明细文件用来做周报统计。这时候我只需要在rule_engine.py里加一个“命中后同时写入明细文件”的处理分支其他模块完全不用动。如果当初把所有逻辑写在一个脚本文件里这个需求至少得动两三个地方改起来风险大得多。所以哪怕你的项目再小我也建议按职责拆分不要图省事写一个五六百行的单文件脚本。3.3 文件监听模块核心代码详解先说一个关键点监控日志文件不是简单“读到文件内容”就完事而是要“持续跟踪文件的新增内容”并且要正确处理文件轮转、文件被截断、文件被删除重建这些边界情况。import time import os from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogFileHandler(FileSystemEventHandler): 监听日志文件目录当目标文件被写入时触发处理 def __init__(self, file_path, callback, position_holderNone): self.file_path os.path.abspath(file_path) self.callback callback # position_holder 用于记录每个文件当前读到的偏移量 self.positions position_holder if position_holder is not None else {} # 打开文件并定位到文件末尾效果等同于 tail -f self._open_file() self._seek_to_end() def _open_file(self): try: self._file open(self.file_path, r, encodingutf-8, errorsignore) except FileNotFoundError: # 文件可能还没创建稍后由事件触发时再尝试打开 self._file None def _seek_to_end(self): 启动时定位到文件末尾避免重复处理历史日志 if self._file: self._file.seek(0, os.SEEK_END) self.positions[self.file_path] self._file.tell() def on_modified(self, event): if event.src_path ! self.file_path: return if self._file is None: self._open_file() if self._file is None: return self.positions[self.file_path] 0 self._read_new_lines() def _read_new_lines(self): 读取从上次位置到文件末尾的新内容 while True: line self._file.readline() if not line: break self.callback(line.rstrip(\n)) self.positions[self.file_path] self._file.tell() def stop(self): if self._file: self._file.close()这个实现里有几个细节值得注意。第一个编码问题。日志文件的编码格式千奇百怪UTF-8是常见情况但有些老系统用的是GBK或者其它编码。这里errorsignore的作用是防止因为个别字节解析失败导致整个读取流程崩溃。代价是这部分内容会丢失但相比监控程序挂掉丢几行乱码日志是可以接受的。如果业务上要求不能丢可以改为errorsreplace用替代字符保住整行内容。第二个定位到文件末尾。启动脚本时默认不处理历史日志避免告警轰炸。但如果你第一次部署时想先跑一遍历史日志检查有没有存量问题可以把_seek_to_end改成可选参数默认跳到最后指定参数时从头开始。第三个文件轮转的问题。日志文件被logrotate重命名后watchdog会触发一系列事件但on_modified里我们只检查event.src_path是否等于目标路径。轮转后原路径指向的是新创建的文件由于我们的_file对象还指向旧文件会一直读不到新内容。这是新手最容易踩的坑。解决思路有两种一种是用on_created事件去判断文件是否被重建重建了就重新open另一种是每次on_modified触发时都检查一下当前文件的inode是否变了。我采用的方案是在模块里加一个简单的轮转检测def on_modified(self, event): if event.src_path ! self.file_path: return # 检查当前文件是否已经被替换轮转场景 if self._file: current_inode os.stat(self.file_path).st_ino old_inode os.fstat(self._file.fileno()).st_ino if current_inode ! old_inode: self._file.close() self._open_file() if self._file: # 新文件从开头读否则会漏掉轮转期间写入的内容 self.positions[self.file_path] 0 self._read_new_lines()3.4 规则引擎正则匹配与多规则处理逻辑文件监听模块把新增日志行传给回调函数后规则引擎开始工作。规则引擎的核心逻辑是遍历规则列表逐条匹配当前行命中了就触发告警。这里有一个性能优化的细节规则多了之后逐条正则匹配是有开销的。如果日志量很大正则引擎会变成瓶颈。我的优化思路是“预过滤加精匹配”import re class RuleEngine: def __init__(self, rules): self.rules [] for rule in rules: compiled { name: rule[name], level: rule.get(level, warning), description: rule.get(description, ), regex: re.compile(rule[pattern]), } self.rules.append(compiled) def match(self, line): 对单行日志做规则匹配返回命中的规则列表 hits [] # 预过滤先做一次普通字符串包含判断能过滤掉大部分无关行 # 注意这里只是为了性能优化不能用它替代正式正则 for rule in self.rules: if rule[regex].search(line): hits.append(rule) return hits你可能注意到我特意写了预过滤的注释。这个技巧的收益取决于规则模式的设计。如果一个规则本身就是简单子串匹配比如Connection refused那么正则和执行if Connection refused in line开销差别不大。但规则一多、日志量大时预过滤能省下不少CPU。再往深一层说规则引擎里还需要考虑“同一个异常连续出现多次”的问题。比如上游服务挂了Nginx的error.log会在短时间内刷出几百行一模一样的错误。如果你每一行都发一条告警告警平台直接被刷爆。这个问题的处理我会在“防告警轰炸”那一节专门展开这里先留个钩子。3.5 主程序把所有模块串起来主程序的职责很简单初始化各个模块启动监听挂载规则引擎和通知模块。它的代码量不大但整个系统是否能优雅退出、是否能处理意外异常都看这一层。import time import logging from file_watcher import LogFileHandler from rule_engine import RuleEngine from notification import Notifier from watchdog.observers import Observer logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.FileHandler(log_monitor.log), logging.StreamHandler(), ], ) logger logging.getLogger(log_monitor) def main(): config load_config() # 初始化通知模块邮件 企业微信机器人 notifier Notifier( email_configconfig[email], webhook_configconfig[webhook], ) # 初始化规则引擎 engine RuleEngine(config[rules]) # 文件监听和事件回调 observer Observer() watcher LogFileHandler( file_pathconfig[log_path], callbacklambda line: handle_line(line, engine, notifier, config), ) # 注意watchdog的schedule是针对目录的不是针对单个文件 # 这里要监听文件所在目录然后在事件处理器里过滤文件名 observer.schedule( watcher, pathconfig[log_dir], recursiveFalse, ) observer.start() logger.info(f日志监控已启动监听文件: {config[log_path]}) try: while True: time.sleep(2) except KeyboardInterrupt: observer.stop() observer.join() def handle_line(line, engine, notifier, config): # 简单地按行处理先不做聚合 hits engine.match(line) for hit in hits: notifier.send( levelhit[level], titlef[日志告警] {hit[name]}, contentline, ) def load_config(): # 实际项目中建议使用YAML或环境变量管理配置 return { log_path: /var/log/nginx/error.log, log_dir: /var/log/nginx, rules: [ { name: upstream_connection_refused, pattern: rconnect\(\) failed.*Connection refused, level: critical, }, ], email: { smtp_server: smtp.example.com, smtp_port: 465, username: monitorexample.com, password: your_password, to: [opsexample.com], }, webhook: { url: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx, }, } if __name__ __main__: main()这里有一个非常容易踩的坑observer.schedule接收的是一个目录路径而LogFileHandler监听的是单个文件。如果你把/var/log/nginx/error.log直接传给schedule代码不会报错但根本不会生效。watchdog对单文件的监听需要你自己在事件处理器里过滤也就是我在代码里写的“监听目录、在on_modified里比对event.src_path”的思路。4. 踩坑最多的三个地方logrotate、编码和正则误报代码能跑通只是第一步真正让日志监控脚本稳定运行需要跨过好几个坑。这三个坑我全部踩过每一个都让我排查了半天时间写出来给大家避避雷。4.1 logrotate轮转导致监控“静默失明”日志轮转是Linux系统管理里的标准操作日志文件按天或按大小切分旧的压缩归档。但对我的监控脚本来说轮转就是一次“换血”文件路径没变但inode变了文件内容被截断或新建。我第一次跑监控脚本时测试阶段一切正常结果第二天早上起来发现昨晚一条告警都没收到。排查了半天最后发现是凌晨logrotate执行了日志轮转监控脚本的文件句柄还指向旧文件新日志全部写到了新文件里而我的脚本一无所知。解决思路就是我上面说的inode比对。每次on_modified触发时检查一下当前路径文件的inode和已打开文件的inode是否一致不一致说明文件被替换了重新打开新文件。这里要特别注意一点轮转后新文件通常是空的如果你打开后从第0行开始读就需要自己处理“文件被重建后要不要读已有内容”的问题。我的处理是直接把偏移量归零再从新文件开头读因为轮转瞬间写入的内容很少重新读一遍最多产生一两条重复告警这个代价可以接受。4.2 编码问题比想象中更恶心很多日志监控脚本的处理逻辑里编码是被忽略的一个问题。直到你第一次遇到UnicodeDecodeError。我的情况是这样的一个Java应用输出的日志里混入了GBK编码的中文而监控脚本默认用UTF-8读取结果某一行解析直接报错。更恶心的是Python的readline()读到一半抛异常后文件指针的位置是不确定的可能导致后续内容全部读取错乱。第一版代码我没加errorsignore结果脚本在运行第6个小时就挂掉了。加上errorsignore之后问题解决但丢失了部分内容。后面对编码问题敏感的场景特别是日志里包含业务数据、中文编码不一致等我改成了更稳妥的方案try: line self._file.readline() if not line: break except UnicodeDecodeError: # 单行解码失败跳过这行但保证文件指针继续前进 # 这里用readline读入二进制再手动decode避免指针错乱 raw_line self._file.buffer.readline() self.callback(f[编码错误] 日志行无法解码原始字节长度: {len(raw_line)}) continue这个方案的逻辑是遇到解码错误时不去尝试修复那一行而是用二进制方式把这一整行读掉保证指针不会卡死然后记录一条“该行无法解码”的信息。这样既不会导致整个监控挂掉也不会丢失大量后续内容。实际部署中我推荐在启动脚本时先对目标日志文件做一次编码探测确认主要的编码格式再针对性地设置解码方案。Python的charset-normalizer库chardet的替代品可以自动检测文件编码但注意检测大文件时速度较慢建议只对样本做检测。4.3 正则规则写得不准告警内容反而误导人正则误报的问题在日志监控里很常见。经典案例你想匹配error结果日志里有一条“Error: something went wrong in the error handler”正好你的正则和业务上下文不匹配就会误报。我的做法是每条规则必须同时提供“必须匹配”和“必须不匹配”的上下文条件。规则引擎写成这样{ name: db_connection_failed, pattern: rdatabase connection (failed|error|refused), exclude_pattern: rretry|timeout waiting for idle connection, level: critical, }排除模式的含义是即使主模式匹配了但如果当前行还包含retry或timeout waiting for idle connection这类表示“正在自动恢复”的内容就不发告警。这个设计来源于一次真实事故数据库连接池在浪涌时会出现大量“connection error”日志但很多是重试中的短暂错误并不代表数据库真挂了。如果全部告警值班同学会收到大量无效消息真正的集群故障反而不容易被注意到。正则的精细化需要不断迭代。我建议上线初期设置一个“调试模式”把正则命中结果和日志原文记录下来连续观察几天不断修正规则确认稳定后再把调试模式关掉。不要指望第一版正则就完美这不现实。5. 警报通知的多种姿势邮件、IM机器人、本地弹窗规则引擎匹配到异常日志后接下来就是通知模块的活了。通知方式的选择直接决定了告警能不能被及时看到、会不会被刷屏反感。这一节我把邮件、企业微信/钉钉机器人、本地桌面通知三种实现方式都讲一遍你可以按需选择组合。5.1 邮件通知经典但不该是唯一渠道用smtplib发邮件是最经典的通知方式好处是部署简单、几乎所有场景都支持而且邮件本身就是正式沟通渠道适合做记录留痕。但它的致命弱点是时效性差。一封告警邮件发出去值班同学不一定会立刻看见尤其是深夜或者在忙别的事情时。所以我把邮件定义为“事后追溯”的告警渠道而不是“第一响应”的渠道。核心系统故障时优先靠IM机器人和短信把人叫醒邮件只是补充。邮件通知的关键代码import smtplib from email.mime.text import MIMEText from email.header import Header def send_email(config, subject, content): msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] config[username] msg[To] ,.join(config[to]) try: if config.get(smtp_port) 465: server smtplib.SMTP_SSL(config[smtp_server], config[smtp_port]) else: server smtplib.SMTP(config[smtp_server], config[smtp_port]) server.starttls() server.login(config[username], config[password]) server.sendmail( config[username], config[to], msg.as_string(), ) server.quit() except Exception as e: # 发邮件失败不能导致主流程崩溃要记录日志 logger.error(f邮件发送失败: {e})两个容易出问题的点一是很多企业邮箱的SMTP端口在25、465、587之间需要对齐465用SSL587用STARTTLS25通常明文。我在代码里做了一个简单判断实际使用时要根据你的邮件服务商来配置。二是邮箱的授权码问题很多邮箱用密码登不上SMTP必须用客户端专用授权码。5.2 IM机器人群消息才是最适合值班的告警方式国内业务团队基本都在用企业微信或钉钉这两者都提供了群机器人Webhook接口发送告警消息到群里。这也是我实际项目里用的主力通知渠道。企业微信机器人的用法是在群里添加一个自定义机器人得到一个Webhook URL用POST方式向这个URL发送一个JSON消息体机器人就会在群里说话。具体实现如下import requests import json def send_webhook(url, content): headers {Content-Type: application/json} payload { msgtype: text, text: { content: content, }, } try: resp requests.post(url, jsonpayload, headersheaders, timeout5) resp.raise_for_status() except Exception as e: logger.error(fWebhook发送失败: {e})企业微信机器人还支持markdown类型的消息体可以做得更精致一点比如用不同颜色标注告警级别、添加超链接跳转到日志平台。钉钉机器人的消息格式略有不同但思路一样。我踩过的一个坑是Webhook URL里有一个key参数这个参数必须被完整保留否则机器人会返回invalid webhook。如果配置在YAML文件里记得给URL加上引号避免特殊字符被解析掉。另一个坑是限流问题。企业微信机器人有频率限制每分钟最多20条。告警风暴时你一秒发几十条消息会触发限流后面的全被丢弃。所以通知模块里我加了一个简单的队列和节流机制import queue import threading import time class ThrottledNotifier: 带节流控制的通知器防止触发IM平台限流 def __init__(self, min_interval3): self.min_interval min_interval self.last_send_time 0 self.lock threading.Lock() self.pending [] def send(self, message): with self.lock: self.pending.append(message) now time.time() if now - self.last_send_time self.min_interval: self._flush() def _flush(self): if not self.pending: return messages self.pending.copy() self.pending.clear() # 多个消息合并成一条发送降低触发频率 combined \n.join(messages) send_webhook(WEBHOOK_URL, combined) self.last_send_time time.time()这个设计的核心是把短时间内多条告警合并成一条消息发送既避免限流也减少群里刷屏。合并消息对值班同学来说也更友好他们只需要看一条汇总而不是把几十条单条消息看完。5.3 本地通知开发调试时的好帮手本地桌面通知适合在开发调试时使用。你把脚本集成到自己的开发环境里日志一有异常就弹一个系统通知立刻看到效果。跨平台的方案是用plyer库from plyer import notification def send_local_notification(title, message): notification.notify( titletitle, messagemessage, timeout10, )这个方案在Windows和macOS上都能用但要注意它依赖系统通知服务在一些精简版Linux桌面环境上可能不工作。服务器上是没有桌面的所以本地通知只用于开发机调试不要部署到生产服务器。5.4 多渠道怎么协同主备关系和降级策略通知模块的真正难点不在于“能发”而在于“发不出去的时候怎么办”。我部署的监控脚本把通知模块做成了主备模式优先IM机器人如果IM发送失败比如网络抖动、机器人被移出群自动降级发邮件如果邮件也失败就把告警写入本地文件。这个降级策略看起来简单但保证了告警不丢失。写入本地文件这个兜底方案虽然不能主动触达值班同学但至少保留了证据事后排查时可以回溯。如果你对告警即时性要求特别高还可以再加上短信或电话告警的接口但需要额外接入短信服务商的API一般场景用不上。6. 防止告警轰炸去重、聚合与升级机制不能少任何告警系统在上线运行一段时间后都会遇到同一个问题——告警疲劳。第一个月还认真看第三个月基本就“又是它不用管”。这个问题的根源通常不是值班同学态度有问题而是告警系统没有做好去重、聚合和升级。6.1 相同告警合并只报一次但持续提醒急性告警场景里同一个异常可能在一分钟内出现几十次。如果每次都要发群消息瞬间爆炸。我的做法是引入“窗口期”概念同一规则或者按规则加内容指纹在窗口期内最多只发一次。实现方式很直接用一个字典记录每个规则的最近告警时间class Deduplicator: def __init__(self, window_seconds300): self.window_seconds window_seconds self.last_alert_time {} def should_alert(self, rule_name): now time.time() last self.last_alert_time.get(rule_name, 0) if now - last self.window_seconds: self.last_alert_time[rule_name] now return True return False但“只报一次”也有问题如果这个异常持续了30分钟值班同学看完第一次告警后可能就去处理了30分钟内没有再收到提醒直到问题升级成事故。所以更合理的方案是在窗口期内不重复发但窗口期结束后如果异常还在再发一次并且告警内容里附上“从第一次出现到现在持续了多久”的信息。这会大幅提升值班效率。6.2 相似告警聚合把“一堆碎消息”变成“一条汇总”比去重更进一步的是聚合。假设Nginx的上游服务挂了error.log里每分钟会产生“connect() failed”和“worker_connections are not enough”两种不同类型的错误。如果分开发就是两条告警。但如果在一个时间窗口内这两种错误都指向同一个根因上游服务宕机导致连接堆积那它们应该合并成一条包含两个维度的告警消息。聚合的一个简单实现是在固定时间窗口比如60秒内把所有命中的告警规则和它们出现的次数统计出来窗口结束时统一发送一次。class Aggregator: def __init__(self, window_seconds60): self.window_seconds window_seconds self.events {} self.lock threading.Lock() def add_event(self, rule, line): with self.lock: if rule not in self.events: self.events[rule] {count: 0, sample_lines: [], first_seen: time.time()} self.events[rule][count] 1 if len(self.events[rule][sample_lines]) 3: self.events[rule][sample_lines].append(line) def flush(self): with self.lock: events self.events self.events {} return events聚合后的告警内容大致长这样[汇总告警] 60秒内共捕获3类异常总计42次 1. upstream_connection_refused | 发生次数: 30 示例: 2026/05/14 10:00:03 [error] connect() failed (111: Connection refused) 2. worker_connections_not_enough | 发生次数: 10 示例: ... 3. upstream_response_timeout | 发生次数: 2 示例: ...值班同学看到这样一条汇总比看到42条零散消息要舒服得多。而且汇总里给了示例日志可以直接去日志平台搜索定位。6.3 告警升级一直没人处理就一直升级去重和聚合解决的是“告警太多”的问题告警升级解决的是“告警被忽略”的问题。我把告警升级分成两个维度第一个维度是“时间升级”一条critical级别的告警发出后如果30分钟内没有被确认ack升级为更高优先级通知同时抄送团队leader。第二个维度是“重复升级”同一规则在1小时内多次触发说明问题没有被解决自动提高告警级别。实现时间升级需要一个“告警确认”机制。你可以在通知消息里附带一个确认链接比如点击后调用Webhook标记已确认但这个会让系统复杂度上一个台阶。如果你不想做得太重退而求其次的方案是定期扫描“未处理告警列表”对超过指定时间的告警发一条“您有一条告警尚未确认”的提醒。这个逻辑用定时任务实现就行不需要引入复杂的状态机。我在实际项目里做了一版简化实现告警按5分钟窗口聚合并发送如果某条规则在连续3个窗口里都命中则触发“持续告警”通知内容明确说明“已持续15分钟未恢复”。这样不用值班同学手动确认系统也能起到升级提醒的作用。6.4 静默与白名单有些告警应该允许被“关掉”体系设计得再完善也总有需要临时调整的时刻。比如某个第三方接口在做计划内变更会短暂出现一批报错日志或者某个已知问题临时无法解决会周期性告警。这时候如果没有“静默”机制告警轰炸只会让人对系统失去信任。我的做法是维护一个静默规则列表支持两种匹配方式按规则名静默、按日志内容关键词静默。{ silent_rules: [ {rule_name: third_party_api_timeout, expire_at: 2026-05-20 18:00:00}, ], silent_keywords: [ {keyword: planned maintenance, expire_at: 2026-05-15 23:59:59}, ], }静默规则必须设置过期时间不允许永久静默。忘记关闭静默是运维事故的常见诱因设置过期时间能强制你重新审视这条规则到底还需不需要。7. 工程化落地从脚本到可维护的小系统前面六节的内容集中在“核心逻辑”上但一个脚本从“我本地能跑”变成“生产环境稳定运行”中间还差着很多工程化的功夫。这一节把部署上线过程中必须考虑的几个问题讲完。7.1 配置管理把“改代码”变成“改配置”第一版脚本我把规则、邮箱账号、Webhook地址全部硬编码在代码里。结果就是每次调整规则都要改代码、重启进程非常痛苦而且容易改错。后来我把配置全部抽离到config.yaml代码逻辑完全不感知配置细节。规则变更、通知渠道变更、静默策略调整全部通过改配置文件完成不需要动代码。# config.yaml monitor: log_path: /var/log/nginx/error.log log_dir: /var/log/nginx rules: - name: upstream_connection_refused pattern: connect\(\) failed.*Connection refused exclude_pattern: level: critical description: 上游服务连接被拒绝服务可能已宕机 alert_window_seconds: 300 - name: worker_connections_not_enough pattern: worker_connections are not enough exclude_pattern: level: warning description: Nginx worker连接数不足 alert_window_seconds: 600 notification: email: smtp_server: smtp.example.com smtp_port: 465 username: monitorexample.com password: your_password to: [opsexample.com] webhook: url: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx silent: - rule_name: third_party_api_timeout expire_at: 2026-05-20 18:00:00用YAML管理配置有个隐藏好处配置文件本身就是规则文档。以后新同学接手这个监控系统翻一翻配置文件就知道目前监控了哪些日志、关注哪些异常、告警发到哪完全不需要去代码里猜。7.2 运行状态可视化一个简单的状态文件脚本运行得好不好不能靠“感觉”。我用一个状态文件记录最近一次启动时间、最近一次读到日志偏移量、最近一次命中规则、最近一次成功发送通知。脚本每处理完一批日志就更新一次状态文件。{ start_time: 2026-05-14 10:00:00, last_read_position: 123456789, last_match_time: 2026-05-14 10:05:30, last_match_rule: upstream_connection_refused, last_notify_time: 2026-05-14 10:05:31, last_notify_success: true, total_lines_processed: 12345, }这个状态文件的核心价值在于当你怀疑监控系统已经“静默失效”时看一眼状态文件里last_read_position是否持续增长就能判断脚本到底是在正常消费日志还是已经卡住。我自己排查问题时这个状态文件帮了大忙。7.3 systemd守护别用nohup后台运行真的把Python脚本部署到Linux服务器上最忌讳直接用nohup python monitor.py 跑。这种方式进程管理能力为零进程挂了没人知道、开机不会自动启动、日志管理混乱。正确做法是写一个systemd service文件[Unit] DescriptionPython Log Monitor Service Afternetwork.target [Service] Typesimple WorkingDirectory/opt/log_monitor ExecStart/usr/bin/python3 /opt/log_monitor/monitor.py Restartalways RestartSec5 Userroot [Install] WantedBymulti-user.target配置好后sudo cp log_monitor.service /etc/systemd/system/ sudo systemctl daemon-reload sudo systemctl enable log_monitor sudo systemctl start log_monitorsystemd会负责守护进程异常退出时自动重启Restartalways服务器重启后自动拉起enable。日常维护只需要systemctl status log_monitor、systemctl restart log_monitor就够了。这一步带来的稳定性提升是任何代码层面的优化都比不了的。7.4 自监控监控脚本自己也会出问题聊一个运维圈经常提到的问题谁来监控监控系统我已经在项目里踩过这个坑——监控脚本因为未捕获的异常挂掉后我整整两天没收到任何告警最后还是看业务同学反馈才发现监控已经失效。从那以后我给自己定了两条规矩。第一条主循环里任何单条日志处理异常都不能导致整个进程退出。异常捕获到并记录到日志后继续处理下一条不能因为一条异常日志就让整个监控停摆。第二条写一个简单的定时任务比如cron每5分钟跑一次检查监控进程是否存活如果不存活就发一条告警#!/bin/bash # /opt/log_monitor/check_monitor.sh if ! systemctl is-active --quiet log_monitor; then curl -s -X POST https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxx \ -H Content-Type: application/json \ -d {msgtype:text,text:{content:【监控告警】日志监控服务未运行请立即检查}} /dev/null fi把这个脚本加进crontab每5分钟执行一次。这样即使监控脚本挂了告警链路也不会断。7.5 性能实测与调优这个脚本能扛多大流量最后说说性能表现。我部署的这套脚本实际监控三个日志文件峰值时的日志写入速度大约每秒2000行。整个脚本的CPU占用率稳定在3%以内内存占用约30MB。对于常规的中小规模应用这个指标完全够用。但如果你的日志量达到每秒几万行甚至百万行级别Python脚本就力不从心了。这时需要换链路用Filebeat做日志采集、Kafka做消息队列、Elasticsearch做存储、Kibana做可视化再用ElastAlert做告警规则。整个技术栈就复杂多了那已经不属于“脚本”的范畴而是一个完整的日志系统。8. 给这段实战经验补个尾巴几个容易被忽视的小细节项目做到这里整个监控系统已经能稳定运行了。但每次回顾这个项目总还能想起一些当时容易被忽略的小细节。这些细节不写进正式文档却实实在在影响使用体验和维护成本。第一个细节是日志时间戳的时区问题。日志文件里的时间往往是服务器本地时间而告警消息发出的时间是你脚本所在机器的时间。如果两者时区不一致排查问题时会出现“告警说10点05分发生日志里却是9点05分”的混乱。我在加载配置时加上了一个timezone参数发告警时把时间统一转换能少掉很多沟通成本。更简单的方法是直接在告警内容里同时附上服务器时间和监控端时间让值班同学自己对照。第二个细节是告警消息里要附上“日志文件的完整路径”加上“日志行号”。这条在排查时非常有用——你直接复制路径到服务器上看文件就行不用再去猜是哪台机器哪个文件。第三个细节是配置文件里的密码和Webhook密钥。这些敏感信息直接写在YAML里如果代码仓库是共享的等于把钥匙交给了所有人。更安全的方式是用环境变量或者专门的密钥管理工具比如Vault代码运行的时候动态获取。如果项目复杂度不高至少也要确保配置文件不要提交到git仓库里。第四个细节是日志监控脚本本身运行时的日志。很多人搭好了监控却忘了监控脚本自己也会产生日志我用的是log_monitor.log。不要小看这个日志文件等你排查“为什么没告警”的时候它就是第一手线索。日志文件也要注意定期清理避免长时间运行后把磁盘塞满。最简单的方式是用crontab定期执行logrotate或者干脆用系统的logrotate配置管理它。最后一个建议如果你第一次做这类日志监控项目不要一上来就贪大求全。先选一个日志文件、配三条规则、用一个通知渠道跑通全流程体验一遍“日志写入 - 脚本捕获 - 匹配规则 - 发出告警”的完整链路。确认一切正常后再逐步扩展规则、增加被监控的日志文件池、接入更多通知渠道。一口吃不成胖子监控系统的迭代和线上系统的迭代一样都是“小步快跑”最稳妥。我这个项目从第一行代码到稳定运行大概用了一周多中间大部分时间不是在写代码而是在打磨规则、调试边界条件、处理各种“意外”。但这一遍下来后续所有类似需求我都有了完整的解决框架和避坑清单这个积累的收益比脚本本身大得多。如果你也要在项目里做日志监控希望这篇文字能帮你少走几段弯路。