前言

接口已经开始返回502，页面上的静态资源也加载失败，但服务器看起来仍然在正常运行。直到用户发来消息，才登录服务器执行tail和grep，从大量日志中寻找真正的错误。

Nginx通常会把文件不存在、权限不足、连接被拒绝和上游超时等问题记录到错误日志中。问题并不是没有线索，而是这些线索只有在人工查看时才会被发现。

如果让脚本持续读取error.log，并在短时间内发现多条异常后主动发送通知，日志就可以从事后排查工具，变成故障发生时的提醒入口。

本文将在CentOS 7中编写一个Python监控脚本，根据Nginx错误级别和关键词识别异常，并通过钉钉机器人将错误时间及日志内容发送到群聊。

脚本测试完成后，再使用systemd将其设置为后台服务，并通过cpolar配置SSH远程入口，方便在外部网络中登录服务器检查和维护这套监控程序。

1.本文有什么作用？

解放人力，告别手动查日志

• 不再需要登录服务器、反复执行tail或grep命令查看Nginx错误日志。系统自动监控，省时省力，尤其适合长期运行的服务或无人值守环境。

实时发现异常，快速响应故障

• 一旦Nginx出现高频错误（如502 Bad Gateway、403权限拒绝、文件未找到等），脚本会立即识别并触发告警，帮助你在用户投诉前就发现问题，极大缩短MTTR（平均修复时间）。

降低运维门槛，提升系统可靠性

• 即使不是专业运维人员，也能通过本文提供的完整步骤，在CentOS 7上快速部署一套“智能日志哨兵”系统，让个人网站、博客、NAS服务等更稳定可靠。

提供可扩展的自动化框架

• 文章不仅给出可用的Python脚本，还涵盖：日志解析逻辑、邮件告警集成、systemd服务化部署，你可以在此基础上轻松扩展，比如接入钉钉通知、自动封禁恶意IP、对接监控等。

适用于真实生产/家庭场景

• 无论是企业小规模服务，还是家庭服务器、树莓派、香橙派上的自建站点，只要用了Nginx，这套方案都能立刻带来价值——让错误“主动找你”，而不是你“到处找错误”。

本文教你用Python + CentOS 7实现Nginx错误日志的自动监控与实时告警，把被动排查变为主动防御，让运维更轻松、服务更健壮。

2.前提条件

• 系统：CentOS 7
• 已安装并运行Nginx
• Nginx错误日志路径（默认）：/var/log/nginx/error.log
• 已安装Python 3（CentOS 7默认是Python 2，需手动安装）

如果未安装Python 3，请先执行：

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

3.获取钉钉Webhook和Secret

打开钉钉群 → 点击右上角设置：

找到智能群助手 → 添加机器人：

添加自定义机器人：

点击添加：

给机器人起个名字，我这里是“nginx告警”：

设置发消息关键词，因为现在钉钉对安全严格，所以需要设置限制，，也可以设置加签或者IP地址：

如果启用了 加签，还会看到一个 密钥（secret），后续脚本中会用到。

点击完成后，复制生成的Webhook，留着备用：

安全提示：不要将token和secret 泄露在公开代码中！

4.编写Python监控脚本

创建脚本文件：/opt/nginx_log_monitor.py

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

5.测试Python监控脚本

python3 /opt/nginx_log_monitor.py

模拟报错，在另一个终端追加 2 条错误日志：

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

成功检测到错误：

当ALERT_THRESHOLD = 2（也就是错误数=2时），发送钉钉告警：

6.设置为系统服务

创建systemd服务：

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

不想手动查日志？用Python自动分析Nginx错误日志，异常请求立刻通过钉钉告警通知你！再搭配cpolar内网穿透，即使服务部署在本地或私有网络，也能安全暴露SSH端口，随时随地远程接入、调试和管理你的监控系统——内网变公网，运维更高效。

7.安装cpolar实现随时随地开发

7.1 什么是cpolar？

