2. 日志文件监控与告警

任务目标

编写一个Python脚本，实时监控指定日志文件的新增内容，当检测到包含错误关键词（如 ERROR、Traceback）的行时，自动发送邮件告警。脚本需支持持续运行，并正确处理日志轮转场景。

环境依赖

仅使用Python标准库，无需安装第三方包。邮件发送基于smtplib与email模块，文件监控使用os与time实现简易的轮询读取。

完整实现代码

import os
import time
import smtplib
from email.mime.text import MIMEText
from email.header import Header

# ========== 配置区 ==========
LOG_FILE = "/var/log/myapp.log"          # 被监控的日志文件
KEYWORDS = ["ERROR", "Traceback"]        # 错误关键词列表
CHECK_INTERVAL = 2                       # 轮询间隔（秒）

# 邮件配置
SMTP_SERVER = "smtp.example.com"
SMTP_PORT = 465
SENDER_EMAIL = "monitor@example.com"
SENDER_PASSWORD = "your_password"
RECEIVER_EMAIL = "admin@example.com"
# ===========================

def send_alert(line, filename):
    """发送告警邮件，包含错误行内容与来源文件"""
    subject = f"[告警] {filename} 出现错误"
    body = f"检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n文件: {filename}\n错误内容: {line.strip()}"
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = Header(subject, "utf-8")
    msg["From"] = SENDER_EMAIL
    msg["To"] = RECEIVER_EMAIL

    try:
        with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:
            server.login(SENDER_EMAIL, SENDER_PASSWORD)
            server.sendmail(SENDER_EMAIL, [RECEIVER_EMAIL], msg.as_string())
        print(f"[{time.strftime('%H:%M:%S')}] 告警邮件已发送: {line.strip()[:50]}")
    except Exception as e:
        print(f"[{time.strftime('%H:%M:%S')}] 邮件发送失败: {e}")

def follow_file(filename):
    """生成器：持续产出文件新增行，支持简单日志轮转"""
    with open(filename, "r", encoding="utf-8", errors="ignore") as f:
        f.seek(0, os.SEEK_END)  # 从文件末尾开始读取
        while True:
            line = f.readline()
            if line:
                yield line
            else:
                # 检测文件是否被轮转（inode变化）
                try:
                    if os.stat(filename).st_ino != os.fstat(f.fileno()).st_ino:
                        f.close()
                        f = open(filename, "r", encoding="utf-8", errors="ignore")
                        continue
                except FileNotFoundError:
                    # 文件暂时被删除，等待重建
                    time.sleep(CHECK_INTERVAL)
                    continue
                time.sleep(CHECK_INTERVAL)

def main():
    if not os.path.exists(LOG_FILE):
        print(f"错误: 日志文件不存在 {LOG_FILE}")
        return

    print(f"开始监控 {LOG_FILE}，关键词: {KEYWORDS}")
    for line in follow_file(LOG_FILE):
        if any(kw in line for kw in KEYWORDS):
            send_alert(line, LOG_FILE)

if __name__ == "__main__":
    main()

运行方式

  1. 将上述代码保存为 log_monitor.py。
  2. 修改代码顶部配置区中的日志文件路径、关键词、SMTP服务器信息及收发件邮箱。
  3. 在终端执行：python3 log_monitor.py

验证方法

按以下步骤确认脚本功能正常：

  1. 基础验证：在另一个终端向被监控的日志文件追加一行包含 ERROR 的测试内容，例如 echo "test ERROR line" >> /var/log/myapp.log。观察脚本终端应输出“告警邮件已发送”的提示，同时收件邮箱收到告警邮件。
  2. 轮转验证：执行 mv /var/log/myapp.log /var/log/myapp.log.old 模拟日志轮转，然后新建原文件并追加错误行，脚本应能自动恢复监控并继续告警。
  3. 判定通过条件：脚本持续运行不退出；每次追加含关键词的行后，均能触发邮件发送；不含关键词的行不触发任何动作。

常见问题

  • 邮件发送失败：确认SMTP服务器端口与加密方式匹配（SSL/STARTTLS），部分邮箱需开启“允许SMTP”或使用授权码而非登录密码。
  • 中文乱码：代码已使用utf-8编码处理，若日志文件为其他编码，请调整open()中的encoding参数。
  • 监控延迟：CHECK_INTERVAL控制轮询频率，设为0.5可降低延迟，但会增加CPU占用。
← 上一章 📖 返回目录 下一章 →