服务器异常？用Python监控日志并通过钉钉机器人实时告警

前言

在当今高度依赖数字系统的时代，服务器的稳定运行已成为业务连续性的生命线。然而，再健壮的系统也难免遭遇磁盘爆满、服务崩溃、网络中断或安全攻击等突发异常。传统的人工巡检不仅效率低下，更难以做到秒级响应——往往等到用户投诉时，故障早已持续数小时。

有没有一种方式，能让服务器“主动说话”，在问题初现端倪时就第一时间通知运维人员？答案是肯定的。借助轻量级脚本与即时通讯工具的结合，我们可以构建一套低成本、高效率、零值守的智能告警系统。

本文将带你从零开始，使用Python编写一个日志监控程序，实时扫描关键日志文件中的错误关键词（如ERROR、no space left on device等），并通过 钉钉机器人 将告警信息精准推送到你的手机或团队群聊。无论你是在家休假，还是深夜熟睡，只要服务器一“喊疼”，你就能立刻收到提醒，真正做到 早发现、快响应、少损失。

无需复杂平台，不依赖商业软件——一行代码，让服务器拥有“自我呼救”的能力。

downloaded-image (6)

1.测试效果

效果预览

当你的/var/log/syslog或应用日志中出现：

May 08 15:30:01 server cpolar[1234]: Failed to write log: no space left on device

你将在钉钉群收到如下消息

【服务器告警】

主机：Z4S-VJFL

时间：2026-05-08 15:30:01

日志路径：/var/log/cpolar/access.log.20260508

内容：Failed to write log: no space left on device

磁盘空间不足！请立即处理！

通过这个Python脚本，你实现了

  • 实时监控任意日志文件
  • 关键词触发精准告警
  • 钉钉消息秒级触达手机
  • 低资源占用，适合嵌入式设备
  • 零依赖复杂框架，纯标准库 + requests

  • 延伸场景：监控Nginx错误日志 → 发现攻击行为

  • 监控数据库慢查询日志 → 优化SQL
  • 监控Docker容器日志 → 异常自动重启

让机器替你“盯屏”，你只管睡觉！

2.获取钉钉Webhook和Secret

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

image-20260325141357694

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

image-20260325141635253

image-20260325141712706

添加自定义机器人：

image-20260325141746662

点击添加：

image-20260325141818703

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

image-20260508163310657

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

image-20260508163449918

点击“加签”，复制生成的秘钥，留着备用：

image-20260509103800058

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

image-20260325143249017

3.编写Python监控脚本

创建要监控的日志文件：

mkdir /var/log/ceshi/
touch /var/log/ceshi/access.log.20260508

创建文件log_monitor.py：

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

# ====== 配置区（请按需修改）======
LOG_FILE = "/var/log/ceshi/access.log.20260508"  # 要监控的日志文件
KEYWORDS = ["error", "failed", "exception", "no space left on device", "disk full"]
HOSTNAME = os.uname().nodename  # 自动获取主机名

# 钉钉机器人配置（从钉钉后台获取）
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=dbf63c2e3c2f2dd"
DINGTALK_SECRET = "SEC8bb4d908c1039a4"

# ====== 钉钉签名函数 ======
def get_dingtalk_sign():
    timestamp = str(round(time.time() * 1000))
    secret_enc = DINGTALK_SECRET.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, DINGTALK_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_alert(message):
    timestamp, sign = get_dingtalk_sign()
    webhook_url = f"{DINGTALK_WEBHOOK}&timestamp={timestamp}&sign={sign}"

    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "【服务器告警】",
            "text": message
        }
    }
    try:
        response = requests.post(webhook_url, json=data, timeout=10)
        if response.status_code != 200:
            print(f"[!] 钉钉发送失败: {response.text}")
    except Exception as e:
        print(f"[!] 钉钉请求异常: {e}")

# ====== 监控日志主函数 ======
def monitor_log():
    log_path = Path(LOG_FILE)
    if not log_path.exists():
        print(f"[!] 日志文件不存在: {LOG_FILE}")
        return

    # 获取文件初始大小（用于断点续读）
    file_size = log_path.stat().st_size
    print(f"[+] 开始监控日志: {LOG_FILE} (初始大小: {file_size} bytes)")

    while True:
        try:
            current_size = log_path.stat().st_size
            if current_size < file_size:
                # 日志被轮转（如 logrotate），重置位置
                print("[+] 检测到日志轮转，重新开始读取")
                file_size = 0

            if current_size > file_size:
                with open(LOG_FILE, 'r', encoding='utf-8', errors='ignore') as f:
                    f.seek(file_size)  # 从上次位置读起
                    lines = f.readlines()
                    file_size = current_size  # 更新已读位置

                    for line in lines:
                        line_lower = line.lower()
                        for keyword in KEYWORDS:
                            if keyword in line_lower:
                                # 构建告警消息
                                now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                                msg = (
                                    f"## 🔴 **【服务器告警】**\n\n"
                                    f"- **主机**: `{HOSTNAME}`\n"
                                    f"- **时间**: `{now}`\n"
                                    f"- **日志路径**: `{LOG_FILE}`\n"
                                    f"- **内容**: `{line.strip()}`\n\n"
                                    f"> ⚠️ **检测到关键词: `{keyword}`**"
                                )
                                print(f"[ALERT] {line.strip()}")
                                send_dingtalk_alert(msg)
                                break  # 避免重复告警同一行
            time.sleep(1)  # 每秒检查一次
        except KeyboardInterrupt:
            print("\n[+] 监控已停止")
            break
        except Exception as e:
            print(f"[!] 监控异常: {e}")
            time.sleep(5)

