Python实现轻量级日志监控告警系统

1. 项目概述日志监控是系统运维中最基础也最重要的环节之一。记得去年我们线上服务突然出现大面积超时排查了半天才发现是磁盘空间被日志文件占满。如果当时有个实时监控系统就能提前收到预警避免事故。这就是为什么我决定用Python开发一个轻量级的日志监控告警系统。这个方案特别适合中小型团队——不需要搭建复杂的ELK或Prometheus用Python几十行代码就能实现核心功能。它能实时扫描系统日志匹配预设的关键词如error、exception、异常模式如5xx状态码激增或正则表达式一旦发现问题立即通过邮件、Slack或Webhook发送告警。2. 核心设计思路2.1 技术选型对比常见的日志监控方案有ELK Stack功能全面但资源消耗大PrometheusGrafana适合指标监控日志分析较弱Splunk商业方案成本高Python脚本轻量灵活可定制性强选择Python主要考虑开发效率高相比Shell更易维护丰富的日志处理库如watchdog、pygtail多协议告警支持SMTP/HTTP等跨平台兼容性Win/Linux/macOS2.2 架构设计graph TD A[日志文件] -- B[Python监控进程] B -- C{规则匹配} C --|匹配成功| D[告警触发] C --|匹配失败| A D -- E[邮件/Slack/Webhook]实际实现时我推荐用watchdog监听文件变化比定期轮询更高效。核心处理流程通过inotify机制监听日志目录使用pygtail记录已读位置防重复处理多线程处理主线程监听工作线程分析异步发送告警避免阻塞3. 关键实现步骤3.1 环境准备安装依赖库pip install watchdog pygtail python-dotenv建议使用.env文件管理配置# .env示例 LOG_PATH/var/log/nginx/error.log ALERT_RULESerror;exception;5\d{2} SLACK_WEBHOOKhttps://hooks.slack.com/services/XXX3.2 日志监听实现from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from pygtail import Pygtail import re class LogHandler(FileSystemEventHandler): def __init__(self, rules): self.rules [re.compile(r) for r in rules.split(;)] def on_modified(self, event): for line in Pygtail(event.src_path): self.check_rules(line) def check_rules(self, line): for pattern in self.rules: if pattern.search(line): send_alert(f匹配到规则 {pattern.pattern}:\n{line}) def start_monitor(path, rules): event_handler LogHandler(rules) observer Observer() observer.schedule(event_handler, path) observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()3.3 告警模块实现邮件告警示例import smtplib from email.mime.text import MIMEText def send_email(subject, content): msg MIMEText(content) msg[Subject] subject msg[From] os.getenv(SMTP_USER) msg[To] os.getenv(ALERT_EMAIL) with smtplib.SMTP(os.getenv(SMTP_HOST), 587) as server: server.starttls() server.login(os.getenv(SMTP_USER), os.getenv(SMTP_PASS)) server.send_message(msg)Slack告警更简单import requests def send_slack(message): requests.post(os.getenv(SLACK_WEBHOOK), json{text: f[日志告警] {message}})4. 高级功能扩展4.1 频率限制与告警合并避免告警风暴from collections import defaultdict from datetime import datetime, timedelta alert_history defaultdict(list) def check_alert_rate(pattern, line): now datetime.now() # 清理1小时前的记录 alert_history[pattern] [ t for t in alert_history[pattern] if now - t timedelta(hours1) ] # 1小时内不超过5次 if len(alert_history[pattern]) 5: alert_history[pattern].append(now) return True return False4.2 日志上下文采集遇到错误时自动采集相关日志def capture_context(log_path, line_number, lines_before5, lines_after5): with open(log_path) as f: all_lines f.readlines() start max(0, line_number - lines_before - 1) end min(len(all_lines), line_number lines_after) return .join(all_lines[start:end])5. 生产环境注意事项权限问题确保Python进程有日志文件读取权限不要用root运行脚本建议创建专用账户性能优化# 使用队列缓冲日志处理 from queue import Queue log_queue Queue(maxsize1000) # 在工作线程中批量处理 def worker(): while True: batch [] while len(batch) 10: batch.append(log_queue.get()) process_batch(batch)异常处理日志文件被rotate时的处理网络中断时告警重试机制使用try-except捕获所有可能的异常6. 监控指标与自愈可以在脚本中暴露metrics接口from prometheus_client import start_http_server, Counter alert_counter Counter(log_alerts_total, Total alert count, [rule]) # 在触发告警时 alert_counter.labels(patternrule).inc()配合简单的自愈机制def auto_heal(pattern): if pattern OutOfMemory: os.system(systemctl restart myapp) send_alert(f已自动重启服务应对内存泄漏)7. 完整部署方案建议用systemd管理进程# /etc/systemd/system/logmon.service [Unit] DescriptionLog Monitor Afternetwork.target [Service] Userlogmon WorkingDirectory/opt/logmon ExecStart/usr/bin/python3 /opt/logmon/main.py Restartalways EnvironmentFile/opt/logmon/.env [Install] WantedBymulti-user.target日志轮转配置示例# /etc/logrotate.d/logmon /var/log/logmon.log { daily rotate 7 missingok notifempty compress postrotate systemctl restart logmon endrotate }8. 常见问题排查问题1监控进程CPU占用高检查是否误用了tail -f轮询使用strace -p PID查看系统调用确认watchdog版本2.1.0早期版本有性能问题问题2漏报日志检查pygtail的offset文件权限确认inotify的watch数量没超限/proc/sys/fs/inotify/max_user_watches问题3告警延迟改用异步IO如asyncio检查网络连接特别是Slack API这套系统在我们生产环境稳定运行了8个月日均处理20GB日志成功预警了37次潜在故障。最惊喜的是它的灵活性——上周业务部门临时需要监控支付异常我们只加了条正则规则/FAILED_PAYMENT|TIMEOUT/就实现了需求。