1. 项目概述在服务器运维和系统管理中日志监控是最基础也最重要的环节之一。传统的日志检查方式需要人工定期查看日志文件不仅效率低下而且无法及时发现突发问题。我在管理十几台生产服务器时就曾因为未能及时发现磁盘爆满的警告日志导致服务中断了2小时。这个Python日志监控脚本就是为了解决这类痛点而设计的。它能够实时监控系统日志文件当检测到预设的关键词如ERROR、CRITICAL、WARNING等时立即通过邮件或企业微信发送警报通知。相比专业的监控系统如Zabbix这个方案更轻量、更灵活特别适合中小型项目或临时性的监控需求。2. 核心功能设计2.1 日志文件实时监控日志监控的核心是实时捕获新产生的日志条目。Python的watchdog库提供了高效的文件系统事件监控功能from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogHandler(FileSystemEventHandler): def on_modified(self, event): if event.src_path /var/log/syslog: with open(event.src_path, r) as f: new_lines f.readlines()[-10:] # 读取最后10行新日志 analyze_logs(new_lines) observer Observer() observer.schedule(LogHandler(), path/var/log) observer.start()注意在生产环境中建议记录上次读取的位置如行号或文件指针而不是简单读取最后几行这样可以避免日志轮转时丢失内容。2.2 关键词匹配与过滤不是所有日志都需要触发警报我们需要定义关键词过滤规则KEYWORDS { ERROR: high, # 高优先级 OOM: critical, # 内存溢出 disk full: high, timeout: medium } def analyze_logs(log_lines): for line in log_lines: for keyword, level in KEYWORDS.items(): if keyword in line: send_alert(level, line) break # 一条日志可能匹配多个关键词只发送一次2.3 警报通知系统2.3.1 邮件通知使用SMTP发送邮件是最通用的通知方式import smtplib from email.mime.text import MIMEText def send_email(subject, content): msg MIMEText(content) msg[Subject] f[系统警报] {subject} msg[From] monitorexample.com msg[To] adminexample.com with smtplib.SMTP(smtp.example.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)2.3.2 企业微信机器人对于需要即时通知的场景企业微信机器人更合适import requests import json def send_wechat(content): webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour-key headers {Content-Type: application/json} data { msgtype: text, text: { content: content, mentioned_mobile_list: [13800001111] # 特定人员 } } requests.post(webhook_url, headersheaders, datajson.dumps(data))3. 高级功能实现3.1 日志上下文捕获当发现错误日志时通常需要查看前后的上下文才能完整理解问题def get_log_context(log_file, error_line_num, lines_before5, lines_after5): with open(log_file, r) as f: all_lines f.readlines() start max(0, error_line_num - lines_before - 1) end min(len(all_lines), error_line_num lines_after) return .join(all_lines[start:end])3.2 频率限制与告警合并避免短时间内重复发送相同错误的警报from collections import defaultdict from datetime import datetime, timedelta alert_history defaultdict(list) def should_send_alert(error_key): now datetime.now() # 过去1小时内相同错误不超过3次 recent_alerts [t for t in alert_history[error_key] if now - t timedelta(hours1)] alert_history[error_key].append(now) return len(recent_alerts) 33.3 多日志文件支持实际系统中可能需要监控多个日志文件# config.yaml log_files: - path: /var/log/syslog keywords: [ERROR, CRITICAL] - path: /var/log/nginx/error.log keywords: [500, 502, 503, 504] - path: /var/log/mysql/error.log keywords: [deadlock, timeout]4. 部署与优化4.1 系统服务化为了让脚本在后台持续运行可以创建systemd服务# /etc/systemd/system/log_monitor.service [Unit] DescriptionPython Log Monitor Afternetwork.target [Service] Userroot ExecStart/usr/bin/python3 /opt/log_monitor/main.py Restartalways [Install] WantedBymulti-user.target启用服务sudo systemctl daemon-reload sudo systemctl enable log_monitor sudo systemctl start log_monitor4.2 性能优化技巧使用inotify替代轮询watchdog默认使用轮询方式在Linux上可以改用inotifyobserver Observer(timeout1)异步发送通知使用多线程或异步IO避免阻塞主监控循环from threading import Thread Thread(targetsend_alert, args(level, message)).start()日志采样调试在开发阶段可以使用lograte限制日志生成速度进行测试sudo lograte -f /var/log/syslog -r 10/s5. 常见问题排查5.1 权限问题监控系统日志通常需要root权限sudo chmod 644 /var/log/syslog # 临时方案 # 更好的方案是将用户加入adm组 sudo usermod -aG adm your_user5.2 日志轮转处理当日志轮转时简单的文件修改监控可能会丢失内容。解决方案监控目录而非单个文件检查文件的inode变化使用日志服务的API如journalctl5.3 编码问题不同系统日志可能使用不同编码with open(log_file, r, encodingutf-8, errorsignore) as f: content f.read()6. 扩展思路这个基础框架可以进一步扩展为日志分析面板集成Flask/Django展示历史警报自动化修复对已知错误自动执行修复脚本机器学习异常检测使用PyOD等库识别异常模式多节点监控通过SSH或API收集多台服务器日志我在实际使用中发现对于Java应用的监控还需要特别注意堆栈跟踪的多行日志处理。一个实用的技巧是设置多行日志的正则表达式模式import re error_pattern re.compile(r^\d{4}-\d{2}-\d{2}.*(ERROR|Exception)) stack_trace False current_error [] for line in log_lines: if error_pattern.match(line): if current_error: # 发送上一个完整的错误 send_alert(\n.join(current_error)) current_error [] stack_trace True if stack_trace: current_error.append(line) if line.strip().endswith(}): # 假设以}结束堆栈 send_alert(\n.join(current_error)) current_error [] stack_trace False Python实现轻量级日志实时监控与告警系统
1. 项目概述在服务器运维和系统管理中日志监控是最基础也最重要的环节之一。传统的日志检查方式需要人工定期查看日志文件不仅效率低下而且无法及时发现突发问题。我在管理十几台生产服务器时就曾因为未能及时发现磁盘爆满的警告日志导致服务中断了2小时。这个Python日志监控脚本就是为了解决这类痛点而设计的。它能够实时监控系统日志文件当检测到预设的关键词如ERROR、CRITICAL、WARNING等时立即通过邮件或企业微信发送警报通知。相比专业的监控系统如Zabbix这个方案更轻量、更灵活特别适合中小型项目或临时性的监控需求。2. 核心功能设计2.1 日志文件实时监控日志监控的核心是实时捕获新产生的日志条目。Python的watchdog库提供了高效的文件系统事件监控功能from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogHandler(FileSystemEventHandler): def on_modified(self, event): if event.src_path /var/log/syslog: with open(event.src_path, r) as f: new_lines f.readlines()[-10:] # 读取最后10行新日志 analyze_logs(new_lines) observer Observer() observer.schedule(LogHandler(), path/var/log) observer.start()注意在生产环境中建议记录上次读取的位置如行号或文件指针而不是简单读取最后几行这样可以避免日志轮转时丢失内容。2.2 关键词匹配与过滤不是所有日志都需要触发警报我们需要定义关键词过滤规则KEYWORDS { ERROR: high, # 高优先级 OOM: critical, # 内存溢出 disk full: high, timeout: medium } def analyze_logs(log_lines): for line in log_lines: for keyword, level in KEYWORDS.items(): if keyword in line: send_alert(level, line) break # 一条日志可能匹配多个关键词只发送一次2.3 警报通知系统2.3.1 邮件通知使用SMTP发送邮件是最通用的通知方式import smtplib from email.mime.text import MIMEText def send_email(subject, content): msg MIMEText(content) msg[Subject] f[系统警报] {subject} msg[From] monitorexample.com msg[To] adminexample.com with smtplib.SMTP(smtp.example.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)2.3.2 企业微信机器人对于需要即时通知的场景企业微信机器人更合适import requests import json def send_wechat(content): webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour-key headers {Content-Type: application/json} data { msgtype: text, text: { content: content, mentioned_mobile_list: [13800001111] # 特定人员 } } requests.post(webhook_url, headersheaders, datajson.dumps(data))3. 高级功能实现3.1 日志上下文捕获当发现错误日志时通常需要查看前后的上下文才能完整理解问题def get_log_context(log_file, error_line_num, lines_before5, lines_after5): with open(log_file, r) as f: all_lines f.readlines() start max(0, error_line_num - lines_before - 1) end min(len(all_lines), error_line_num lines_after) return .join(all_lines[start:end])3.2 频率限制与告警合并避免短时间内重复发送相同错误的警报from collections import defaultdict from datetime import datetime, timedelta alert_history defaultdict(list) def should_send_alert(error_key): now datetime.now() # 过去1小时内相同错误不超过3次 recent_alerts [t for t in alert_history[error_key] if now - t timedelta(hours1)] alert_history[error_key].append(now) return len(recent_alerts) 33.3 多日志文件支持实际系统中可能需要监控多个日志文件# config.yaml log_files: - path: /var/log/syslog keywords: [ERROR, CRITICAL] - path: /var/log/nginx/error.log keywords: [500, 502, 503, 504] - path: /var/log/mysql/error.log keywords: [deadlock, timeout]4. 部署与优化4.1 系统服务化为了让脚本在后台持续运行可以创建systemd服务# /etc/systemd/system/log_monitor.service [Unit] DescriptionPython Log Monitor Afternetwork.target [Service] Userroot ExecStart/usr/bin/python3 /opt/log_monitor/main.py Restartalways [Install] WantedBymulti-user.target启用服务sudo systemctl daemon-reload sudo systemctl enable log_monitor sudo systemctl start log_monitor4.2 性能优化技巧使用inotify替代轮询watchdog默认使用轮询方式在Linux上可以改用inotifyobserver Observer(timeout1)异步发送通知使用多线程或异步IO避免阻塞主监控循环from threading import Thread Thread(targetsend_alert, args(level, message)).start()日志采样调试在开发阶段可以使用lograte限制日志生成速度进行测试sudo lograte -f /var/log/syslog -r 10/s5. 常见问题排查5.1 权限问题监控系统日志通常需要root权限sudo chmod 644 /var/log/syslog # 临时方案 # 更好的方案是将用户加入adm组 sudo usermod -aG adm your_user5.2 日志轮转处理当日志轮转时简单的文件修改监控可能会丢失内容。解决方案监控目录而非单个文件检查文件的inode变化使用日志服务的API如journalctl5.3 编码问题不同系统日志可能使用不同编码with open(log_file, r, encodingutf-8, errorsignore) as f: content f.read()6. 扩展思路这个基础框架可以进一步扩展为日志分析面板集成Flask/Django展示历史警报自动化修复对已知错误自动执行修复脚本机器学习异常检测使用PyOD等库识别异常模式多节点监控通过SSH或API收集多台服务器日志我在实际使用中发现对于Java应用的监控还需要特别注意堆栈跟踪的多行日志处理。一个实用的技巧是设置多行日志的正则表达式模式import re error_pattern re.compile(r^\d{4}-\d{2}-\d{2}.*(ERROR|Exception)) stack_trace False current_error [] for line in log_lines: if error_pattern.match(line): if current_error: # 发送上一个完整的错误 send_alert(\n.join(current_error)) current_error [] stack_trace True if stack_trace: current_error.append(line) if line.strip().endswith(}): # 假设以}结束堆栈 send_alert(\n.join(current_error)) current_error [] stack_trace False