IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

监控系统日志的错误脚本如何写

wen 实用脚本 1

本文目录导读：

监控系统日志的错误脚本如何写

  1. 方案一：Shell 脚本（快速、轻量级，适合服务器巡检）
  2. 方案二：Python 脚本（更强大、支持正则、多种日志格式、告警通道）
  3. 方案三：结合 journalctl 监控 systemd 服务日志（Linux系统服务）

编写监控系统日志的错误脚本通常是为了实时或定期检测日志中的异常模式，并在发现错误时触发告警（如通知、邮件、执行修复命令）。

以下我会根据你的需求（是仅检测关键字？还是统计错误数？还是复杂的多模式匹配？）提供几个不同复杂度的脚本示例，主要使用 Shell + grep/awk（Linux标准环境）和 Python（更灵活、支持复杂逻辑）。

Shell 脚本（快速、轻量级，适合服务器巡检）

适用场景：检查单个日志文件，匹配特定的错误关键字（如 ERROR、FATAL、Exception）,并将结果输出或记录。

基础版本：检测最近1小时内是否有新错误

#!/bin/bash
# file: check_error.sh
LOG_FILE="/var/log/syslog"  # 改成你的日志路径
KEYWORDS="ERROR|FATAL|CRITICAL|OOM|PANIC"
LOG_DIR="/var/log/monitor"
ALERT_LOG="${LOG_DIR}/error_alert.log"
NOW=$(date +"%Y-%m-%d %H:%M:%S")
# 创建日志目录
mkdir -p ${LOG_DIR}
# 从日志中提取匹配关键字的行（只检查最近1小时生成的日志行）
# 使用 grep 的 --after-context 可以同时显示错误行前后的几行上下文
ERROR_LINES=$(grep -i -E "${KEYWORDS}" "${LOG_FILE}" 2>/dev/null | tail -100)
if [ -n "${ERROR_LINES}" ]; then
    echo "[${NOW}] 发现错误:" >> "${ALERT_LOG}"
    echo "${ERROR_LINES}" >> "${ALERT_LOG}"
    echo "---" >> "${ALERT_LOG}"
    # 可选：发送邮件或短信告警
    # echo "${ERROR_LINES}" | mail -s "系统日志错误告警" admin@example.com
fi
# 统计最近1小时错误数量（如果日志包含时间戳）
echo "[${NOW}] 当前错误记录：$(grep -c -i -E "${KEYWORDS}" "${LOG_FILE}")"

执行：通过 cron 每5分钟运行一次：

*/5 * * * * /path/to/check_error.sh > /dev/null 2>&1

改进版本：支持自动轮转日志文件，避免重复告警

#!/bin/bash
# 记录已检查的文件偏移量，只分析新内容
LAST_CHECK_FILE="/var/log/monitor/last_check_pos"
# 检查日志文件是否存在，获取当前大小
if [ ! -f "${LOG_FILE}" ]; then
    exit 0
fi
CURRENT_SIZE=$(stat -c%s "${LOG_FILE}" 2>/dev/null || echo 0)
# 读取上次检查的位置（如果文件不存在或已被截断则从0开始）
LAST_POS=0
if [ -f "${LAST_CHECK_FILE}" ]; then
    LAST_POS=$(cat "${LAST_CHECK_FILE}")
    if [ "${LAST_POS}" -gt "${CURRENT_SIZE}" ]; then
        LAST_POS=0  # 日志文件被轮转或截断
    fi
fi
# 用 dd 或 tail 读取新增内容
NEW_CONTENT=$(dd if="${LOG_FILE}" bs=1 skip="${LAST_POS}" 2>/dev/null)
if [ -n "${NEW_CONTENT}" ]; then
    # 检查新增内容中是否有错误
    echo "${NEW_CONTENT}" | grep -i -E "${KEYWORDS}" && {
        echo "$(date): 检测到新错误在 ${LOG_FILE}" >> "${ALERT_LOG}"
        # 执行告警动作
    }
    # 更新检查点
    echo "${CURRENT_SIZE}" > "${LAST_CHECK_FILE}"
fi

Python 脚本（更强大、支持正则、多种日志格式、告警通道）

适用场景：需要解析结构化日志（JSON格式）、跨多台机器、发送企业微信/短信、或进行更复杂的模式匹配。

示例：监控 nginx 错误日志，统计每分钟错误数量，触发告警

#!/usr/bin/env python3
# file: monitor_log_errors.py
import re
import time
import os
from collections import defaultdict, deque
# 配置
LOG_PATH = "/var/log/nginx/error.log"  # 或 syslog、应用日志
ALERT_THRESHOLD = 10  # 每分钟错误超过10条则告警
COOLDOWN_SECONDS = 300  # 两次告警间隔至少5分钟
# 可以匹配多种错误模式
ERROR_PATTERNS = [
    r'ERR[OR]{0,4}',
    r'FATAL|CRITICAL',
    r'Connection timed out',
    r'500|502|503',
    r'OutOfMemory',
    r'Traceback',
    # 可以自定义更多正则
]
compiled_patterns = [re.compile(p, re.IGNORECASE) for p in ERROR_PATTERNS]
# 错误计数滑动窗口（最近60秒）
error_deque = deque(maxlen=600)  # 最多存10分钟的数据（每秒可多条）
last_alert_time = time.time()
def get_current_log_size():
    try:
        return os.path.getsize(LOG_PATH)
    except FileNotFoundError:
        return 0
