还在手动 tail 日志？用 Python + systemd 做一个 Nginx 异常告警哨兵

sudo yum install -y epel-release
sudo yum install -y python3 python3-pip

vi /opt/nginx_log_monitor.p

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time
import json
import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime
from collections import deque
import requests

# ================== 配置区 ==================
NGINX_ERROR_LOG = "/var/log/nginx/error.log"
CHECK_INTERVAL = 10  # 每10秒检查一次
ALERT_THRESHOLD = 2  # 达到此数量触发告警，自定义

# 钉钉机器人配置（替换为你自己的）
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com......."
DINGTALK_SECRET = None  # 先关闭加签测试（设为字符串则启用加签）

# ===========================================

def get_sign_and_timestamp(secret):
    """生成钉钉加签所需的 timestamp 和 sign"""
    if not secret:
        return str(int(time.time() * 1000)), None
    timestamp = str(round(time.time() * 1000))
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign

def send_dingtalk_message(msg_text):
    """发送钉钉消息"""
    headers = {'Content-Type': 'application/json'}
    timestamp, sign = get_sign_and_timestamp(DINGTALK_SECRET)

    webhook_url = DINGTALK_WEBHOOK
    if sign:
        webhook_url += f"×tamp={timestamp}&sign={sign}"

    data = {
        "msgtype": "text",
        "text": {
            "content": msg_text
        }
    }

    print(f"[DEBUG] 发送钉钉请求 URL: {webhook_url}")
    try:
        resp = requests.post(webhook_url, headers=headers, data=json.dumps(data), timeout=10)
        result = resp.json()
        if result.get('errcode') == 0:
            print(f"[+] 钉钉消息发送成功: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        else:
            print(f"[-] 钉钉发送失败: errcode={result['errcode']}, errmsg={result['errmsg']}")
    except Exception as e:
        print(f"[-] 请求异常: {e}")

def is_error_line(line):
    """判断是否为 Nginx 错误日志行"""
    line_lower = line.lower()
    # 显式错误级别
    if any(level in line for level in ["[error]", "[crit]", "[alert]", "[emerg]"]):
        return True
    # 关键词匹配（覆盖无级别但含错误的行）
    error_keywords = [
        "failed", "denied", "not found", "no such file",
        "connection refused", "timeout", "permission denied",
        "upstream prematurely closed", "broken pipe"
    ]
    return any(kw in line_lower for kw in error_keywords)

def monitor_log():
    try:
        with open(NGINX_ERROR_LOG, "r", encoding='utf-8', errors='ignore') as f:
            f.seek(0, 2)  # 跳到文件末尾
            recent_errors = deque(maxlen=20)
            error_count = 0
            last_check_time = time.time()

            print(f"[INFO] 开始监控日志: {NGINX_ERROR_LOG}")
            print(f"[INFO] 告警阈值: {ALERT_THRESHOLD} 条错误 / {CHECK_INTERVAL} 秒")

            while True:
                line = f.readline()
                if line:
                    # [调试] 打印原始日志行（可注释掉）
                    print(f"[RAW LOG] {repr(line)}")
                    if is_error_line(line):
                        error_count += 1
                        recent_errors.append(line.strip())
                        print(f"[DEBUG] 检测到错误 ({error_count}): {line.strip()}")
                else:
                    current_time = time.time()
                    if current_time - last_check_time >= CHECK_INTERVAL:
                        print(f"[INFO] 检查周期结束 - 当前错误数: {error_count}")
                        if error_count >= ALERT_THRESHOLD:
                            content = (
                                f"🚨 Nginx 异常告警\n"
                                f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                                f"过去 {CHECK_INTERVAL} 秒内检测到 {error_count} 条错误日志：\n\n"
                                + "\n".join(list(recent_errors)[-5:])
                            )
                            send_dingtalk_message(content)
                            # 告警后重置
                            error_count = 0
                            recent_errors.clear()
                        # 注意：未达阈值时不重置，继续累积
                        last_check_time = current_time  # ← 已修复！

                    time.sleep(1)

    except FileNotFoundError:
        print(f"[-] 日志文件不存在: {NGINX_ERROR_LOG}")
    except KeyboardInterrupt:
        print("\n[!] 监控已停止")

if __name__ == "__main__":
    monitor_log()

chmod +x /opt/nginx_log_monitor.py

python3 /opt/nginx_log_monitor.py

echo '2026/05/27 11:20:01 [error] 123#0: *1 open() "/xxx" failed (2: No such file or directory)' | sudo tee -a /var/log/nginx/error.log
echo '2026/05/27 11:20:02 [error] 123#0: *2 connect() failed (111: Connection refused)' | sudo tee -a /var/log/nginx/error.log

sudo tee /etc/systemd/system/nginx-dingtalk-monitor.service <<EOF
[Unit]
Description=Nginx Error Log Monitor with DingTalk Alert
After=network.target nginx.service

[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /opt/nginx_log_monitor.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable nginx-dingtalk-monitor
sudo systemctl start nginx-dingtalk-monitor

sudo systemctl status nginx-dingtalk-monitor

sudo curl https://get.cpolar.sh | sh

sudo systemctl status cpolar

ssh -p 12178 root@2.tcp.cpolar.top