凌晨两点电话把我叫醒。线上支付回调超时接口日志疯狂刷着Timeout exceeded但监控看板上一切指标都正常——CPU、内存、磁盘都平稳得很用户却已经下单失败了。问题出在那套监控平台看的是机器指标根本没看日志内容。第二天我写了个二十来行的 Python 脚本只要日志里出现Timeout exceeded就立刻推送群机器人这类事故从此再也没过夜。“用 Python 监控系统日志并发送警报”听起来是个很基础的需求但真做起来里面藏着的细节比你想象的多日志文件怎么稳定追踪、轮转了怎么处理、正则怎么写不漏报、警报怎么发才不烦人、怎么去重和限流。这篇文章把我实际用来盯生产环境的一套方案完整拆开讲包含可以直接拿去改的 Python 代码、配置文件和部署说明。不管你是在 Linux 服务器上盯业务日志还是想给某个应用的报错做关键词报警这套东西都能直接落地。1. 为什么我最终选择用 Python 做日志监控警报先说个现实情况现在开箱即用的监控平台很多Prometheus、Grafana、夜莺、Zabbix每一个拿出来都很成熟。那为什么还要自己用 Python 写因为指标监控和日志监控是两码事。Prometheus 擅长抓取 CPU、内存、QPS 这类数值指标但它不会去读你的应用日志更不会在你业务代码的except Exception分支里出现特定报错时给你发消息。ELK 这类方案能收日志也能做告警但它需要一整套 Elasticsearch、Logstash、Kibana 的集群小团队单个项目为了几条告警规则去维护这么一套东西成本明显不划算。1.1 现有监控平台到底缺什么拿我自己的使用体验来说当时公司有两套监控一套 Prometheus 拉取指标一套 Grafana 画看板。线上服务挂了进程退出这两套东西能立刻发现但如果服务没挂只是在某些请求下疯狂抛KeyError机器指标完全无感用户被卡住半天你都不知道。日志监控的价值恰好在这里它是离业务最近的一层信号能反映出代码层面的异常而这些异常往往早于资源耗尽。我用过一段时间 ELK 来做日志告警最后放弃了原因有三个。第一维护成本实在高Elasticsearch 集群需要单独规划磁盘和内存为了几条规则养一整个集群怎么看都不划算。第二告警规则配置不直观Kibana 里配阈值、配时间窗口门槛比写正则高一些团队成员要专门培训。第三当你想实现“某个用户 ID 的错误重复出现 3 次才报警”这种逻辑时通用平台的配置项很难表达得绕一大圈。换成 Python 脚本就不一样了这种规则本质上是几十行代码的事。下面这个表格可以比较直观地看出几种方案的差异方案擅长领域日志内容级告警维护成本适合规模Prometheus Grafana机器指标监控弱中大规模ELK 全家桶日志集中检索中高大规模Shell crontab简单任务弱低一次性场景Python 脚本定制化日志告警强极低中小规模1.2 最适合脚本监控的几类场景并不是所有场景都得用脚本但下面这几类我实测下来用 Python 最顺手。第一类是错误关键词监控。比如应用日志里出现Traceback、ERROR、FATAL、Timeout、Connection refused这类关键字直接报警。这是最普遍的诉求通常在日志里 grep 一下能看到的错误都可以转成脚本规则。第二类是安全类日志。比如 SSH 登录失败次数突然上涨这往往是有人尝试爆破机器或者业务系统里某个账号连续多次登录失败这种安全事件需要及时发现。第三类是业务规则类。例如支付回调连续失败、订单状态机出现异常跳转、任务队列积压但没有报错——这些问题不体现在通用错误日志中但在业务日志里有明显的模式可以匹配。第四类是定时任务的结果确认。每天跑的 cron 任务成功和失败都会写日志脚本可以只盯失败行失败就发警报比每天起床翻日志靠谱。2. 方案设计一条从日志到警报的完整流水线用 Python 监控日志本质上是在搭一条流水线日志文件被写入新内容脚本读取后做解析和匹配命中的行进入告警判断最后通过邮件或 Webhook 推出去。这个链条并不复杂但每个环节都有几个值得注意的设计点。2.1 完整链路拆解整条链路分成六个环节读取、解析、过滤、告警、抑制、记录。前三个是核心后三个决定这套系统是否真的能在生产环境活下去。读取环节负责从日志文件尾部增量读取新行而不是每次扫全量文件。这借鉴了 Linux 下tail -f的思路也是整个脚本性能和准确性的基础。解析环节把一行文本转成结构化内容至少要能区分出时间、级别、消息正文。过滤环节用正则或者关键词判断这一行是否命中了告警规则这在代码里通常表现为几个re.search条件。告警环节决定以什么方式通知人邮件、钉钉群机器人、企业微信机器人、Slack 都可以但生产环境我推荐用群机器人关键信息直接推给所有人。抑制环节是很多人忽略的它解决的是“同一个错误别五分钟推十条”的问题通常做法是记录最近一次告警的时间短时间内同类告警只发一条。记录环节把日志处理进度和已存在的告警状态持久化这样脚本重启后不会重复告警也不会把启动前的老日志误报一遍。这六个环节看起来多但每个都很简单。真正决定一套监控脚本好不好的是抑制和状态记录做没做到位而不是正则写得漂不漂亮。2.2 为什么不是 Shell 加 crontab很多人第一反应是写个 Shell 脚本用grep扫日志再配合 crontab 定时跑。这个方案我试过在日志量小的时候确实能凑合但日志一大就露馅。最典型的问题是全量扫描crontab 最短只能每分钟跑一次每分钟grep -E ERROR /var/log/app.log一次几百 MB 的日志每次扫好几秒高峰期直接把磁盘 IO 打满。而且 grep 扫一遍之后你怎么知道哪些是新出现的错误要么记录上次扫描的行号要么把上次的结果存下来做 diff这两种逻辑在 Shell 里写起来都很别扭。另一个问题是告警逻辑复杂了以后Shell 脚本会变成一坨无法维护的代码。比如要求“在一分钟内同一个错误出现超过 5 次才告警”“某些关键词在周末不告警”“告警前先调一个接口查询用户是否存在”这些逻辑用 Shell 写基本是灾难用 Python 写就是普通的函数调用。Python 在这个场景下的优势很明确文件处理能力足够强正则表达式是标准库又自带了smtplib、urllib这种做告警通知需要的模块几乎没有第三方依赖。开发效率比 Go 高运行效率比 Shell 高对一个监控脚本来说这是最舒服的平衡点。如果你要同时监控几十个文件且单文件每秒产生上千条日志那确实可以考虑 Go 或 Rust但大多数中小项目的日志体量Python 跑起来毫无压力。2.3 理解日志文件的三种常见形态写读取器之前先得搞清楚日志文件有哪些常见形态因为不同形态对应不同的读取策略。由于最常见的就是纯文本行式日志。每一行代表一条记录比如2025-01-15 10:23:45 ERROR 支付回调超时, order_id10086 2025-01-15 10:23:46 INFO 支付回调重试, order_id10086这类日志解析最简单按行读取正则匹配即可。第二种是 JSON 结构化日志。现在很多应用框架直接把日志输出成 JSON一行就是一个对象里面有timestamp、level、message、context字段。解析时用json.loads把每一行转成字典再按字段匹配比正则更可靠。因为 JSON 字段不会因为日志内容里的特殊符号而误匹配。第三种是需要特别小心轮转log rotation的日志。系统日志和绝大多数应用日志都会配置按大小或按天轮转。轮转方式通常有 rename 和 copytruncate 两种。rename 是把当前日志改名为带日期后缀的文件然后新建一个空文件继续写入logrotate默认行为就是这样copytruncate 是先把当前文件复制一份然后清空原文件。脚本如果只是简单地打开一个文件读到最后遇到轮转就会出问题要么读不到新日志要么把老文件重复读一遍。后面的代码部分会专门处理这个问题。3. 核心代码实现读、判、报三段式下面进入正题。我要给的这套代码经过了生产环境验证核心思路是三个独立模块日志读取器、规则引擎、告警发送器外加一个主控制流程。每一段我都会讲清楚它解决的问题和要注意的坑。3.1 日志读取器稳定追踪增量和日志轮转一个可靠的日志读取器至少要解决三个问题从文件尾部开始读、每次只读新增内容、轮转后自动切换文件。先看基础版本import os import time class LogFollower: def __init__(self, filepath, start_from_endTrue): self.filepath filepath self.fp None self.inode None self._open(start_from_end) def _open(self, start_from_endTrue): self.fp open(self.filepath, rb) st os.fstat(self.fp.fileno()) self.inode st.st_ino if start_from_end: # 跳到文件末尾不读历史内容 self.fp.seek(st.st_size) else: # 从文件开头开始读适用于首次启动想全量扫描 self.fp.seek(0) def read_new_lines(self): lines [] # 先检查文件是否发生了轮转 st os.stat(self.filepath) if st.st_ino ! self.inode: # 文件被 rename 了重新打开新文件 self.fp.close() self._open(start_from_endTrue) while True: line self.fp.readline() if line: lines.append(line) else: break return lines这里有两个关键点。第一文件始终以rb二进制模式打开。原因是日志文件编码可能不是 UTF-8如果直接用文本模式打开遇到非法字节会抛异常二进制模式先把字节读回来后面再由解析层做解码更稳。第二用st_ino判断文件是否被轮转。当logrotate把当前日志改名后原打开的文件句柄对应的 inode 还在但路径已经指向一个新文件。所以每次读之前都os.stat(self.filepath)发现 inode 变了就关闭旧句柄重新打开新文件。有些系统上logrotate会配置copytruncate这种情况下路径没有变文件也没有新 inode只是文件大小被清零了此时readline从当前指针继续读会自然地走到 EOF 去等待逻辑上不会出错所以不需要额外处理。这个读取器在实际运行中还有个问题如果脚本启动前文件已经很大而你希望它只监控启动之后的日志那么start_from_endTrue是合理的。但如果你想在脚本首次启动时把已有日志里的错误全部扫描出来并告警就需要start_from_endFalse。我建议把这个参数做成配置项部署的时候根据需求设置。3.2 规则引擎正则匹配与告警阈值读取器拿到的是二进制字节规则引擎负责把它变成告警决策。我习惯先把字节解成字符串再逐条规则跑正则。import re class Rule: def __init__(self, name, pattern, levelerror, suppress_seconds300): self.name name self.pattern re.compile(pattern) self.level level self.suppress_seconds suppress_seconds self.last_alert_time 0 def match(self, text): return self.pattern.search(text) is not None这个类本身不复杂但它背后有一个设计决策值得说明每个规则独立维护自己的last_alert_time。好处是不同规则的告警互不干扰ERROR关键词五分钟内只报一次不影响Connection refused一分钟内报一次。如果用一个共享的去重表就会因为 A 规则刚发过告警而导致 B 规则被误抑制。规则匹配的正文建议统一走一层解析。如果是 JSON 日志先把json.loads解成字典再取message字段和规则匹配如果是普通文本就整行匹配。这层解析的好处是避免匹配到时间戳或节点名这种干扰信息。比如你想匹配“订单失败”但日志里有个字段叫order_statusFAILED如果不对消息体做抽取正则很容易误报。正则本身是另一个容易出问题的地方。我建议规则文件里写简单、明确的关键词用re.escape处理包含特殊符号的内容。例如要匹配包含order_id10086的错误直接写re.escape(order_id10086)可以避免和其他正则语法符号带来意外。如果你不确定自己的正则写没写对可以在脚本里加一个--test参数指定一条日志行测试所有规则这是最省事的验证方式。触发阈值也是生产环境刚需。有时候单个错误行本身不值得告警但在一分钟内连续出现多次说明系统确实在出问题。可以给规则加一个计数窗口class ThresholdRule(Rule): def __init__(self, name, pattern, threshold5, window_seconds60, *args, **kwargs): super().__init__(name, pattern, *args, **kwargs) self.threshold threshold self.window_seconds window_seconds self.hits [] def is_triggered(self): now time.time() self.hits [t for t in self.hits if now - t self.window_seconds] return len(self.hits) self.threshold def record_hit(self): self.hits.append(time.time())这个ThresholdRule的实现是记命中的时间戳超过窗口的命中会被踢掉。如果一分钟内命中次数达到阈值就触发告警。实际使用中我一般把阈值规则用在安全告警场景比如 SSH 登录失败、验证码错误这类短时间内密集出现的事件。3.3 警报发送邮件、Webhook 与标准输出警报发送是整个链路里最容易敷衍、也最容易出问题的一块。很多人的第一版脚本都是在终端 print 一下就算了这确实能用来调试规则但要真正发挥作用必须把消息送到人能看到的地方。我在这套方案里实现了三种发送方式标准输出、SMTP 邮件、Webhook。标准输出用于开发调试和本地跑测试邮件适合正式、低频的告警Webhook 适合需要所有人实时感知的场景比如钉钉群机器人或者企业微信机器人。先看 SMTP 邮件import smtplib from email.mime.text import MIMEText from email.header import Header def send_mail(alert, smtp_config): msg MIMEText(alert.full_message, plain, utf-8) msg[Subject] Header(f[监控] {alert.rule_name}, utf-8) msg[From] smtp_config[from_addr] msg[To] smtp_config[to_addr] server smtplib.SMTP(smtp_config[host], smtp_config[port], timeout10) if smtp_config.get(use_tls): server.starttls() if smtp_config.get(username): server.login(smtp_config[username], smtp_config[password]) server.sendmail(smtp_config[from_addr], smtp_config[to_addr].split(,), msg.as_string()) server.quit()这封信的标题里带规则名正文里带命中的日志原文对接收方来说足够定位了。注意一个细节收件人地址用split(,)切分方便配置多个收件人用英文逗号分隔就行。Webhook 发送更简单本质上就是一个 HTTP POSTimport json import urllib.request def send_webhook(alert, webhook_url): payload { msgtype: text, text: { content: f[监控] {alert.rule_name}\n{alert.hit_line} } } data json.dumps(payload).encode(utf-8) req urllib.request.Request(webhook_url, datadata, headers{Content-Type: application/json}) with urllib.request.urlopen(req, timeout10) as resp: resp.read()这个 payload 结构符合钉钉群机器人自建应用的格式。企业微信机器人的结构略有差异但只需要把msgtype改成text并调整外层字段名即可。实际接入时打开群设置找到机器人 Webhook 地址填进配置文件就能用。需要注意Webhook 地址属于敏感信息一旦泄露别人就能往你的群里发消息建议存在环境变量或单独配置文件中不要提交到代码仓库。我自己的生产环境选的是群机器人而不是邮件原因很朴素邮件太容易被淹没。群机器人会把告警直接推到值班群没一会儿就刷到最顶上。而且群机器人天然支持多人同时收到不用维护邮件组。3.4 主程序与状态持久化最后是把读取器、规则引擎、告警发送串起来的主程序。这个主程序必须是一个长期运行的守护进程而不是被 crontab 每分钟拉起的那种一次性任务。import time import json import os class MonitorApp: def __init__(self, config): self.config config self.follower LogFollower(config[log_path], start_from_endconfig.get(start_from_end, True)) self.rules [Rule(namer[name], patternr[pattern]) for r in config[rules]] self.state_file config.get(state_file, /tmp/log_monitor.state) def load_state(self): if os.path.exists(self.state_file): with open(self.state_file, r) as f: return json.load(f) return {} def save_state(self, state): tmp_file self.state_file .tmp with open(tmp_file, w) as f: json.dump(state, f) os.replace(tmp_file, self.state_file) def run(self): state self.load_state() while True: try: new_lines self.follower.read_new_lines() for raw_line in new_lines: text raw_line.decode(utf-8, errorsreplace) for rule in self.rules: if rule.match(text): now time.time() if now - rule.last_alert_time rule.suppress_seconds: alert Alert(rule_namerule.name, hit_linetext.strip()) send_webhook(alert, self.config[webhook_url]) rule.last_alert_time now time.sleep(1) except KeyboardInterrupt: break except Exception: time.sleep(5)主程序的逻辑很直白每秒钟读一次新增日志逐行匹配规则命中且不在抑制窗口内就发告警。为什么固定睡一秒而不是更短因为对大多数应用日志来说1 秒的延迟完全可接受而脚本能保持极低的 CPU 占用。如果你监控的是交易系统这种对实时性要求很高的场景可以改成 0.1 秒代价是略高的 CPU 消耗。状态持久化这块代码里我预留了load_state和save_state的框架实际用途是把读到的文件位置和告警时间存下来。这样脚本重启后可以从上次读到的位置继续而不是重新从末尾开始导致漏掉重启期间产生的日志。这个机制在常驻进程意外退出后非常有用务必补上。简单的做法是把当前follower.fp.tell()和每条规则的last_alert_time组装成字典每处理完一批日志就写一次。4. 部署、调试与常见问题排查代码本身跑起来不难但把它部署到生产环境并稳定运行还是会踩不少坑。我把环境准备和常见问题的排查方法完整过一遍。4.1 Linux 环境准备与依赖安装这套脚本基本只用 Python 标准库所以依赖安装几乎没有成本。但有一个前提Python 版本不能太老。建议 3.8 以上原因是用到了os.replace、f-string 以及类型注解等语法的完善版本。现在大多数 Linux 发行版的默认python3已经是 3.10 左右直接用就行。创建运行环境时我建议做一个独立的虚拟环境避免和系统 Python 环境混在一起。命令行操作如下cd /opt/log-monitor python3 -m venv venv source venv/bin/activate python script.py如果你的系统没有python3-venv先安装一下apt install python3-venv开发阶段如果想在 VS Code 里跑这个脚本直接在.venv里选解释器就行。生产环境我个人习惯用 root 以外的用户跑监控任务因为很多日志文件权限有限用太高的权限跑脚本本身也是一个不必要的风险。当然/var/log/auth.log这类文件普通用户没有读权限要么把用户加入adm组要么用sudo启动根据实际情况权衡。4.2 常见问题速查表我在实际运维中整理了一份问题速查表基本覆盖了新手最容易碰到的几种情况现象可能原因处理方式脚本运行但收不到告警正则没匹配上先用--test或grep手动确认关键词告警一条都没触发但手动 grep 能看到错误读取位置从末尾初始化检查start_from_end配置改成False试一次日志文件轮转后脚本不再读取未处理 inode 变化确认LogFollower.read_new_lines里检查了st_ino中文日志乱码日志非 UTF-8 编码解码时用errorsreplace再确认实际编码收到大量重复警报无抑制逻辑给规则加suppress_seconds告警发送超时导致脚本卡住Webhook/邮件服务响应慢给发送函数加 10 秒超时并做异常捕获脚本重启后把历史错误全部报一遍未持久化日志读取位置补上状态文件记录file.tell()这几种情况里最值得留意的就是正则匹配不到。日志里的字符可能包含不可见控制字符比如终端配色转义序列肉眼看不见但正则匹配会失败。碰到这种情况先repr(line)把字符串的转义形式打出来一目了然。4.3 我踩过的几个实战坑第一个坑是启动位置的选择。刚开始做的版本一定默认从文件末尾开始读这样部署当天如果日志文件特别大不会把历史错误全部扫出来刷屏。但后来有一次日志文件被轮转清空后脚本打开空文件等待期间产生的错误全部漏掉了。原因是脚本启动后才打开文件而文件在启动前已经被轮转过一次启动位置从 0 开始以为自己读的是尾部。把位置信息持久化之后这个坑才算填平。第二个坑是守护进程的管理。早期我用一个python monitor.py 直接丢后台结果进程莫名其妙退出后没有自动拉起日志监控空缺了两天都没人发现。后来我把脚本写成了 systemd 服务加上Restartalways进程一挂就自动拉起状态文件也能保证重启后继续上次的位置。如果你也在生产环境用强烈建议直接上 systemd。第三个坑是告警消息格式。一开始我把匹配到的整条日志原文推出去长日志动不动就好几百字符群消息根本看不完。后来调整成只推时间、错误摘要和关键字段看起来清爽很多。这跟日志解析的结构化程度直接相关解析层做得好告警消息就能做得精致。给一个 systemd unit 文件做参考[Unit] DescriptionLog Monitor Alert Service Afternetwork.target [Service] ExecStart/opt/log-monitor/venv/bin/python /opt/log-monitor/main.py Restartalways RestartSec5 Usermonitor [Install] WantedBymulti-user.target把这份配置放在/etc/systemd/system/log-monitor.service然后systemctl enable --now log-monitor就能开机自启并在崩溃时自动拉起了。5. 进阶扩展从单文件监控到小型监控系统如果你的日志监控需求不止一个文件这套脚本也有很自然的扩展路径不用推倒重来。5.1 多文件与目录监控最简单的扩展是把LogFollower实例放到一个字典里每个文件独立追踪自己的 inode 和位置。主循环里遍历所有 follower逐行跑规则。这样做的代价是每周期都要遍历多个文件但文件数量在十个以内时性能毫无问题。如果你监控的目录里文件会动态创建比如按天生成日志可以用watchdog库监听目录事件新文件出现时自动添加 follower。5.2 统计型告警与可视化指标脚本除了发告警也可以顺手把日志里统计到的指标暴露出去。比如记录每个错误关键词每小时的命中次数写入一个文本文件再让 Grafana 用一个简单的 text 面板读取。这虽然不是真正的 metrics 系统但对小型项目来说已经能解决“想看看趋势图但没精力搭 Prometheus”的需求。再进一步也可以把计数器通过/metricsHTTP 接口暴露直接接续 Prometheus 的 pull 模式这样一个脚本既覆盖了告警又补上了指标监控的缺口。5.3 热更新规则与免重启生产环境有一个很常见的烦恼改一条规则要重启服务。重启本身不可怕可怕的是重启瞬间主进程可能丢日志。我给脚本加了一个信号处理监听SIGHUP收到信号后重新读取配置文件并重建规则列表。这样改完配置文件后只需要kill -HUP pid就能生效不需要重启。import signal def reload_handler(signum, frame): # 重新加载配置文件重建 rules pass signal.signal(signal.SIGHUP, reload_handler)这个思路是从 Nginx 平滑重载日志文件那儿借鉴来的。进程不退出文件句柄不关闭自然不会有日志遗漏。唯一要注意的是配置文件必须校验通过后再覆盖现有规则否则写错了就会造成监控空窗。回到文章开头那个凌晨的电话。现在监控平台上依然显示一切指标正常CPU、内存看起来都没问题但 Python 脚本会在支付回调一超时就往群里丢一条告警顺带把订单号带上。处理问题的人从“扛着睡意翻日志”变成了“看一眼群消息就能定位”。这套方案我用了大半年越用越觉得它的价值不在代码本身而在可以让规则贴合自己的业务。不会发生“这个场景不好配算了不告警了”这种妥协。最后分享一个自己的习惯每次上线新规则前先开启调试模式把脚本输出打到标准输出跑半小时确认没有误报、没有漏报再切到正式的 Webhook。宁可多花半天验证也不要让告警轰炸把同事的信任消耗掉。日志里藏着的是最怕错过的那条错误脚本造出来就是为了必须看住它。