写这篇分享之前先说个真实背景。去年我在内网一台服务器上调一个反复崩溃的服务日志里疯狂刷“捕获到标准c异常”当时脑子里第一个想法是上一套监控平台但真动手的时候发现不对——就三台机器、一个业务、十来个日志文件为这事去维护一套完整监控体系成本明显不合理。后来我用Python写了一个不到两百行的日志监控脚本专门盯着日志文件匹配到错误关键字就发邮件和Webhook运行到现在一直很稳定。这就是这篇文章想聊的东西用Python监控系统日志并发送警报到底应该怎么做。我会从最核心的文件跟踪原理讲起到报警通道的接入再到部署成常驻服务、线上踩过的坑最后给一点往Prometheus/Grafana方向扩展的思路。适合刚入门Python想做个靠谱练手项目的人也适合小团队里需要快速解决“日志出问题能第一时间知道”的运维或者后端同学。1. 为什么我先问自己“一定要上Prometheus吗”很多朋友一听到“监控”两个字第一反应就是Prometheus加Grafana或者Zabbix、夜莺这一套。这个思路本身没错但得先分清场景。在决定用Python写脚本之前我整理了一次需求发现核心诉求其实很简单日志里出现特定错误关键字时能有人第一时间知道。这个诉求用监控平台当然能做到但有点杀鸡用牛刀的意思。1.1 轻量监控场景三台机器也要引一整套生态当时我的环境是这样的内网隔离机器不多没有专门的监控运维岗位日志主要散落在/var/log/myapp/和Windows应用日志目录下。如果上Prometheus我需要部署Prometheus服务端还要为每个日志文件写exporter配套Grafana面板再配置Alertmanager告警路由一整套下来至少两三个组件要长期维护。这对一个小团队来说不是“顺手装一下”的事而是给自己找了一个长期维护的负担。Python脚本正好卡在这个位置不需要额外守护进程不需要开放端口不需要时序数据库一个解释器加一个.py文件就够。它跟“捕获到标准c异常有关详细信息请参见系统日志”这类场景天然契合——典型的桌面软件或Windows服务会把异常细节写进应用日志你要做的只是盯住那个日志文件而不是为它建立一套庞大的监控基建设施。1.2 Python脚本、Prometheus、Zabbix的适用边界我做了个表格方便大家按自己的场景对号入座方案部署成本适合的场景不适合的场景Python日志监控脚本极低单文件即可少量机器、日志关键字报警、自定义逻辑复杂、需要快速上线大规模指标采集、历史趋势分析、多团队告警路由Prometheus Grafana中高多个组件指标采集、时序数据、可视化大盘、集群监控只盯日志关键字不想维护额外组件Zabbix中Agent式部署服务器/网络设备数量多、模板化监控、资产管理临时任务、复杂文本日志解析这里的核心判断标准是“数据形态”。日志监控本质上是事件检测而Prometheus本质是指标采集。如果你的需求就是“ERROR关键字出现就通知我”事件检测用Python脚本是更直接的路径如果你需要的是“过去一小时错误率的曲线”那才需要指标系统。两种不是替代关系而是不同层面的工具。2. 日志监控脚本的基本盘文件跟踪与关键字匹配脚本的核心功能拆开就两件事读取新增的日志内容、判断是否触发警报。第一步最关键因为日志文件不是数据库不会主动通知你有新数据你需要像tail -f那样持续跟踪文件增长。2.1 实现tail -fseek定位与阻塞读取tail -f的原理很多人知道是“持续读文件尾部”但落到代码上有一个细节必须处理第一次启动时你是从文件开头读还是从文件末尾读通常情况下应该从末尾开始读否则老日志会把告警刷爆。实现方式是用seek()把文件指针挪到最后import os import time class LogTail: def __init__(self, filepath, encodingutf-8): self.filepath filepath self.encoding encoding self.f open(filepath, rb) self.f.seek(0, os.SEEK_END) # 直接跳到文件末尾 def follow(self): while True: line self.f.readline() if line: yield line.decode(self.encoding, errorsreplace).rstrip(\n) else: time.sleep(0.5)这里我用二进制模式打开文件而不是直接open(filepath, r)原因后面会专门讲——Windows和Linux的日志编码不统一二进制读取再把解码权掌握在自己手里兼容性最好。follow()是一个生成器每次产生一行新日志。没有新日志时sleep(0.5)防止空转把CPU吃满。这个延迟可以根据日志量和告警时效需求调整需要秒级响应就sleep(0.2)日志量不大就sleep(1)对CPU占用几乎可以忽略。2.2 多关键字规则表从单行匹配到正则上下文匹配逻辑不建议写成一堆if ERROR in line的硬编码应该做成规则表。每条规则至少包含三个字段规则名、正则表达式、报警渠道。我用的规则结构是这样RULES [ { name: c_exception, pattern: r捕获到标准c\\异常|std::exception, channels: [email, wecom], }, { name: db_timeout, pattern: rDB_(TIMEOUT|CONNECTION_FAILED)|timeout waiting for connection, channels: [email], }, { name: oom_killer, pattern: rOutOfMemory|Killed process|oom-killer, channels: [email, wecom], }, ]匹配时用re.search而不是re.match因为错误关键字经常出现在一行日志的中间而不是行首。正则表达式建议写具体一点比如ERROR.*DB_TIMEOUT比单独ERROR误报率低得多这一点后面讲误报处理时还会再展开。规则表最直接的好处是以后要加一条新规则只需要改配置不需要动代码。我后来把RULES抽到了JSON配置文件里线上加规则不用发布脚本这是让这个小型工具具备可维护性的关键一步。2.3 多文件与动态日志目录用配置驱动而不是改代码单文件监控只能算Demo真正用起来要同时盯好几个日志文件。我的做法是维护一个watch_paths列表支持具体的日志文件路径也支持带*的目录通配WATCH_PATHS [ /var/log/myapp/app.log, /var/log/myapp/error/*.log, /data/services/gateway/logs/*.log, ]启动时用glob.glob展开通配符为每个文件创建一个LogTail实例再统一放入一个轮询循环。这里有一个容易踩的坑如果某个通配符暂时没匹配到任何文件比如error目录还是空的不要直接报错退出应该创建目录并定期重新扫描因为日志文件经常是程序运行时才创建的。def scan_log_files(): files set() for pattern in WATCH_PATHS: for f in glob.glob(pattern): if os.path.isfile(f): files.add(os.path.abspath(f)) return files扫描不需要太频繁每30秒或每60秒做一次就够了重点是保证新出现的日志文件能在1分钟内被纳入监控。3. 让警报真正“发出去”邮件与Webhook的落地细节日志跟到了规则也匹配了接下来是“发送警报”。这块我踩过的坑不比其他环节少尤其是限流、超时、配置方式这些细节写代码一小时调这些反而花了半天。3.1 SMTP邮件报警的配置与异常处理邮件报警用Python标准库就能搞定不需要装第三方包。核心是smtplib加email.message.EmailMessage。需要注意的点有几个现在主流邮箱都需要授权码而不是登录密码必须显式开启starttlssendmail调用要设置超时否则SMTP服务器无响应时脚本会一直卡住。import smtplib from email.message import EmailMessage SMTP_HOST smtp.qq.com SMTP_PORT 465 SMTP_USER monitorexample.com SMTP_PASS your_authorization_code # 用环境变量传入不要硬编码 MAIL_TO [opsexample.com, devexample.com] def send_email(subject, content): msg EmailMessage() msg[Subject] subject msg[From] SMTP_USER msg[To] , .join(MAIL_TO) msg.set_content(content) with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout10) as server: server.login(SMTP_USER, SMTP_PASS) server.send_message(msg)关键点是timeout10。没有这个参数一旦网络抖动发信函数可能阻塞几分钟而监控脚本正好是单线程轮询模型SMTP卡住会导致后面所有日志都处理不了。我后来甚至把发信动作放到独立线程里主循环只管记录“该报警了”发送由线程异步处理这样即使邮箱服务慢一点也不会拖垮主流程。3.2 企业微信/钉钉/飞书Webhook推送邮件适合正式通知但如果是夜间紧急情况微信或钉钉的机器人推送显然更及时。三种机器人的调用方式大同小异就是往Webhook地址POST一条JSON。以企业微信群机器人为例import requests WECOM_WEBHOOK https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key def send_wecom(content): payload { msgtype: text, text: { content: content } } requests.post(WECOM_WEBHOOK, jsonpayload, timeout5)注意每个渠道对消息格式有细微要求钉钉自定义机器人要求请求头Content-Type: application/json; charsetutf-8而且安全设置里要么加关键词、要么加签否则消息根本发不出去飞书机器人则是用msg_type字段而不是msgtype。这些差异不复杂但对第一次接入的人来说最容易踩的就是“明明POST成功了群里却没消息”九成情况是安全设置没配对。3.3 告警去重与冷却别让凌晨三点手机被刷爆这是整套脚本里最“值得一写”的经验。如果不做去重日志里一旦出现循环报错比如某条异常每秒钟刷一次你的手机会在凌晨三点被几十上百条警报震到没电。我做了一个基于冷却期的去重机制同一条规则匹配成功后记录最近一次通知时间冷却期内即使再次匹配到也只更新计数不重复发送。last_notify_time {} alert_count {} def notify_with_cooldown(rule_name, message, cooldown300): now time.time() last last_notify_time.get(rule_name, 0) if now - last cooldown: alert_count[rule_name] alert_count.get(rule_name, 0) 1 return # 发送前先把上次冷却期内的聚合次数带上 if alert_count.get(rule_name): message f\n[冷却期内同类告警次数: {alert_count[rule_name]}] send_wecom(message) send_email(f[告警] {rule_name}, message) last_notify_time[rule_name] now alert_count[rule_name] 0冷却期一般设置为5分钟比较合理既不至于漏掉持续报错又不会被刷屏。这条经验放到很多告警系统里都通用告警不在于多而在于每条都有价值真正做到让值班的人不麻木。4. 从“前台脚本”到“常驻服务”systemd、supervisor与Docker脚本写完之后最大问题是“怎么让它一直跑”。直接python3 monitor.py放前台终端一关就没了用nohup放后台进程万一崩了也没人拉起来。要让它像服务一样稳定运行我试了三种方式最后实际生产用的是systemd。4.1 systemd Unit开机自启与崩溃自动拉起在Linux服务器上systemd是首选。写一个单元文件放到/etc/systemd/system/log-monitor.service[Unit] DescriptionLog Monitor Service Afternetwork.target [Service] Typesimple Userroot WorkingDirectory/opt/log-monitor EnvironmentFile/etc/log-monitor/env ExecStart/usr/bin/python3 /opt/log-monitor/monitor.py Restartalways RestartSec5 StandardOutputappend:/var/log/log-monitor/run.log StandardErrorappend:/var/log/log-monitor/run.log [Install] WantedBymulti-user.targetEnvironmentFile用来加载邮箱密码、Webhook地址这些敏感配置不写死在代码里。Restartalways很关键进程因为任何原因退出systemd都会在5秒后重新拉起。我之前遇到过脚本因为一次未捕获的网络异常挂掉没有这个参数就彻底失联了。启动和开机自启命令sudo systemctl daemon-reload sudo systemctl enable log-monitor sudo systemctl start log-monitor查看运行状态和日志sudo systemctl status log-monitor sudo journalctl -u log-monitor -f4.2 supervisor和Docker两条替代路径有些发行版默认不用systemd或者你已经装了supervisor那也可以用supervisor管理。配置很简单放在/etc/supervisor/conf.d/log-monitor.conf[program:log-monitor] command/usr/bin/python3 /opt/log-monitor/monitor.py directory/opt/log-monitor autorestarttrue redirect_stderrtrue stdout_logfile/var/log/log-monitor/supervisor.log stderr_logfile/var/log/log-monitor/supervisor.log environmentPYTHONUNBUFFERED1environmentPYTHONUNBUFFERED1是Python日志实时落盘的关键不加这个print输出会被缓冲supervisor日志里的内容会延迟好一会儿。如果你对Python环境隔离有要求也可以用Docker方式。Docker的优势是环境干净、不污染宿主机但要注意挂载宿主机日志目录进去否则容器里啥也看不到。可以用python:3.11-slim作为基础镜像FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY monitor.py . CMD [python, monitor.py]运行命令加上--restart unless-stopped保证宿主机重启后容器自动起来日志目录用-v挂载docker run -d --name log-monitor \ --restart unless-stopped \ -v /var/log/myapp:/var/log/myapp:ro \ -v /opt/log-monitor/config:/app/config \ log-monitor:latest我个人建议优先systemd少一层容器也能少一些日志挂载的坑。Docker适合本身就跑在容器编排里的团队能统一管理镜像和配置。4.3 监控脚本自身的运行日志与健康自检监控脚本自己也要写运行日志。它没有界面挂了谁也不知道所以必须在脚本内部记录启动、加载规则、匹配告警这些关键动作。用标准logging库就好了按天滚动import logging from logging.handlers import TimedRotatingFileHandler logger logging.getLogger(log_monitor) handler TimedRotatingFileHandler( /var/log/log-monitor/run.log, whenmidnight, backupCount7, ) formatter logging.Formatter(%(asctime)s [%(levelname)s] %(message)s) handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO)还有一个很实用的自检思路每天固定时间比如早上九点给自己发一条心跳消息内容是“监控脚本运行中最近24小时匹配到N条告警规则”。这样既确认脚本没挂又让值班的人有个心理预期。这个动作在脚本里就是加个定时器不复杂但价值很高。5. 线上踩过的坑日志轮转、编码、误报三大问题把脚本跑起来只是开始真正让这套东西“耐用”的是处理了下面几个线上问题。这些问题不遇到还好遇到了如果不系统性排查很容易让你对脚本稳定性失去信心。5.1 日志轮转后文件句柄失效inode判断与自动重开第一个坑来自日志轮转。Linux下logrotate会定期把app.log重命名成app.log.1再创建新的app.log。但我的脚本打开的是文件句柄不是文件路径——重命名之后脚本还握着旧文件的句柄继续读它的新内容而新日志其实写在新的app.log里脚本完全看不到。这就是经典的tail -f轮转失联问题命令没报错但你看不到新日志了。解决办法是每次读文件前检查文件底层的inode是否变化。同一文件的st_ino和st_dev应该保持不变一旦发现变化说明文件被轮转或替换了要关闭旧句柄重新打开新路径class LogTail: def __init__(self, filepath): self.filepath filepath self.f open(filepath, rb) self.dev, self.ino self._get_file_id() self.f.seek(0, os.SEEK_END) def _get_file_id(self): st os.fstat(self.f.fileno()) return st.st_dev, st.st_ino def check_rotation(self): try: st_new os.stat(self.filepath) except FileNotFoundError: return if (st_new.st_dev, st_new.st_ino) ! (self.dev, self.ino): self.f.close() self.f open(self.filepath, rb) self.dev, self.ino self._get_file_id() logger.info(日志文件轮转检测到已重新打开 %s, self.filepath)这个坑最阴险的地方在于它不会立刻让你知道。可能轮转后几个小时你才发现某段重要日志根本没被监控到。所以轮转检测必须在每次循环都执行千万不要图省事只在启动时检测一次。5.2 Windows下GBK日志乱码编码探测与兜底另一个高频坑是编码。Windows下很多软件生成日志用GBK/GB2312编码直接用utf-8解码会直接抛UnicodeDecodeError或者是一堆乱码。现在很多日志文件是UTF-8但老系统、工业软件、上位机程序经常还是GBK。热词里出现过的“nx12捕获到标准c异常有关详细信息请参见系统日志 文件:o:...”就是Windows C程序典型的日志形态这类文件在中文Windows上往往以系统区域编码写入。我在LogTail里做了两级解码兜底先尝试UTF-8失败就用GBK再失败就转成替代字符line raw.decode(utf-8)改为def decode_log(raw): for encoding in (utf-8, gbk, latin-1): try: return raw.decode(encoding).rstrip(\n) except UnicodeDecodeError: continue return raw.decode(utf-8, errorsreplace).rstrip(\n)另外Windows下如果目标是系统事件日志事件查看器直接读.evtx文件不方便常见做法是用win32evtlog模块主动查询。不过那种方式偏Windows运维方向和“脚本监控日志文件”的模式不太一样这里不展开太多。如果你的场景主要在Windows上又需要读系统事件日志可以单独研究一下pywin32的win32evtlog模块思路是完全不同的。5.3 误报比漏报更可怕正则写宽严之间的取舍刚开始写规则时我犯过一个典型错误正则写得太宽。比如规则ERROR本意是“任何ERROR级别日志都要发”但生产环境里很多服务正常运行时也会打ERROR级别日志比如某个外部接口偶发超时但程序有重试机制根本不构成事故。结果是告警邮件一天几十封团队从头两天还很警觉到后面直接无视所有报警邮件——这就是典型的“狼来了”效应误报比漏报更损害监控体系的可信度。我调规则的思路是三层收敛关键字组合把“级别”和“业务上下文”绑定比如ERROR.*DB_TIMEOUT、Exception.*java.sql.SQLException而不是只匹配ERROR。用正则排除已知的正常告警比如ERROR.*(retry|fallback)可以在告警前过滤掉。匹配到关键字后把命中行的前后几行一并打包进通知内容。这个对人工判断特别有帮助很多错误单看一行根本不知道发生了什么但有了上下文值班的人一眼就能看出是数据库抖动还是代码bug。上下文采集代码很简单用collections.deque维护一个滚动窗口保留最近N行from collections import deque context_lines deque(maxlen5) def handle_line(line): context_lines.append(line) current_line len(context_lines) - 1 for rule_name, pattern in RULES.items(): if re.search(pattern, line): context list(context_lines)[max(0, current_line-2): current_line3] notify(rule_name, line, context)6. 把单机脚本扩展成轻量监控体系脚本跑稳定之后你就会开始想“能不能更进一步”。这里分享两个我实际尝试过的扩展方向一个偏事件管理一个偏指标可视化都不需要推翻现有脚本重写。6.1 告警事件落库给复盘留证据告警消息发出去了但它是瞬时性的过几天查“上周这台机器到底报过什么错”就变成难题。最简单的做法是把每次告警事件写进本地SQLitePython自带sqlite3连依赖都不用加import sqlite3 conn sqlite3.connect(/var/lib/log-monitor/alerts.db) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_name TEXT, message TEXT, matched_at TIMESTAMP, notify_status TEXT ) ) def save_alert(rule_name, message): conn.execute( INSERT INTO alerts (rule_name, message, matched_at, notify_status) VALUES (?, ?, ?, ?), (rule_name, message, datetime.now().isoformat(), sent) ) conn.commit()这张表可以按天、按规则做聚合统计比如“最近7天哪个错误出现最频繁”“哪个服务最不稳定”排查问题时非常有说服力也方便周报里写清楚监控覆盖情况和告警量趋势。6.2 对接Prometheus/Grafana的思路自定义exporter如果你的团队已经上了Prometheus和Grafana但你又不想用脚本重写一套规则引擎有个折中方案让脚本兼任一个自定义exporter。脚本每隔一分钟扫描一次日志文件统计最近一分钟内各类错误的出现次数然后暴露一个/metricsHTTP端点Prometheus定期抓取Grafana负责画趋势图。from http.server import BaseHTTPRequestHandler, HTTPServer error_counter {} class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /metrics: body for rule_name, count in error_counter.items(): body flog_alert_total{{rule{rule_name}}} {count}\n self.send_response(200) self.send_header(Content-Type, text/plain) self.end_headers() self.wfile.write(body.encode()) else: self.send_response(404) self.end_headers() HTTPServer((0.0.0.0, 9102), MetricsHandler).serve_forever()这个方案的定位很清晰实时告警还是由脚本第一时间发Webhook指标曲线交给Grafana慢慢画。两者不冲突反而是分工配合。需要注意脚本的错误计数器需要周期性清零避免累计值误导。6.3 一条值得长期维护的监控链路日志→事件→人把单机脚本扩展成一个轻量监控体系本质上是理顺一条链路日志文件是数据源脚本负责持续跟踪和分析规则引擎负责把“原始日志”转成“有意义的事件”事件分两路走一路通过邮件、Webhook通知到具体的人一路落到SQLite或者被Prometheus抓取形成指标人根据通知内容处理问题事后通过落库记录复盘。这条链路的每一环都不复杂但组合起来已经具备一个完整监控系统的雏形。如果你所在的环境机器数量再多一些也可以考虑把脚本放到一台中控机上通过SSH或者共享文件系统采集多台机器的日志不过那就是另一个话题了。写到这里结合我这一年的实际运维感受再补两句。日志监控脚本看起来简单但真正让它可靠运行的难点从来不是代码而是对文件轮转、编码、告警去重、规则收敛这些细节的处理。如果你正准备上手建议先跑通最小闭环一个日志文件、一条规则、一个Webhook然后逐步叠加功能和规则。这样出了问题排查范围小也不会一上来就被各种细节淹没。