def check_errors_in_chunk(chunk):
    """检查一段文本中是否包含错误模式"""
    for pattern in compiled_patterns:
        if pattern.search(chunk):
            return True
    return False
def send_alert(message):
    """示例告警函数：打印到控制台，实际场景可改为发送邮件、钉钉、企业微信"""
    print(f"[ALERT] {time.strftime('%Y-%m-%d %H:%M:%S')}: {message}")
    # 例如发送企业微信机器人的 python 代码：
    # import requests
    # data = {"msgtype": "text", "text": {"content": message}}
    # requests.post("https://qyapi.weixin.qq.com/cgi-bin/webhook/...", json=data)
def main():
    global last_alert_time
    # 记录上次读取的文件位置
    last_pos = 0
    # 初始化：先将文件指针移动到末尾，避免重读整文件
    try:
        with open(LOG_PATH, 'r', errors='ignore') as f:
            f.seek(0, os.SEEK_END)
            last_pos = f.tell()
    except FileNotFoundError:
        print(f"日志文件 {LOG_PATH} 不存在，等待创建...")
    print("开始监控日志...")
    while True:
        try:
            current_size = get_current_log_size()
            if current_size < last_pos:
                # 文件被轮转或截断，重置
                last_pos = 0
                continue
            if current_size > last_pos:
                # 读取新增内容
                with open(LOG_PATH, 'r', errors='ignore') as f:
                    f.seek(last_pos)
                    new_data = f.read()
                    last_pos = f.tell()
                if new_data:
                    # 按行检查
                    lines = new_data.splitlines()
                    error_count_in_batch = 0
                    for line in lines:
                        if check_errors_in_chunk(line):
                            error_count_in_batch += 1
                            # 立即打印到控制台以便调试
                            print(f"检测到错误行: {line[:100]}")
                    # 将错误计入滑动窗口
                    now = time.time()
                    error_deque.extend([now] * error_count_in_batch)
                    # 检查最近60秒内的错误总数
                    cutoff = now - 60
                    while error_deque and error_deque[0] < cutoff:
                        error_deque.popleft()
                    recent_error_count = len(error_deque)
                    # 如果错误数超过阈值且距上次告警超过冷却期
                    if (recent_error_count >= ALERT_THRESHOLD 
                        and (now - last_alert_time) > COOLDOWN_SECONDS):
                        msg = (f"日志错误告警: 最近60秒内检测到 {recent_error_count} 条错误 "
                               f"(阈值: {ALERT_THRESHOLD}), 请检查 {LOG_PATH}")
                        send_alert(msg)
                        last_alert_time = now
            time.sleep(1)  # 每秒检查一次
        except KeyboardInterrupt:
            print("监控已停止")
            break
        except Exception as e:
            print(f"发生异常: {e}")
            time.sleep(5)
if __name__ == "__main__":
    main()

运行：

python3 monitor_log_errors.py

结合 journalctl 监控 systemd 服务日志（Linux系统服务）

如果你的服务由 systemd 管理，可以直接用 journalctl 更高效地获取错误：

#!/bin/bash
# 监控 systemd 服务 "myapp" 的最近错误（非实时）
SERVICE_NAME="myapp"
ERROR_COUNT=$(journalctl -u ${SERVICE_NAME} --since "1 minute ago" -p err | wc -l)
if [ "${ERROR_COUNT}" -gt 0 ]; then
    echo "检测到 ${ERROR_COUNT} 条错误，发送告警..."
    journalctl -u ${SERVICE_NAME} --since "1 minute ago" -p err 
fi
  1. 日志位置：确保脚本有权限读取日志文件（考虑 sudo 或 adm 组）。
  2. 性能：对于大日志文件，使用 seek/tail 而非从头全量 grep（尤其是 real-time 场景）。
  3. 错误定义：明确你的业务错误模式（关键字、正则、状态码、异常栈）。
  4. 告警疲劳：务必设置冷却时间或滑动窗口阈值,避免重复告警。
  5. 归档：日志轮转后,脚本应能正确处理文件被截断的场景。

如果你的需求更具体（比如针对某个特定应用的日志格式、需要发送到企业微信/钉钉、或需要分析结构化日志如 JSON），请补充细节,我可以给出更定制的脚本。

上一篇自动配置防火墙规则的脚本

下一篇Shell脚本如何实现子shell执行

相关文章

抱歉，评论功能暂时关闭!