Python实现轻量级日志监控告警系统
1. 项目概述
日志监控是系统运维中最基础也最重要的环节之一。记得去年我们线上服务突然出现大面积超时，排查了半天才发现是磁盘空间被日志文件占满。如果当时有个实时监控系统，就能提前收到预警避免事故。这就是为什么我决定用Python开发一个轻量级的日志监控告警系统。
这个方案特别适合中小型团队——不需要搭建复杂的ELK或Prometheus，用Python几十行代码就能实现核心功能。它能实时扫描系统日志，匹配预设的关键词（如"error"、"exception"）、异常模式（如5xx状态码激增）或正则表达式，一旦发现问题立即通过邮件、Slack或Webhook发送告警。
2. 核心设计思路
2.1 技术选型对比
常见的日志监控方案有：
- ELK Stack：功能全面但资源消耗大
- Prometheus+Grafana：适合指标监控，日志分析较弱
- Splunk：商业方案成本高
- Python脚本：轻量灵活，可定制性强
选择Python主要考虑：
- 开发效率高（相比Shell更易维护）
- 丰富的日志处理库（如watchdog、pygtail）
- 多协议告警支持（SMTP/HTTP等）
- 跨平台兼容性（Win/Linux/macOS）
2.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_RULES=error;exception;5\d{2} SLACK_WEBHOOK=https://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(hours=1) ] # 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_before=5, lines_after=5): 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(maxsize=1000) # 在工作线程中批量处理 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(pattern=rule).inc()配合简单的自愈机制：
def auto_heal(pattern): if pattern == "OutOfMemory": os.system("systemctl restart myapp") send_alert(f"已自动重启服务应对内存泄漏")7. 完整部署方案
建议用systemd管理进程：
# /etc/systemd/system/logmon.service [Unit] Description=Log Monitor After=network.target [Service] User=logmon WorkingDirectory=/opt/logmon ExecStart=/usr/bin/python3 /opt/logmon/main.py Restart=always EnvironmentFile=/opt/logmon/.env [Install] WantedBy=multi-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/就实现了需求。