用 Python 写一个日志实时监控小工具：tail -f 跟随与关键词告警

一、为什么要自己写一个日志监控小工具

线上服务出问题时，第一反应往往是 tail -f 看日志。但人工盯着屏幕既累又容易漏掉关键报错：ERROR 一闪而过，等你反应过来已经刷出几十屏。

能不能让脚本替我们“盯”着日志，出现关键词（比如 ERROR、Traceback、OutOfMemory）就立刻高亮并告警？今天我们就用 Python 标准库写一个百行左右的小工具 logwatch.py，实现：

• 像 tail -f 一样实时跟随日志增长；
• 支持关键词列表匹配，命中即告警；
• 能感知日志轮转（logrotate 之后不丢数据）；
• 命令行参数开箱即用。

二、核心思路

1. 跟随增长：打开文件后 seek 到末尾，循环读取；读不到新内容就 sleep 一小会儿。
2. 检测轮转：每次循环检查文件的 inode 是否变化，变化说明日志被轮转，重新打开文件。
3. 关键词告警：逐行检查是否包含关键词，命中则输出带颜色的告警行，并可选择写入告警文件。

三、完整代码

#!/usr/bin/env python3
"""logwatch.py - 实时跟随日志并做关键词告警"""
import argparse
import os
import sys
import time

RED = "\033[31m"
YELLOW = "\033[33m"
RESET = "\033[0m"


def parse_args():
    p = argparse.ArgumentParser(description="实时监控日志文件，关键词命中即告警")
    p.add_argument("logfile", help="要监控的日志文件路径")
    p.add_argument("-k", "--keywords", nargs="+",
                   default=["ERROR", "Traceback"],
                   help="告警关键词列表")
    p.add_argument("-i", "--interval", type=float, default=0.5,
                   help="轮询间隔（秒）")
    p.add_argument("-o", "--output", help="告警写入文件（可选）")
    return p.parse_args()


def alert(line, keywords, out_fh):
    hit = [k for k in keywords if k in line]
    if not hit:
        return
    msg = f"{RED}[ALERT]{RESET} 命中关键词 {YELLOW}{','.join(hit)}{RESET} : {line.rstrip()}"
    print(msg, flush=True)
    if out_fh:
        out_fh.write(f"[ALERT] {','.join(hit)} : {line}")
        out_fh.flush()


def follow(path, keywords, interval, out_fh):
    fh = open(path, "r", encoding="utf-8", errors="replace")
    fh.seek(0, os.SEEK_END)  # 从末尾开始，只看新增内容
    inode = os.fstat(fh.fileno()).st_ino
    print(f"开始监控 {path}，关键词：{keywords}")
    try:
        while True:
            line = fh.readline()
            if line:
                alert(line, keywords, out_fh)
                continue
            # 检测日志轮转
            try:
                if os.stat(path).st_ino != inode:
                    print("检测到日志轮转，重新打开文件…")
                    fh.close()
                    fh = open(path, "r", encoding="utf-8", errors="replace")
                    inode = os.fstat(fh.fileno()).st_ino
            except FileNotFoundError:
                pass
            time.sleep(interval)
    except KeyboardInterrupt:
        print("\n停止监控")
    finally:
        fh.close()


def main():
    args = parse_args()
    if not os.path.isfile(args.logfile):
        sys.exit(f"文件不存在：{args.logfile}")
    out_fh = open(args.output, "a", encoding="utf-8") if args.output else None
    try:
        follow(args.logfile, args.keywords, args.interval, out_fh)
    finally:
        if out_fh:
            out_fh.close()


if __name__ == "__main__":
    main()

四、使用步骤

1. 保存并赋予执行权限

vim logwatch.py   # 粘贴上面的代码
chmod +x logwatch.py

2. 基本用法

# 监控 app.log，默认关键词 ERROR / Traceback
./logwatch.py /var/log/app.log

# 自定义关键词与轮询间隔
./logwatch.py /var/log/app.log -k ERROR 500 Timeout -i 1

# 告警同时写入文件，方便事后复盘
./logwatch.py /var/log/app.log -o alerts.txt

3. 验证效果

另开一个终端往日志里追加内容：

echo "INFO 服务启动" >> /var/log/app.log
echo "ERROR 数据库连接超时" >> /var/log/app.log

监控端会立刻输出一行红色 [ALERT] 告警，ERROR 行无所遁形。

4. 配合 logrotate

很多服务每天轮转日志（app.log → app.log.1）。本工具每次轮询都会比对 inode，检测到轮转就自动重新打开新文件，不会卡在旧文件上，也不会重复告警。

五、可以继续扩展的方向

• 告警去重：同一关键词 1 分钟内只报一次，避免刷屏；
• 多通道通知：命中后调用 webhook 把告警发到企业微信或钉钉；
• 正则匹配：把关键词升级为正则表达式，支持更复杂的模式；
• 多文件监控：用线程池同时跟随多个日志文件。

六、小结

这个小工具只用了 Python 标准库（argparse、os、time），没有第三方依赖，任何一台服务器都能直接跑。核心就是三招：seek 到末尾跟随、inode 比对感知轮转、逐行关键词匹配。把它放进你的运维工具箱，下次线上出问题，告警会第一时间跳到你眼前，而不是淹没在日志海洋里。