监控日志告警的脚本如何编写

本文目录导读：

1. 基础Shell脚本方案
2. Python方案（推荐）
3. 高级监控脚本
4. 使用systemd服务管理
5. 实用示例
6. 使用建议

基础Shell脚本方案

实时监控tail + grep

#!/bin/bash
# 监控日志文件
LOG_FILE="/var/log/application.log"
KEYWORDS=("ERROR" "FATAL" "Exception")
ALERT_EMAIL="admin@example.com"
# 使用tail -F持续监控
tail -F $LOG_FILE | while read line; do
    for keyword in "${KEYWORDS[@]}"; do
        if echo "$line" | grep -iq "$keyword"; then
            # 发送告警
            echo "$(date): 发现关键字: $keyword" | mail -s "日志告警" $ALERT_EMAIL
            echo "告警: $line"
        fi
    done
done

定时检查模式

#!/bin/bash
LOG_FILE="/var/log/application.log"
ALERT_EMAIL="admin@example.com"
CHECK_INTERVAL=60  # 检查间隔(秒)
while true; do
    # 检查最近1分钟的错误日志
    RECENT_LOGS=$(tail -n 100 $LOG_FILE | grep "ERROR")
    if [ -n "$RECENT_LOGS" ]; then
        # 发送告警邮件
        echo "$RECENT_LOGS" | mail -s "$(hostname) 日志告警" $ALERT_EMAIL
        # 或写入告警文件
        echo "$(date): 发现错误日志" >> /var/log/alert_history.log
    fi
    sleep $CHECK_INTERVAL
done

Python方案（推荐）

基础监控脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time
import re
import argparse
import logging
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import requests
import subprocess
import os
class LogMonitor:
    def __init__(self, log_file, keywords, alert_method='email'):
        self.log_file = log_file
        self.keywords = keywords
        self.alert_method = alert_method
        self.last_position = 0
        self.setup_logging()
    def setup_logging(self):
        """设置日志记录"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    def follow(self):
        """实时跟踪日志文件"""
        try:
            with open(self.log_file, 'r') as f:
                # 移动到文件末尾
                f.seek(0, 2)
                while True:
                    line = f.readline()
                    if line:
                        self.process_line(line.strip())
                    else:
                        time.sleep(0.1)
        except FileNotFoundError:
            self.logger.error(f"日志文件不存在: {self.log_file}")
        except Exception as e:
            self.logger.error(f"监控错误: {e}")
    def process_line(self, line):
        """处理每行日志"""
        for keyword in self.keywords:
            if re.search(keyword, line, re.IGNORECASE):
                self.trigger_alert(f"发现关键字: {keyword}\n日志内容: {line}")
    def trigger_alert(self, message):
        """触发告警"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        alert_msg = f"[{timestamp}] {message}"
        self.logger.warning(alert_msg)
        if self.alert_method == 'email':
            self.send_email_alert(alert_msg)
        elif self.alert_method == 'webhook':
            self.send_webhook_alert(alert_msg)
        elif self.alert_method == 'command':
            self.run_alert_command(alert_msg)
    def send_email_alert(self, message):
        """发送邮件告警"""
        msg = MIMEText(message)
        msg['Subject'] = f'日志告警 - {self.log_file}'
        msg['From'] = 'monitor@example.com'
        msg['To'] = 'admin@example.com'
        try:
            with smtplib.SMTP('localhost') as server:
                server.send_message(msg)
        except Exception as e:
            self.logger.error(f"发送邮件失败: {e}")
    def send_webhook_alert(self, message):
        """发送Webhook告警"""
        webhook_url = "https://hooks.slack.com/services/xxx/xxx/xxx"
        payload = {
            "text": message,
            "username": "LogMonitor",
            "icon_emoji": ":warning:"
        }
        try:
            response = requests.post(webhook_url, json=payload)
            response.raise_for_status()
        except Exception as e:
            self.logger.error(f"发送Webhook失败: {e}")
    def run_alert_command(self, message):
        """执行命令式告警"""
        command = f'echo "{message}" | wall'  # 发送到所有终端
        try:
            subprocess.run(command, shell=True)
        except Exception as e:
            self.logger.error(f"执行命令失败: {e}")
    def check_rate_limit(self, keyword, interval=300):
        """检查频率限制"""
        # 实现在指定时间间隔内不重复告警
        pass