if __name__ == "__main__":
    monitor_log()

image-20260509104053176

4.安装依赖 & 赋予执行权限

进入你的项目目录:

cd /root/ceshi

创建虚拟环境（Python 3.6+ 自带 venv）

python3 -m venv venv

激活虚拟环境:

source venv/bin/activate

此时提示符会变成 (venv)，再安装包

pip install requests

image-20260509104758213

5.测试运行

运行你的脚本：

python log_monitor.py

image-20260509111944293

编辑/var/log/ceshi/access.log.20260508监控文件：

vi access.log.20260508

May 08 15:30:01 server cpolar[1234]: Failed to write log: no space left on device

image-20260509112027550

钉钉成功发送告警：

image-20260509112142262

6.运行 & 开机自启

方式 1：使用nohup临时运行

nohup python3 /path/to/log_monitor.py > /var/log/log_monitor.log 2>&1 &

方式 2：创建systemd服务（推荐，支持开机自启）

创建服务文件：

sudo vim /etc/systemd/system/log-monitor.service

[Unit]
Description=Log Monitor for Server Alerts
After=network.target

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

[Install]
WantedBy=multi-user.target

启动服务：

sudo systemctl daemon-reload
sudo systemctl enable --now log-monitor.service
sudo systemctl status log-monitor

image-20260509141926887

在现实运维中，很多开发者、创客甚至中小企业都把关键服务部署在家庭NAS、老旧服务器或内网测试机上——它们没有固定公网IP，藏在路由器后面，平时靠cpolar暴露个Web界面勉强够用。可一旦半夜磁盘爆满、Docker容器崩溃、系统日志里刷出“no space left on device”或“cpolar failed to reconnect”，你却远在千里之外出差、旅游，甚至正在熟睡。这时候，人工巡检毫无意义，而商业监控平台又太重、太贵。

但如果你提前做了两件事：

第一，在服务器上跑一个轻量Python脚本，持续扫描 /var/log/cpolar/ 或系统日志，一旦发现关键词就通过钉钉机器人推送告警到手机；

第二，用cpolar配置一条TCP隧道，把本地22端口（SSH）映射到公网，哪怕只是临时分配的地址。

那么当故障发生时，你的手机会立刻收到钉钉消息：“【告警Z4S主机磁盘已满！建议通过 cpolar SSH登录清理”。你只需打开终端，输入ssh -p 22222 root@192.xxx.xxx.xxx，就能像坐在机器前一样，删日志、重启服务、查进程——整个过程不到五分钟，无需回家，不用求人。这不仅是技术组合，更是对“运维最后一公里”的真实补全：让每一台沉默的内网设备，都能在危急时刻主动呼救，并为你留一扇随时可进的门。 尤其对于用极空间、树莓派、旧笔记本搭建家庭实验室的人来说，这套方案成本近乎为零，却能极大提升系统的可靠性和你的安心感。

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

7.1 什么是cpolar？

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

7.2 部署cpolar

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

❤️以下是安装cpolar步骤：

官网在此：https://www.cpolar.com

使用一键脚本安装命令：

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

image-20250725104019896

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

sudo systemctl status cpolar

22e5adfaf290a17fc3384bb296055259

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

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

8a6698b1bf26d64ba3645827fbfb1c29

8.配置公网地址

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

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

image-20260509153640177

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

  • tcp 表示使用的协议类型

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

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

image-20260509153758249

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

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

image-20260509155111600

9.保留固定TCP公网地址

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

image-20251210160529622

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

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

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

image-20260509155255401

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

image-20260509155316176

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

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

点击更新。

image-20260509155401559

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

image-20260509155418632

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

总结

在当今分布式、边缘化的计算环境中，大量关键服务正运行在无公网IP的家庭服务器、老旧主机或内网测试机上——它们稳定时默默无闻，一旦深夜突发磁盘写满、进程崩溃或网络中断，往往因无法远程接入而酿成严重后果。本文提出的解决方案，正是针对这一现实痛点：通过一个轻量级Python脚本持续监控系统或应用日志（如cpolar、Docker、Nginx等），一旦检测到“no space left on device”“failed to connect”等异常关键词，立即触发钉钉机器人，将结构化告警信息推送至运维人员手机。整个过程无需复杂平台，成本近乎为零，却能实现秒级感知与通知。更进一步，若提前配置cpolar的TCP隧道穿透SSH（22端口），收到告警后即可直接通过ssh -p 公网端口 user@公网IP远程登录内网机器，快速排查清理，真正形成“异常发现—即时告警—远程修复”的完整运维闭环。这不仅是一套技术方案，更是对中小开发者、创客和家庭用户而言，提升系统可靠性与自身掌控感的实用实践。

感谢您对本篇文章的喜爱，有任何问题欢迎留言交流。cpolar官网-安全的内网穿透工具 | 无需公网ip | 远程访问 | 搭建网站

Share:

鸽子

鸽子

我是鸽子，是这个网站的文章编辑之一！喜欢研究各种有趣的开源项目。希望大家喜欢我的文章！

发表回复

目录

On Key

推荐文章