1. 项目概述与核心需求1.1 为什么你需要自己写日志监控做运维或者后端开发的人多多少少都经历过这样的场景凌晨两点被电话吵醒说线上服务挂了你迷迷糊糊爬起来连上服务器翻了一大堆日志文件才在某个不起眼的角落发现了堆栈报错。整个过程少则十分钟多则半小时服务已经影响了一大批用户。我最早也是这么过来的。后来项目规模大了服务器从三五台增加到几十台日志文件从几MB膨胀到几十GB光靠人眼盯日志文件已经不是效率低的问题而是根本盯不过来。这才下定决心用Python写了一套日志监控和警报系统专门替我做“盯梢”这件事。这套系统的核心需求其实非常简单实时读取系统日志文件发现异常关键字就触发警报通过多种渠道通知到负责人。听起来不复杂但真正做起来会发现很多细节问题——日志文件的编码格式、轮转策略、多文件并发监控、警报去重、通知渠道的稳定性等等每一个都能让你踩半天坑。我做的这套方案选型很简单主要基于两个核心库watchdog负责监控文件变化smtplib和requests负责发送邮件和HTTP通知。这套技术栈没有用任何重型框架部署起来轻松依赖少一套代码可以复制到任意一台机器上跑非常符合“轻量级运维工具”的定位。1.2 这套系统能解决什么问题先说结论这套日志监控警报系统适合以下几类人和场景。个人开发者或小团队没有预算采购商业化监控平台需要维护几台服务器希望用最低成本实现基础告警能力。运维工程师需要快速定制监控规则不想被现有监控平台的配置方式限制住希望直接用代码表达监控逻辑。后端开发人员想给自己的微服务加上简单的日志巡检能力比如特定错误码统计、登录失败次数检测、异常频率突增告警。我做的这套系统最终实现了以下功能实时监控多个日志文件能够自动跟随文件的新增内容基于正则表达式匹配异常关键字支持多规则配置支持按时间段统计异常出现次数做到频率型告警告警消息自动去重防止同一故障反复刷屏同时支持邮件、钉钉/企业微信Webhook通知所有配置都写在YAML文件里改规则不用动代码这套系统在我在维护的数台测试服务器上运行了半年多累计捕获异常告警超过两百次其中真正需要人工介入的故障大约占四成左右其余是瞬时抖动或者重复日志产生的内容经过告警去重后已经不会产生打扰了。2. 技术选型与方案设计思路2.1 轮询模式还是事件驱动模式做日志监控第一步需要决定的就是用轮询(Polling)的方式定时读取日志文件还是用事件驱动的方式实时感知文件变化。轮询模式很好理解就是每隔固定时间比如1秒或者5秒打开日志文件从上次读取的位置继续读取新内容然后进行匹配。这种方式的优点是实现简单不用依赖操作系统底层的事件通知机制在Windows和Linux上行为完全一致。缺点也很明显实时性稍差如果轮询间隔设置过长告警可能滞后而且频繁打开文件IO开销较大。事件驱动模式则依靠操作系统的文件系统事件通知机制。在Linux上主要是inotifymacOS上是FSEventsWindows上是ReadDirectoryChangesW。Python的watchdog库对这三者做了很好的封装你只需要注册一个Observer它就会在文件内容变化时自动触发回调函数。这种方式实时性好IO开销小在高并发写入的场景下优势明显。我最后选择了watchdog作为文件监控层同时保留了一个兜底机制每间隔5秒主动检查一次文件大小和最后修改时间。这么做的原因是我在测试中发现某些特殊情况比如文件被移动后重新创建、日志写入频率极低下事件通知并不可靠主动轮询可以确保不漏报。2.2 为什么不用现成的商业监控平台做技术选型的时候圈子里讨论最多的问题往往是日志监控用现成平台不就行了为什么要自己写确实市面上监控平台不少比如ELK、Splunk、甚至云厂商自带的日志服务。但对我来说这些方案都有点“重”。ELK全家桶部署起来需要占用较多系统资源对测试服务器或者小型生产环境来说有点小题大做。云厂商自带的日志服务需要额外付费而且日志数据出了自己的服务器特殊行业场景下可能还会有一些数据合规方面的顾虑。自己写这套Python方案核心优势在于轻量只需要安装两三个pip包、可控逻辑都在自己的代码里想怎么改就怎么改、透明出了问题可以直接调试源码不用等厂商排障。当然自己写的方案也有缺点比如没有现成的UI界面告警规则调整需要改配置文件没有历史数据趋势分析。但作为运维自动化体系里的一块补位工具它完全够用而且边际成本几乎为零。2.3 整体架构和模块划分思路理清楚之后我把整个系统拆成了四个模块日志文件监听模块负责监控文件变化读取新增内容规则引擎模块加载配置中的正则规则对日志内容进行匹配告警决策模块判断是否触发告警、是否进行去重、是否合并同类告警通知发送模块封装邮件、Webhook等不同渠道的发送逻辑这四个模块之间通过简单的函数调用串联不需要消息队列不需要数据库。为什么不做成生产者-消费者模式因为在当前场景下日志量每小时不超过几百MB时单线程同步处理完全够用引入并发反而增加代码复杂度。等到真的遇到海量日志场景再上asyncio或者多进程也不迟性能瓶颈也不会出现在匹配和通知这两个环节上。模块化设计带来的最大好处是单个模块出了问题不影响其他模块运行。比如邮件发送失败只会记录一条错误日志不会导致文件监听停摆。3. 环境准备与基础配置3.1 Python环境与依赖安装这套方案对Python版本没有特殊要求3.7以上的版本都可以跑。我本人在线上环境中使用的是Python 3.9没有遇到兼容性问题。需要安装的依赖只有两个pip install watchdog pip install pyyamlwatchdog用于文件系统监控pyyaml用于加载配置文件。如果你不想用YAML改成JSON配置也行那就连pyyaml都不用装。只不过长期维护的话我还是推荐YAML它支持注释规则多了以后可读性会好很多。建议顺手建一个虚拟环境来跑这套脚本毕竟它是要长期挂在后台的进程。虚拟环境的好处是不跟系统Python环境互相污染未来升级Python版本或者卸载依赖都不会影响到其他服务。3.2 项目目录结构说明我的项目目录是这样的log_monitor/ ├── monitor.py # 主入口脚本 ├── config.yaml # 监控规则配置 ├── requirements.txt # 依赖列表 ├── logs/ │ └── monitor.log # 系统自身的运行日志 └── rules/ └── syslog_patterns.json # 预置规则模板主逻辑全部放在monitor.py里一个小技巧是并没有把长篇幅的函数拆成十几个模块文件因为这套代码总共只有三百多行。逻辑虽然多但放在一个文件里反而容易阅读和维护。只有当代码量达到两三千行的时候才考虑进一步拆分。3.3 配置文件初版编写配置文件是整个系统的核心我习惯先写配置再写代码因为代码都是围绕配置展开的。初始版本的配置文件如下# config.yaml monitor_files: - path: /var/log/syslog encoding: utf-8 rules: - name: error regex: ERROR|Exception|Traceback enabled: true - name: critical regex: CRITICAL|FATAL enabled: true alert_threshold: 3 alert_channels: email: enabled: true smtp_server: smtp.qq.com smtp_port: 465 sender: monitorexample.com password: your_password receivers: - opsexample.com webhook: enabled: true url: https://your-server/webhook/alert headers: Content-Type: application/json dedup: window_seconds: 300 max_alerts: 3monitor_files列表里定义了要监控的文件路径、编码格式和匹配规则。dedup配置区用来控制警报去重的窗口时间和最大发送次数。下面写代码时我会解释这些字段具体怎么被使用。4. 核心功能实现与关键代码解析4.1 文件事件监听器的实现监听文件变化我是用watchdog的PatternMatchingEventHandler来实现的。import time from watchdog.observers import Observer from watchdog.events import PatternMatchingEventHandler class LogHandler(PatternMatchingEventHandler): def __init__(self, config, matcher, alerter): super().__init__(patterns*.log, ignore_directoriesTrue) self.config config self.matcher matcher self.alerter alerter self.file_positions {} def on_modified(self, event): file_path event.src_path self.process_file(file_path) def process_file(self, file_path): if file_path not in self.file_positions: self.file_positions[file_path] 0 position self.file_positions[file_path] try: with open(file_path, r, encodingutf-8, errorsignore) as f: f.seek(position) new_lines f.readlines() self.file_positions[file_path] f.tell() for line in new_lines: self.matcher.match_line(file_path, line) except Exception as e: print(f读取文件失败 {file_path}: {e})注意errorsignore这个参数我在实际运行中碰到过一个问题有些日志文件某一行的编码格式跟文件整体的编码声明不一致日志输出中途换过编码、或者从别的系统导入过内容不加这个参数readlines()会直接抛UnicodeDecodeError导致整个文件读取中断后续日志全部漏掉。加了errorsignore之后即使个别行解析不了也只是跳过那一行的匹配系统的健壮性明显提升。on_modified回调函数会在文件被写入时自动触发但这只能感知文件的修改事件。如果日志文件被轮转logrotate把旧文件改名、新文件顶替监听器可能会漏掉新文件的创建事件所以我还额外覆写了on_created事件让它也能捕获新文件出现的情况。4.2 读取偏移量管理的关键细节上面代码中self.file_positions这个字典保存了每个文件的读取位置它的作用很像书签。每次进程重新启动后文件位置会重置导致重新打开文件时从头开始读取一遍——这通常不是我们想要的行为。所以我在实际代码中把文件位置持久化到本地磁盘用pickle序列化保存到monitor_cache.pkl文件里。每次读取完文件更新字典后立刻保存。进程重启后先从缓存文件恢复位置再继续读取。这里有一个边界条件需要处理如果日志文件被清空或者重写文件大小变小原来的位置就失效了。我的处理方式是在读取前先判断当前文件大小是否小于上次记录的位置如果是说明文件被重置直接把位置归零重新读取。4.3 规则引擎与关键字匹配规则引擎的逻辑判断比较简单难的是让规则足够灵活既能覆盖异常模式又不会产生太多误报。我实现的规则匹配函数import re import json from datetime import datetime, timedelta class RuleMatcher: def __init__(self, rule_config): self.rules rule_config self.alert_state {} def match_line(self, file_path, line): for rule in self.rules: if not rule.get(enabled, True): continue pattern rule[regex] if re.search(pattern, line): self.handle_match(rule, file_path, line) def handle_match(self, rule, file_path, line): now datetime.now() rule_id rule[name] state self.alert_state.setdefault(rule_id, { count: 0, first_time: None, last_alert: None }) state[count] 1 if state[first_time] is None: state[first_time] now threshold rule.get(alert_threshold, 1) if state[count] threshold: return if state[last_alert] and (now - state[last_alert]) timedelta(secondsrule.get(silent_seconds, 300)): return alert_message { rule: rule_id, file: file_path, line: line.strip(), count: state[count], time: now.strftime(%Y-%m-%d %H:%M:%S) } self.alerter.send_all(alert_message) state[count] 0 state[last_alert] now单独看这段代码重点在于alert_threshold和silent_seconds的设置。alert_threshold控制每多少条匹配才产生一条告警用于过滤偶发性的低概率错误silent_seconds控制同一条规则的最小告警间隔防止故障恢复前每秒钟刷屏。举个例子假设配置文件里给某个匹配规则设了alert_threshold: 3这意味着只有当同一关键字的日志出现3次时才会触发告警。连续出现2次就消失的瞬时抖动不会打扰你。silent_seconds设成300秒的意思是第一次告警后5分钟内再触发同一规则不会再次发消息。如果故障持续超过5分钟会再收到一条新告警这样既不会攻击性刷屏也不会彻底失声。4.4 告警去重与合并机制除了规则级别的silent_seconds我在整个系统层面加了一个简单但有效的去重机制基于时间窗口的告警合并。class AlertDeduplicator: def __init__(self, window_seconds, max_alerts): self.window_seconds window_seconds self.max_alerts max_alerts self.alerts [] def should_alert(self, alert_key): now time.time() self.alerts [a for a in self.alerts if now - a[0] self.window_seconds] count sum(1 for a, k in self.alerts if k alert_key) if count self.max_alerts: return False self.alerts.append((now, alert_key)) return Truealert_key我一般取“文件名规则名”的组合。比如/var/log/syslog-error就对应一个key。窗口期内的重复key只放行指定数量超出部分直接丢。这个去重机制和silent_seconds有什么区别silent_seconds是规则内部去重主要针对频繁出现的同类型日志窗口去重是跨规则的全局去重防止多个规则同时命中时瞬间向所有渠道发消息。两者结合起来告警的轰炸问题基本能解决。4.5 多通道通知发送通知模块同时支持邮件和Webhook核心逻辑如下import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import requests class AlertSender: def __init__(self, email_cfg, webhook_cfg): self.email_cfg email_cfg self.webhook_cfg webhook_cfg def send_email(self, subject, content): config self.email_cfg msg MIMEMultipart() msg[From] config[sender] msg[To] , .join(config[receivers]) msg[Subject] subject msg.attach(MIMEText(content, plain, utf-8)) try: smtp smtplib.SMTP_SSL(config[smtp_server], config[smtp_port]) smtp.login(config[sender], config[password]) smtp.sendmail(config[sender], config[receivers], msg.as_string()) smtp.quit() except Exception as e: print(f邮件发送失败: {e}) def send_webhook(self, payload): try: resp requests.post( self.webhook_cfg[url], jsonpayload, headersself.webhook_cfg.get(headers), timeout5 ) print(fWebhook响应: {resp.status_code}) except requests.RequestException as e: print(fWebhook发送失败: {e}) def send_all(self, alert_message): subject f[告警] {alert_message[rule]} - {alert_message[file]} content f规则: {alert_message[rule]}\n文件: {alert_message[file]}\n计数: {alert_message[count]}\n时间: {alert_message[time]}\n内容: {alert_message[line]} if self.email_cfg.get(enabled): self.send_email(subject, content) if self.webhook_cfg.get(enabled): self.send_webhook(alert_message)邮件使用SMTP_SSL直接连接465端口这比starttls方式更省事兼容性也更好。如果你用的是某些需要授权码的邮件服务商填入授权码而不是登录密码。Webhook的timeout参数很重要。我在测试时遇到过一种情况Webhook地址不可达但requests.post()默认没有超时限制导致进程卡在发送环节十几秒不动文件监控也跟着卡住。加了5秒超时之后这个问题得到解决最坏情况也只是告警延迟5秒不会阻塞整个进程。4.6 主程序框架与调度逻辑主程序负责把上面这些模块串联起来还要做异常兜底处理和优雅退出。def main(): config load_config(config.yaml) matcher RuleMatcher(config[monitor_files]) alert_dedup AlertDeduplicator( config[dedup][window_seconds], config[dedup][max_alerts] ) alerter AlertSender( config[alert_channels][email], config[alert_channels][webhook] ) observer Observer() for file_conf in config[monitor_files]: event_handler LogHandler(file_conf, matcher, alerter) path file_conf[path] dir_path os.path.dirname(path) observer.schedule(event_handler, pathdir_path, recursiveFalse) observer.start() try: while True: time.sleep(10) except KeyboardInterrupt: observer.stop() observer.join()这里有个容易被忽视的点observer.schedule()监听的是目录而不是文件。因此我在LogHandler里用patterns*.log过滤出日志文件再对每个文件做处理。如果你要监听的文件路径不在固定的已知目录下就需要递归监听子目录。不过对于日志场景通常文件路径是固定的递归没有意义反而可能误触发无关文件。主循环里的sleep(10)纯粹是为了让进程挂住不退出同时给KeyboardInterrupt留下响应窗口。如果你想使用进程守护工具来管理这个脚本可以在外层套用systemd或者supervisor而不是在脚本里自己实现守护逻辑。5. 部署运行与效果验证5.1 模拟测试环境的搭建写完代码后自然要先验证功能是否正确。我没有直接用生产环境的日志文件做测试而是先建了一个临时目录手动模拟各种日志场景。在/tmp/test_logs/下创建app.log文件然后开启这个监控系统。接着执行命令模拟异常日志echo INFO: request finished /tmp/test_logs/app.log echo ERROR: Connection refused at 10.0.0.5:8080 /tmp/test_logs/app.log sleep 2 echo ERROR: Connection refused at 10.0.0.5:8080 /tmp/test_logs/app.log sleep 2 echo ERROR: Connection refused at 10.0.0.5:8080 /tmp/test_logs/app.log第一个INFO不会触发告警前两条ERROR也没有触发因为阈值设的是alert_threshold: 3第三条ERROR出现后告警条件满足邮件和Webhook就会收到消息。整个过程不到10秒比手动检查日志快了很多。5.2 日志轮转场景的模拟验证日志轮转是日志监控系统最容易失效的场景之一。Linux的logrotate通常在某个固定时间点执行默认的行为是把当前日志文件改名为带日期后缀的文件然后新建一个空文件继续写。我的模拟方法是mv /tmp/test_logs/app.log /tmp/test_logs/app.log.1 touch /tmp/test_logs/app.log echo ERROR: test after rotation /tmp/test_logs/app.log在没有重启监控进程的情况下watchdog能否感知到新文件的创建和写入我实测中大部分情况下能感知到但确实存在偶发漏检的情况。所以我强烈建议在日志轮转后主动做一次文件位置校正或者把轮转后的新文件路径提前加入监控列表。稳妥的方案是写一个简单的继电器监听目录下的app.log.1这类文件出现事件时把位置归零切换到新文件上。5.3 实测效果与延迟分析部署在测试服务器上连续跑了48小时我记录了以下数据场景日志写入方式告警延迟消息送达高频持续写入每毫秒数十条 1秒正常低频间歇写入每小时几条事件触发后立即正常日志轮转轮转后新文件写入2-5秒正常进程重启恢复重启后继续写入立即正常告警延迟主要取决于watchdog的事件通知速度。实测下来watchdog在Linux上的inotify通知延迟基本可以忽略主要耗时反而在邮件发送环节。SMTP认证和连接大约需要几百毫秒到一两秒不等。所以整体延迟控制在1-2秒内完全没问题。5.4 长期运行的资源占用作为后台常驻进程内存占用和CPU占用不能太高。实测结果如下内存占用约40MB包含Python解释器基础开销CPU占用平均0.1%以下高并发日志写入时短暂达到5%磁盘占用除系统日志外每秒几乎没有额外写入CPU占用比预期低的原因在于watchdog是事件驱动的只有文件被写入时才触发回调逻辑不会做大量空转的轮询。加上5秒兜底轮询的代码逻辑非常简单只是os.stat检查文件大小开销可以忽略。6. 常见问题与疑难解答6.1 文件读取到一半就停止响应这是一个我遇到过且非常典型的问题系统运行了几个小时突然发现告警停止发送检查进程发现还活着但文件大小不更新、内容也不匹配了。排查后发现问题出在文件句柄上。某些日志写入方式会打开文件后不关闭导致其他进程无法在该文件上执行seek操作。另外如果日志文件路径发生了变化比如通过软链接指向的文件被替换Python打开的旧文件句柄会指向已经被删除的inode继续read()只会读到EOF永远没有新内容。解决办法是在每次读取文件前重新验证文件路径的inode是否变化。简单说就是import os current_stat os.stat(file_path) if current_stat.st_ino ! self.last_inode[file_path]: self.file_positions[file_path] 0 self.last_inode[file_path] current_stat.st_ino这样做能让监控系统自动适应文件轮转和软链接切换不至于悄悄失去作用。6.2 正则表达式匹配性能与准确性日志量大的时候正则表达式的性能不能忽视。我在实际测试中发现一个复杂的正则表达式在每天百万级日志行中会占用大约30%的匹配时间。如果配置了多个复杂表达式匹配性能会被明显拖慢。优化策略有几个经验可以分享优先使用字符类而不是懒匹配例如ERROR|FATAL|CRITICAL比Err.*更高效避免灾难性回溯不要写(a)这种嵌套量词预编译正则表达式而不是每次匹配都重新编译因此在规则配置加载阶段我应该把所有正则预先编译好而不是在match_line里每次调用re.search时隐式编译。compiled_pattern re.compile(pattern)预编译后性能提升非常明显。我测过的一个场景表达式数量从10个增加到30个如果不预编译匹配耗时增加了将近3倍预编译之后增加的时间基本可以忽略。6.3 邮件被判定为垃圾邮件的问题邮件告警发送了几次之后我注意到一个尴尬的问题有些邮箱服务商把自动推送的告警邮件归到了垃圾邮件文件夹。这对于紧急告警来说是致命的因为运维人员根本不会去看垃圾箱。解决这个问题的关键并不在代码层面而在邮件内容设计上。我的经验是主题行避免全大写和过多的感叹号这会触发垃圾邮件过滤规则邮件正文要包含足够多的正文文本纯HTML或者纯数字的邮件更容易被判定为垃圾邮件发送方的域名做好SPF和DKIM记录避免伪造嫌疑尽量不要用服务器IP直接发邮件域名和服务器绑定好实际修改后被误判为垃圾邮件的概率大幅下降。这个坑如果不是踩过一回真的很难注意到。6.4 Webhook请求失败导致进程卡死前面提过给requests.post设置了超时时间但还有另一个坑requests库在默认情况下不会自动处理重试如果Webhook接口刚好在告警高峰期不可用消息就会丢失。我的策略是发送Webhook失败后重试两次第二次重试在30秒后。如果两次都失败就把告警消息追加到本地文件alert_pending.log。后续手动处理或者写一个配套脚本重放即可至少数据不会丢。从工程角度讲告警系统应该是“尽力送达”而非“完美送达”能在极端情况下保证告警消息不丢失已经达到了生产可用标准。6.5 多服务器部署与集中管理的取舍这套系统默认是每台服务器独立部署一套告警各自发送。如果你管理的服务器数量多了比如超过十台这种模式就会变得混乱——同一故障可能在多台服务器上都触发告警运维人员会同时收到多条内容相似的提醒。优化思路有两种方向一是部署一个集中的日志采集端把日志统一汇总后再匹配规则二是给每台部署的实例加上不同的hostname标签在Webhook发送时附带服务器标识方便接收方有一个全局视图。我实际采用的是第二种方案因为改动成本最低。配置文件中增加一个hostname字段在AlertSender组装消息时自动带上当前机器的hostname。这样告警内容里能看到是哪台服务器发出的消息问题定位效率提升明显。7. 实践经验分享与改进方向7.1 这套系统的局限性先说清楚这套方案的边界在哪里避免读者盲目复制到超大规模场景后踩大坑。它不适合做日志分析和检索。只能抓取和匹配不能存历史日志内容不能画趋势图也不能做多条件复合查询。要做这些还是得考虑ELK或者ClickHouse这类存储分析组件。它不适合监控大量日志文件。目前我对单目录内文件数量的建议是不超过100个如果超过这个量级watchdog的事件处理可能产生明显的延迟文件位置管理也会变得复杂。它不适合做复杂的告警聚合和降噪。同一条告警只能做整条去重不能做到“相似告警自动聚合为一条”。如果你的场景需要应对大量相似但不完全一样的错误日志可能需要引入更成熟的告警管理平台。但这并不意味着这套系统没有价值。它的定位是“轻量级补位工具”用极低的成本填补现有监控体系的空白比如云平台自带的告警只覆盖了云服务状态应用层的异常日志它管不到。这就给这套脚本留了很大的发挥空间。7.2 后续可以扩展的功能代码跑稳定之后我对这套系统做了几项小扩展都取得了不错的效果新增了基于时间窗口的异常计数告警。有些故障不是偶发单条日志就能判断的可能是某个时间段内错误数量异常飙升比如正常每小时错误日志只有10条突然某小时暴涨到200条。这个扩展就是在原有匹配逻辑上加一个时间窗口计数超过预设基数就触发告警。增加了静态规则文件加载。把规则独立成JSON文件后运维同事不需要看代码只要在syslog_patterns.json里添加匹配规则重启进程就能生效。这对团队协作很友好降低了使用门槛。支持对接外部通知平台。邮件和Webhook之外我加了一个syslog转发通道把告警条目直接转发到中心日志系统方便做统一归档和审计。7.3 对维护方式的一点个人体会运行了大半年我认为这类轻量级工具最重要的不是功能多么完善而是可靠性足够强、排查问题足够方便。它不应该成为运维体系里最复杂的一环相反它应该像水龙头一样打开就能用坏了好修。因此我会刻意控制这个项目的代码复杂度和依赖数量不求大而全只求稳而精。每次有新的监控需求先问自己这个问题是不是用一个简单规则就能解决如果可以就加一条规则如果不行再看是否需要引入外部组件。这套“克制”的维护方式在我看来比什么都重要。如果你也想在自己的服务器上搭建这么一套轻量级日志监控系统照着上面的代码和思路落地即可。遇到跟我不一样的环境和需求就在规则引擎和通知方式上做调整很快就能稳定运行起来。