def main():
    parser = argparse.ArgumentParser(description='日志监控告警工具')
    parser.add_argument('--file', '-f', required=True, help='要监控的日志文件')
    parser.add_argument('--keywords', '-k', required=True, help='关键字列表(逗号分隔)')
    parser.add_argument('--method', '-m', default='email', 
                        choices=['email', 'webhook', 'command', 'stdout'],
                        help='告警方式')
    args = parser.parse_args()
    keywords = [k.strip() for k in args.keywords.split(',')]
    monitor = LogMonitor(args.file, keywords, args.method)
    monitor.follow()
if __name__ == "__main__":
    main()

高级监控脚本

带状态跟踪和阈值告警

#!/usr/bin/env python3
import time
import hashlib
import json
from collections import defaultdict
from datetime import datetime, timedelta
class AdvancedLogMonitor:
    def __init__(self):
        self.state_file = "/tmp/monitor_state.json"
        self.error_counts = defaultdict(int)
        self.last_alert_time = {}
        self.load_state()
    def load_state(self):
        """加载监控状态"""
        try:
            with open(self.state_file, 'r') as f:
                data = json.load(f)
                self.last_alert_time = data.get('last_alert_time', {})
        except:
            pass
    def save_state(self):
        """保存监控状态"""
        with open(self.state_file, 'w') as f:
            json.dump({
                'last_alert_time': self.last_alert_time
            }, f)
    def check_threshold(self, error_type, count, threshold=5, window=300):
        """
        检查是否达到告警阈值
        :param error_type: 错误类型
        :param count: 当前计数
        :param threshold: 阈值
        :param window: 时间窗口(秒)
        """
        now = time.time()
        # 检查是否在冷静期
        if error_type in self.last_alert_time:
            last_time = self.last_alert_time[error_type]
            if now - last_time < 300:  # 5分钟冷静期
                return False
        # 检查阈值
        if count >= threshold:
            self.last_alert_time[error_type] = now
            self.save_state()
            return True
        return False
    def analyze_log_pattern(self, log_line):
        """
        分析日志模式
        :param log_line: 日志行
        :return: 错误类型或None
        """
        patterns = {
            'database': ['ORA-', 'MySQL', 'connection refused', 'timeout'],
            'application': ['NullPointer', 'ArrayIndexOutOfBounds', 'StackOverflow'],
            'system': ['OutOfMemory', 'Disk full', 'CPU overload'],
            'security': ['Unauthorized', 'Access denied', 'Invalid token']
        }
        for error_type, keywords in patterns.items():
            for keyword in keywords:
                if keyword.lower() in log_line.lower():
                    return error_type
        return None
    def aggregate_errors(self, log_file, time_window=60):
        """
        聚合错误统计
        :param log_file: 日志文件
        :param time_window: 时间窗口(秒)
        """
        error_pattern = re.compile(r'(ERROR|FATAL|CRITICAL):\s*(.*)')
        cut_off = datetime.now() - timedelta(seconds=time_window)
        try:
            with open(log_file, 'r') as f:
                for line in f:
                    # 解析时间戳
                    timestamp_match = re.match(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})', line)
                    if timestamp_match:
                        log_time = datetime.strptime(timestamp_match.group(1), 
                                                    "%Y-%m-%d %H:%M:%S")
                        if log_time > cut_off:
                            error_match = error_pattern.search(line)
                            if error_match:
                                error_type = self.analyze_log_pattern(line)
                                if error_type:
                                    self.error_counts[error_type] += 1
        except FileNotFoundError:
            print(f"日志文件不存在: {log_file}")
    def monitor_loop(self, log_file, check_interval=60):
        """
        主监控循环
        :param log_file: 日志文件
        :param check_interval: 检查间隔(秒)
        """
        print(f"开始监控日志文件: {log_file}")
        print(f"检查间隔: {check_interval}秒")
        while True:
            self.error_counts.clear()
            self.aggregate_errors(log_file)
            for error_type, count in self.error_counts.items():
                if self.check_threshold(error_type, count):
                    alert_msg = f"[告警] {error_type}错误频繁: {count}次/分钟"
                    print(alert_msg)
                    # 发送告警
                    self.send_alert(alert_msg)
            time.sleep(check_interval)
    def send_alert(self, message):
        """发送告警"""
        # 多种告警方式
        self.send_wechat_alert(message)
        self.send_email_alert(message)
        self.log_alert(message)
    def send_wechat_alert(self, message):
        """发送微信告警"""
        # 实现企业微信机器人告警
        pass
    def send_email_alert(self, message):
        """发送邮件告警"""
        # 实现邮件告警
        pass
    def log_alert(self, message):
        """记录告警日志"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open("/var/log/monitor_alerts.log", "a") as f:
            f.write(f"[{timestamp}] {message}\n")
# 监控多个日志文件
def monitor_multiple_logs(log_files):
    """监控多个日志文件"""
    import threading
    threads = []
    for log_file in log_files:
        monitor = AdvancedLogMonitor()
        thread = threading.Thread(target=monitor.monitor_loop, args=(log_file,))
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()
if __name__ == "__main__":
    monitor = AdvancedLogMonitor()
    monitor.monitor_loop("/var/log/application.log")

使用systemd服务管理

创建systemd服务文件

# /etc/systemd/system/log-monitor.service
[Unit]
Description=Log Monitor Service
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/local/bin/log_monitor.py --file /var/log/application.log --keywords ERROR,FATAL --method email
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

启动服务

# 重载systemd配置
sudo systemctl daemon-reload
# 启动服务
sudo systemctl start log-monitor
# 设置开机自启
sudo systemctl enable log-monitor
# 查看状态
sudo systemctl status log-monitor

实用示例

简单监控脚本(生产可用)

#!/bin/bash
# 生产环境日志监控脚本
LOG_DIR="/var/log/applications"
ALERT_EMAIL="ops@company.com"
SLACK_WEBHOOK="https://hooks.slack.com/services/xxx"
# 监控API错误
monitor_api_errors() {
    for log_file in $LOG_DIR/*/api.log; do
        # 查找5分钟内的错误
        errors=$(tail -n 1000 "$log_file" | grep "HTTP 5[0-9][0-9]" | tail -20)
        if [ -n "$errors" ]; then
            # 构建告警消息
            message="API错误告警 - $(hostname)\n服务: $(basename $(dirname $log_file))\n时间: $(date)\n错误:\n$errors"
            # 发送告警
            curl -X POST -H 'Content-type: application/json' \
                --data "{\"text\":\"$message\"}" \
                $SLACK_WEBHOOK
        fi
    done
}
# 监控磁盘使用
monitor_disk_usage() {
    threshold=90
    df -h | grep -v Filesystem | awk '{print $5, $6}' | while read output; do
        usage=$(echo $output | awk '{print $1}' | sed 's/%//')
        partition=$(echo $output | awk '{print $2}')
        if [ $usage -ge $threshold ]; then
            # 记录到系统日志
            logger "磁盘使用告警: $partition 使用率 $usage%"
            # 发送告警
            mail -s "磁盘告警: $partition 使用率 $usage%" $ALERT_EMAIL
        fi
    done
}
# 监控进程状态
monitor_process() {
    processes=("nginx" "mysql" "redis" "java")
    for proc in "${processes[@]}"; do
        if ! pgrep -x "$proc" > /dev/null; then
            message="进程告警: $proc 未运行"
            echo "$message" | mail -s "进程告警" $ALERT_EMAIL
            # 尝试重启
            systemctl restart "$proc" 2>/dev/null || service "$proc" restart 2>/dev/null
        fi
    done
}
# 主循环
while true; do
    monitor_api_errors
    monitor_disk_usage
    monitor_process
    sleep 60
done

使用建议

1. 选择合适的告警级别：ERROR/FATAL直接告警，WARNING可以汇总
2. 设置告警去重：相同的错误不要重复告警
3. 实施冷静期：同一错误至少间隔5分钟再告警
4. 分级告警：重要服务直接电话通知，普通错误邮件通知
5. 记录告警历史：便于事后分析和优化
6. 测试告警通道：定期验证告警是否正常送达

这些脚本可以根据实际需求修改,建议先在测试环境验证后再部署到生产环境。