用Python手写日志监控告警脚本：从tail -f到企业微信推送

凌晨两点半手机连着震了七八下群里有人我“支付服务挂了日志里全是报错什么时候能恢复”我登录服务器tail -f 翻了几屏发现数据库连接池从晚上十一点就开始报错愣是没人发现。那天晚上我就在琢磨必须得给日志加个哨兵。在Linux服务器上维护服务“看日志”是最朴素也最绕不开的运维动作但人的注意力撑不过24小时尤其是半夜三更。用Python监控系统日志并发送警报本质就是把“tail -f 眼睛盯着 发现问题后通知人”这套流程自动化。这篇文章不聊什么高深的分布式链路追踪就讲我从零写起的一套日志监控告警脚本完整拆开日志怎么读、规则怎么配、告警怎么发、怎么部署成常驻服务以及跑了一个月踩到的那些坑。适合小团队里管着几台服务器、又不想为了日志监控专门搭一套ELK的运维和开发同学参考。1. 为什么放着现成的监控平台不用非要自己写Python1.1 日志监控到底在解决什么问题先把问题说透。服务出故障很少是瞬间崩溃更多是先冒出一些征兆数据库连接失败、内存报警、某个接口响应突然变慢。这些征兆基本都会写进日志里问题只是没人盯着。人盯日志有两个天然缺陷看不过来也熬不住。看不过来是因为一个服务一小时能刷几千行日志靠肉眼从里面挑异常效率极低而且容易疲劳漏看熬不住更直白——你不可能24小时守在终端前凌晨三点注意力本来就涣散更别说还有排班、请假这些事。日志监控脚本就是为了补这两个窟窿它不睡觉不眨眼逐行扫日志命中规则就地发出通知。1.2 和ELK、Zabbix这类平台比自己写的优劣势很多团队一聊日志监控下意识就会想到专业平台。我自己也折腾过几种这里直接给结论方案擅长的事不擅长的事Prometheus / ZabbixCPU、内存、磁盘等指标告警很成熟解析日志文本、判断业务层异常门槛比较高ELKElasticsearch Logstash Kibana海量日志存储、全文检索、可视化强组件多资源占用大小服务器先被吃掉小半内存云厂商日志服务免运维按量计费接入快数据要外传长期跑成本要算清楚自己写Python监控轻量、和业务同机部署、规则完全可控没有海量存储和复杂检索能力规则要自己维护我并不是说专业平台没用。如果你的日志一天能产生几十GB需要全文检索、聚合统计、多人共享分析那老老实实上ELK或云厂商服务别用破脚本硬扛。但如果只是几个服务、一天几百MB到几个GB的日志异常不能漏那自己写Python反而更实在一个脚本加一张配置文件部署在被监控的机器上不依赖外部存储不引入额外服务。想加一条告警规则就加一条不用先过一遍平台配置的流程。还要想清楚一个边界这个脚本是“哨兵”不是“档案馆”。它不负责把历史日志存下来只负责盯文件末尾新增的内容发现匹配规则的异常就通知人。真正的历史日志存储还是留在原文件里要翻旧账时继续用 grep 和 less 就行。1.3 我给脚本定的几个基本原则动手之前我给这个脚本定了几个目标后面所有设计都围绕它们展开单文件所有代码尽量收敛在一个文件里部署就是拷贝一个脚本加一个配置。配置和代码分离日志路径、告警规则、冷却时间都写进YAML改配置不用动代码。依赖极简除了requests和PyYAML不再引其他第三方库降低安装和环境出问题的概率。能长跑脚本要能开机自启、崩溃自愈遇到单行日志问题不能整个进程退出。这几条看着简单实际跑起来才发现每一条都对应着真实的坑后面会逐个讲到。2. 动手前先做三件事日志摸底、通道确认、环境准备2.1 先摸清日志格式和路径写代码前别急着打开编辑器先在服务器上泡一两天把要监控的日志文件挨个看一遍搞清楚三件事路径在哪是 /var/log/myapp/app.log还是 Nginx 的 access.log / error.log又或者某个Java服务的多行日志目录。路径不固定后面配置文件里要写清楚。编码是什么多数是UTF-8但偶尔会碰到GBK甚至ISO-8859-1的。用file /var/log/myapp/app.log这种命令先确认一下。这个细节很关键很多脚本跑几天突然崩溃就是因为某一行日志编码不在预期内。正常波动有多高用 grep 统计一下关键字频率。比如grep -c ERROR /var/log/myapp/app.log看一天有多少条。如果平时就有几百条ERROR说明业务本身噪音就大阈值就得调高如果平时一周才一两条那规则可以设置得很严格漏一条都不行。我当时监控的是一套Java服务加一个Nginx。Java服务是log4j输出的格式比较规整有时间、级别、类名、消息Nginx的access.log则是行式文本结构简单但量很大。这两类日志的匹配重点完全不同Java日志要抓Exception和内存错误Nginx日志主要盯5xx状态码突然变多。2.2 选告警通道我推荐企业微信机器人告警通道我优先推荐企业微信、钉钉、飞书这类群机器人。理由很朴素手机推送直接、配置极其简单、免费而且现在公司内部微信群基本人人都在。相比之下邮件告警实在容易淹没在垃圾邮件里短信又要对接服务商、还要花钱。我最后选的是企业微信群机器人。创建方式非常简单在企业微信里建一个群或者用已有的告警群群设置里找到群机器人添加一个机器人复制它的Webhook地址就行。这个Webhook就是一个HTTPS链接向它POST一段JSON群里就会收到消息。不需要开发企业应用不需要审核五分钟能搞定。钉钉和飞书的机器人原理完全一样Webhook的JSON格式稍有差别但换起来很轻松。如果你所在的公司用的是钉钉就换成钉钉思路和方法完全通用。2.3 Python环境与依赖Python版本建议3.8以上我用的是3.10。系统自带的Python 3.6也能跑但有一些语法上的小差异不差这点升级的时间。依赖只有两个pip install requests PyYAML有人可能会问纯粹用标准库也能写为什么要加两个第三方包坦白说requests发HTTP请求时对超时、异常的处理明显比urllib顺手少写很多样板代码PyYAML则是因为用YAML写配置比硬编码在代码里好维护得多。这两个包都是纯Python的装起来没有编译负担。脚本我直接放在被监控的服务器上目录就选/opt/logmonitor/里面放log_monitor.py和config.yaml。不需要另搞一套agent体系一台机器上跑一个监控脚本就够了。3. 日志采集从一次性读文件到跟随式tail -f3.1 为什么不能直接把日志文件整个读进来最直观的写法是open(file_path).readlines()然后逐行匹配但这种写法在生产环境基本就是找死。第一个问题是内存一个日志文件动辄几个GB全部读进内存服务器直接卡顿监控脚本先把自己监控的机器搞挂了这算什么事。第二个问题是效率就算文件不大每次从开头读到结尾处理已经见过的历史日志毫无意义时间和IO全被浪费。日志监控的正确姿势是“增量读”——只处理文件新增的那部分内容就像tail -f做的事一样。所以核心思路就一句话记录当前文件读到哪了下次只读新增部分。3.2 手写一个mini版tail -f有现成的tail -f命令但我们不能直接在Python里调系统命令来拿日志流匹配逻辑还是在Python里做更灵活。手写一个生成器不超过20行import os import time def tail_follow(file_path, sleep_seconds0.5): with open(file_path, r, encodingutf-8, errorsreplace) as f: f.seek(0, os.SEEK_END) while True: line f.readline() if line: yield line.rstrip(\n) else: time.sleep(sleep_seconds)逻辑很简单打开文件后先把游标seek到文件末尾这样脚本只关心启动之后产生的新日志历史日志直接跳过。然后进入无限循环readline()读出一行就yield出去没有新行就睡0.5秒再继续。sleep_seconds0.5这个参数是试出来的太频繁会白白消耗CPU太慢告警延迟会变高。0.5秒对日志监控来说完全够用一般告警延迟一两秒不会造成什么影响。3.3 用inode识别日志轮转防止漏读日志文件不是永远不变的。Linux下最常见的logrotate每天或每周会把日志文件轮转一次处理方式主要有两种create模式先把原日志文件重命名成带日期的文件比如 app.log.1再在原来的路径下创建一个新的空文件。这种情况下文件的inode会变。copytruncate模式先复制原文件内容到另一个文件然后清空原文件。inode不变。对我们来说最大的坑是create模式。如果脚本一直用固定路径打开文件它读到的其实是那个已经被重命名的旧文件新文件的内容全都会被漏掉。这时候需要用inode来判定文件是否被重建一旦inode变了就重新打开路径import os import time def get_inode(path): return os.stat(path).st_ino def watch_file(path, sleep_seconds0.5): while True: try: current_inode os.stat(path).st_ino except FileNotFoundError: time.sleep(sleep_seconds) continue try: with open(path, r, encodingutf-8, errorsreplace) as f: f.seek(0, os.SEEK_END) while True: line f.readline() if line: yield line.rstrip(\n) else: time.sleep(sleep_seconds) try: new_inode os.stat(path).st_ino except FileNotFoundError: break if new_inode ! current_inode: # 文件已经被轮转重新打开路径 break except FileNotFoundError: time.sleep(sleep_seconds)核心逻辑是内层循环每次没有新行时顺便检查一次inode。如果inode变了说明文件被轮转直接跳回外层循环重新打开新文件。如果文件在轮转瞬间被删除重建os.stat会抛FileNotFoundError捕获后睡一秒重试。这样不管logrotate怎么折腾脚本都能跟上。顺带提一句copytruncate模式inode不变但文件可能被清空重建。这种情况比较少见稳妥的做法是同时对比文件大小如果发现文件大小比当前已经读到的位置还小说明文件被截断需要重新seek到开头。我没在脚本里加这个因为我的环境用的是create模式但如果你遇到copytruncate记得补上这个判断。3.4 多个日志文件同时监控怎么办如果只监控一个文件按上面的代码跑就行。但多数场景下要盯好几个文件应用日志、Nginx错误日志、系统安全日志。这时候最直接的方式是每个文件开一个线程。from concurrent.futures import ThreadPoolExecutor def monitor_one_file(path, config): for line in watch_file(path): handle_line(line, path, config) with ThreadPoolExecutor(max_workerslen(config[files])) as executor: for path in config[files]: executor.submit(monitor_one_file, path, config)tail_follow本质是sleep循环单线程逐个处理的话排在后头的文件延迟会越来越高。给每个文件分配一个worker互不干扰日志量不大时完全够用。4. 告警规则关键词、正则和冷却时间去重4.1 规则配置文件要长什么样我习惯把规则写在YAML里和代码分开。一张示例配置files: - /var/log/myapp/app.log rules: - name: database_connection_failed pattern: connection (refused|timed out) level: critical cooldown: 300 - name: out_of_memory pattern: OutOfMemoryError|GC overhead limit exceeded level: critical cooldown: 600 - name: slow_request pattern: request time.*[5-9][0-9]{2,}ms level: warning cooldown: 120每个规则有四个字段含义如下字段说明示例name规则名告警消息里会展示database_connection_failedpattern正则表达式匹配日志行connection (refused|timed out)level告警级别critical才发通知criticalcooldown冷却时间单位秒300加载配置时用re.compile把每个pattern编译一次存起来不要在每行日志匹配时才现编译。这是性能最容易被忽略的一点正则编译很贵日志量大时现编译会让CPU飙得很难看。4.2 关键词预过滤加正则精确匹配CPU省一大半直接拿每行日志去跑所有正则规则一多CPU就吃紧。我加了一层预过滤只有包含error、exception、fatal、warn、fail这些关键词的行才值得拿去跑正则。import re QUICK_KEYWORDS (error, exception, fatal, warn, fail) def match_rules(line, config): lowered line.lower() if not any(k in lowered for k in QUICK_KEYWORDS): return None for rule in config[rules]: compiled rule.get(compiled) if compiled and compiled.search(line): return rule return None这个优化在高峰期效果非常明显。大部分日志行是INFO级别的正常记录根本不包含异常关键词一行in判断就能过滤掉完全不需要碰正则。只有少数行进入正则匹配环节几十条规则也跑得动。4.3 冷却时间同一个错误5分钟内只叫一次服务重启、数据库闪断这种场景下应用会疯狂重试日志里同一类报错一分钟能刷几十行。如果每条都发告警手机一晚上能被打爆。冷却时间机制就是为了解决这个问题对每个规则维护一个last_alert_time在冷却窗口内即使又命中了相同规则也只记日志不再发通知。窗口过了才允许下一条告警发出。import time _last_alert_time {} def need_alert(rule_name, cooldown): now time.time() if now - _last_alert_time.get(rule_name, 0) cooldown: _last_alert_time[rule_name] now return True return False冷却时间一定要作为规则的一个字段而不是全脚本统一写死。数据库连接失败这种严重错误5分钟之内发一条够了而某个业务参数异常导致的报错可能1分钟一条才合适。不同规则的容忍度不同统一阈值反而会两头不讨好。4.4 分级处理不是所有错误都值得手机震动人的注意力是有限的。如果手机上每天收到几十条告警很快就会麻木真正重要的告警反而被无视。所以我把告警分成两级critical真的出问题了必须立刻通知发企业微信。warning值得记录但不需要立刻处理。只写进本地日志文件不推送。比如Nginx的5xx状态码暴增可能是上游服务挂了算critical而某个接口偶尔出现一次慢请求算warning就行。这样告警的“信噪比”会高很多。宁可漏掉一次不重要的也不要误报十次把人搞疲劳。5. 告警发送把错误信息推进微信群机器人5.1 申请一个群机器人Webhook前面说过企业微信群机器人创建非常简单这里把步骤列出来在企业微信里创建一个群或者拉一个已有的告警群。群设置里找到“群机器人”点击“添加机器人”。按提示给机器人起个名字复制生成的Webhook地址。得到的Webhook地址长这样https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx这个key就是机器的身份凭证注意别泄露到公开仓库里。告警群建议用专门的群不要混在日常业务群里否则重要消息容易被刷过去。5.2 发送函数超时和异常处理必须做有了Webhook发消息其实就是向这个地址POST一段JSONimport requests WEBHOOK_URL https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx def send_wecom(content): payload { msgtype: text, text: { content: content, }, } try: resp requests.post(WEBHOOK_URL, jsonpayload, timeout5) data resp.json() if data.get(errcode) ! 0: print(告警发送失败, errcode:, data.get(errcode)) except requests.RequestException as e: print(告警发送异常:, e)timeout5这个参数非常关键。如果不设超时万一企业微信的接口出问题requests.post会一直挂在那里把监控主循环整个卡住。告警通道自己把监控脚本拖死这画面太讽刺了。发送失败的处理原则是“不阻断主流程”打印一行本地日志就继续跑不要因为一次发送失败退出整个监控。真正的故障会反复命中规则冷却时间结束后会再次尝试发送不差这一次。5.3 消息包装让手机上的告警一眼看懂告警消息的格式决定了人收到消息后能不能不加思考就判断出问题方向。我的消息长这样【CRITICAL】数据库连接失败 规则database_connection_failed 时间2025-01-01 00:00:00 主机web-01 文件/var/log/myapp/app.log ------------------------------------------------ ERROR 2025-01-01 00:00:00 db pool exhausted, retry 3/5 failed ERROR 2025-01-01 00:00:01 db pool exhausted, retry 4/5 failed ERROR 2025-01-01 00:00:02 db pool exhausted, retry 5/5 failed包含的信息有告警级别、规则名、精确时间、哪台机器、哪个日志文件以及命中的日志和上下文。这样手机一响不用登录服务器就能判断事情大概严重到什么程度。为什么要带上下文因为单看一行ERROR往往不知道前因后果但如果附上前面几行日志就能看出是内存爆了还是连接池耗尽还是上游超时。我在代码里维护了一个deque(maxlen3)始终保留最近三行日志命中的时候带着上下文一起发出去。from collections import deque context deque(maxlen3) for line in watch_file(path): context.append(line) rule match_rules(line, config) if rule and rule[level] critical and need_alert(rule[name], rule[cooldown]): send_wecom(build_message(rule, line, context, path))企业微信机器人还支持markdown格式的消息能加粗、变色。但我实际用下来还是text格式最稳各种客户端渲染都不出问题。告警消息追求的是可读和可靠不是花哨。6. 从脚本到服务systemd托管与健壮性设计6.1 为什么不用nohup糊弄自己很多人跑后台脚本第一反应是nohup python monitor.py 。这个做法有两个硬伤第一进程挂了没人管不会自动重启第二没有开机自启服务器一重启脚本就消失了。日志监控这种基础设施最怕的就是它自己不监控自己。所以必须交给systemd托管这是Linux上最标准的做法。6.2 一个能开机自启的unit文件在/etc/systemd/system/logmonitor.service里写[Unit] DescriptionPython Log Monitor Afternetwork-online.target Wantsnetwork-online.target [Service] Typesimple WorkingDirectory/opt/logmonitor ExecStart/usr/bin/python3 /opt/logmonitor/log_monitor.py Restartalways RestartSec10 StandardOutputappend:/var/log/logmonitor/monitor.log StandardErrorappend:/var/log/logmonitor/monitor_error.log EnvironmentPYTHONUNBUFFERED1 [Install] WantedBymulti-user.target几个关键配置说明一下Restartalways不管进程是正常退出还是异常退出都会拉起来。日志监控这种常驻服务没有主动停掉的命令时就不应该消失。RestartSec10崩溃后等10秒再重启避免陷入快速重启循环也给系统留点缓冲时间。EnvironmentPYTHONUNBUFFERED1让Python的标准输出不缓冲。否则print的内容要等缓冲区满了才落盘日志会看起来很滞后。Afternetwork-online.target Wantsnetwork-online.target确保网络就绪后再启动。不然开机阶段脚本启动时Webhook请求必然失败虽然不致命但看着烦。保存后执行sudo systemctl daemon-reload sudo systemctl enable logmonitor sudo systemctl start logmonitorenable是开机自启start是立即启动。之后想看监控脚本自己的状态用systemctl status logmonitor想跟着看它处理日志的过程用journalctl -u logmonitor -f或者直接看 /var/log/logmonitor/monitor.log。6.3 脚本自身的异常捕获和内存控制systemd能保证进程活着但代码自身的健壮性才是根基。最容易让脚本挂掉的几类问题得提前规避。一是逐行处理日志时单行坏数据不能拖垮整个进程。比如日志里混进了一行无法解码的二进制内容如果不做处理UnicodeDecodeError会直接抛出整个循环终止。我的做法是外层套一个异常捕获单行处理出错就打印错误并继续往下读不退出。def handle_line(line, path, config): try: rule match_rules(line, config) if rule and rule[level] critical and need_alert(rule[name], rule[cooldown]): send_wecom(build_message(rule, line, context, path)) except Exception as e: print(处理日志行出错:, e)二是内存不能无限增长。日志本身是流式的逐行读进来逐行丢掉内存占用很稳定。但如果你为了保留上下文用了普通列表每行都append列表会无限变长跑到最后内存直接爆炸。上文已经提过用deque(maxlen3)固定只保留3行最近内容不会多占一字节。三是网络异常不能卡死主循环。send_wecom里已经加了超时和异常处理这还不够——如果Webhook连续失败requests.post每5秒超时一次主循环会被反复拖住。我的做法是发送失败时记录一个连续失败计数达到5次就暂停发送10分钟期间只往本地日志写不让网络问题拖死整个监控。7. 运行一个月踩过的坑编码、轮转和告警风暴7.1 日志编码混用UTF-8和GBK打架真实案例我在一台服务器上同时监控Java应用日志和第三方SDK日志。Java应用输出UTF-8第三方SDK却固执地输出GBK。用UTF-8去读GBK的行读到某个特殊字符时直接抛UnicodeDecodeError监控进程当场退出而后面的重要日志全都没人盯了。解法就是前面所有open都带上的那个参数errorsreplace。遇到无法解码的字节不报错替换成。虽然消息里偶尔会看到乱码但监控不会断。对日志监控来说“宁可要乱码也不能断档”。7.2 轮转瞬间的FileNotFoundError有一次logrotate执行的时间点和脚本检查inode的时间点撞上了文件在重命名和重新创建之间有一个极短的间隙路径会短暂不存在。脚本在那个瞬间调用os.stat直接抛了FileNotFoundError。光一个异常还不至于致命但如果没写捕获整个循环就会终止。我当时测试时没覆盖这个场景结果日志轮转当天监控就静默熄火了等发现问题已经是两天后。解法就是watch_file里那个外层try/except FileNotFoundError捕获后sleep一秒重试。这个坑很典型生产环境的logrotate时间是不可控的脚本必须对文件路径短暂缺失有容忍力。7.3 告警风暴一次故障收到上百条消息最猛烈的一次告警风暴发生在数据库实例迁移的时候。应用与数据库的每三次重试里必有一次ERROR冷却时间设的10分钟根本不够看——一晚上过去企业微信群里刷了上百条告警手机通知栏滑不到底。事后我把逻辑改成了“冷却时间 命中计数聚合”的组合方案。核心思路是冷却窗口结束之后如果这段窗口内这条规则又继续命中了很多次就只发一条聚合消息把计数带上而不发逐条消息。hit_counter {} if need_alert(rule[name], rule[cooldown]): hits hit_counter.get(rule[name], 0) send_wecom(f【{rule[level]}】{rule[name]} 近{rule[cooldown]//60}分钟累计命中{hits 1}次) hit_counter[rule[name]] 0 else: hit_counter[rule[name]] hit_counter.get(rule[name], 0) 1这样消息量从上百条降到一两条但信息量一点没少——反而多了“这个错误持续在刷屏”这个重要信号。告警降噪的核心就一句话宁可少发不可刷屏。发出一条能被认真对待的消息远比发一百条没人看的消息有价值。后续还能怎么扩展这个脚本跑稳定之后我又在上面加了两件小事一个是对外端口探活每30秒检查关键端口是否可达一个是磁盘水位检查超过80%就告警。都是在同一个脚本里加规则不需要引入新服务。如果你想做更复杂的告警聚合、值班人轮换、故障自愈也可以在这个基础上慢慢加但第一步永远是先把日志监控跑起来把规则库积累起来。我个人的体会是这种轻量监控最忌一上来就设计得很复杂先跑通一条链路再根据自己的真实场景慢慢迭代比什么都重要。