cpolar是一款安全高效的内网穿透工具，无需公网IP或复杂配置，只需一条命令，即可将本地服务器、Web服务或任意端口映射到公网，让你随时随地远程访问内网应用，特别适合开发调试、远程运维和应急部署等场景。

7.2 部署cpolar

cpolar 可以将你本地电脑中的服务（如 SSH、Web、数据库）映射到公网。即使你在家里或外出时，也可以通过公网地址连接回本地运行的开发环境。

❤️以下是安装cpolar步骤：

使用一键脚本安装命令：

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

安装完成后，执行下方命令查看cpolar服务状态：（如图所示即为正常启动）

sudo systemctl status cpolar

Cpolar安装和成功启动服务后，在浏览器上输入虚拟机主机IP加9200端口即:【http://ip:9200】访问Cpolar管理界面，使用Cpolar官网注册的账号登录,登录后即可看到cpolar web 配置界面,接下来在web 界面配置即可：

打开浏览器访问本地9200端口，使用cpolar账户密码登录即可,登录后即可对隧道进行管理。

8.配置公网地址

通过配置，你可以在本地WSL或Linux系统上运行SSH服务，并通过Cpolar将其映射到公网，从而实现从任意设备远程连接开发环境的目的。

• 隧道名称：可自定义，本例使用了:ssh，注意不要与已有的隧道名称重复
• 协议：tcp
• 本地地址：22
• 端口类型：随机临时TCP端口
• 地区：China Top

创建成功后，打开左侧在线隧道列表,可以看到刚刚通过创建隧道生成了公网地址，接下来就可以在其他电脑或者移动端设备（异地）上，使用任意一个地址在终端中访问即可。

• tcp 表示使用的协议类型

• 2.tcp.cpolar.top是 Cpolar 提供的域名

• 12178是随机分配的公网端口号

通过Cpolar提供的公网地址和端口，就可以进行远程部署啦！

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

9.保留固定TCP公网地址

使用cpolar为其配置TCP地址，该地址为固定地址，不会随机变化。

选择区域和描述：有一个下拉菜单，当前选择的是“China VIP”。
右侧输入框，用于填写描述信息。
保留按钮：在右侧有一个橙色的“保留”按钮，点击该按钮可以保留所选的TCP地址。
列表中显示了一条已保留的TCP地址记录。

• 地区：显示为“China Top”。

• 地址：显示为“ 16.tcp.cpolar.top:14775”。

登录cpolar web UI管理界面，点击左侧仪表盘的隧道管理——隧道列表，找到所要配置的隧道ssh，点击右侧的编辑。

修改隧道信息，将保留成功的TCP端口配置到隧道中。

• 端口类型：选择固定TCP端口
• 预留的TCP地址：填写保留成功的TCP地址

点击更新。

创建完成后，打开在线隧道列表，此时可以看到随机的公网地址已经发生变化，地址名称也变成了保留和固定的TCP地址。

这样我们连接到目标主机就没有任何的阻碍啦！

总结

完成以上配置后，Python脚本已经可以持续读取Nginx的错误日志，并根据错误级别和关键词识别文件缺失、权限异常、连接失败及上游超时等问题。

当检测到的错误数量达到设定阈值时，脚本会整理发生时间和相关日志内容，再通过钉钉机器人发送告警消息。这样即使没有一直登录服务器查看日志，也能及时收到异常提醒。

将脚本配置为systemd服务后，监控程序可以在后台持续运行，并跟随系统启动，不需要每次手动打开终端执行。

通过模拟写入两条错误日志，本文也完成了从日志产生、脚本识别到钉钉收到消息的完整测试，确认自动告警链路已经正常工作。

最后，通过cpolar为服务器的SSH端口创建公网TCP地址，可以在其他网络环境中远程登录主机，查看监控服务状态、修改脚本配置或继续排查Nginx问题。

从Python读取日志、钉钉发送告警，到systemd后台运行和cpolar远程管理，本文完成了一套轻量化的Nginx错误日志自动监控方案，让原本依赖人工检查的故障信息能够主动发送出来。