服务器半夜报错没人看？用 Python 盯日志、钉钉通知，再通过 cpolar SSH 回去处理

服务器半夜报错没人看？用 Python 盯日志、钉钉通知，再通过 cpolar SSH 回去处理

前言

我做服务器运维时，最怕的不是故障本身，而是故障已经发生很久，自己却还不知道。磁盘写满、服务报错、容器异常这类问题，如果全靠人定时登录服务器翻日志，平时没事时觉得还能接受，真到半夜或者人在外面时就很容易漏掉。对我来说，监控最有价值的地方不是堆一套很重的平台，而是先把最关键的闭环做出来：日志里出现异常，有东西替我盯着；命中关键词以后，手机能收到消息；确认确实需要处理时，我还能从外面直接 SSH 回去。而且我希望告警只是把问题及时送到眼前，不是替我做判断；关键词命中往往只是线索，收到消息以后还得结合日志上下文确认，避免每一行 error 都被当成真正故障。

这次我用一个 Python 脚本持续读取日志增量，匹配 error、failed、exception、no space left on device、disk full 等关键词，再通过钉钉自定义机器人发送 Markdown 告警；脚本先手动验证，再用 nohup 或 systemd 常驻。最后安装 cpolar，把本地 SSH 22 端口映射成公网 TCP 地址，先验证随机端口，再切到固定 TCP 地址。整篇我更关心的是“发现异常—收到提醒—远程处理”能不能真正连成一条链，而不是把一次关键词命中写成一套完整监控平台。

1. 先明确这套方案解决什么问题

这套方案不打算替代完整的 Prometheus、Zabbix 或商业监控平台。

它做的是一件更轻、更直接的事：

持续读日志 → 命中关键词 → 发钉钉告警 → 必要时远程 SSH 回服务器处理。

当日志里出现类似下面的内容：

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

告警消息里会带上：

• 主机名；
• 时间；
• 日志路径；
• 命中的原始日志；
• 触发关键词。

示例里最终希望看到的是“磁盘空间不足”这类能直接定位方向的提醒。

不过我不会把“每秒检查一次文件”直接写成“钉钉一定秒级送达”。脚本轮询频率、网络和钉钉接口都会影响实际到达时间。

2. 先准备钉钉机器人

打开钉钉群设置。

进入：

智能群助手 → 添加机器人

选择自定义机器人。

点击添加。

机器人名称这里使用：

服务器告警

接着配置安全设置。

钉钉这里可以设置关键词，也可以使用加签或 IP 地址限制。

这次选择的是：

加签

复制生成的 Secret。

完成以后，再复制机器人 Webhook。

这两个值后面都会写进 Python 脚本。

我会把它们当成凭据处理：演示里虽然直接写在代码中，但真正长期运行时，不建议把 Webhook 和 Secret 到处复制或公开。

3. 创建测试日志，再编写监控脚本

先创建测试目录和日志文件：

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}×tamp={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()

这段脚本的核心逻辑并不复杂：

这里还有几个我会先记住的细节。

脚本真正监控的是：

/var/log/ceshi/access.log.20260508

而前面的效果预览里展示过：

/var/log/cpolar/access.log.20260508

两者不是同一个路径。真正使用时要以 LOG_FILE 为准，或者把它改成自己确实想盯的日志文件。

另外，脚本里的 Webhook 拼接行写的是：

×tamp={timestamp}&sign={sign}

这段按现有代码保留。如果实际复现时钉钉请求异常，我会优先检查这里的签名 URL 拼接是否与自己机器人要求一致，而不是先怀疑日志读取逻辑。

4. 建 Python 虚拟环境并安装 requests

进入项目目录：

cd /root/ceshi

创建虚拟环境：

python3 -m venv venv

激活：

source venv/bin/activate

然后安装：

pip install requests

虽然脚本大部分逻辑使用 Python 标准库，但 HTTP 请求这一段依赖：

requests

所以运行环境里必须能导入这个包。

5. 先手动运行，别一上来就做开机自启

执行：

python log_monitor.py

这时程序会读取 LOG_FILE 当前大小，然后开始等待新增内容。

接下来进入测试日志。

当前命令写的是：

vi access.log.20260508

然后加入这一行：

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

这里还有一处路径关系要特别注意。

前面已经进入：

/root/ceshi

而 vi access.log.20260508 是相对路径；脚本实际监控的是：

/var/log/ceshi/access.log.20260508

所以真正测试时，需要确认自己写入的是脚本正在监控的那个文件。否则“日志里明明写了错误但钉钉没消息”，原因可能只是改错了文件。

当目标日志真正新增这行以后，钉钉收到告警。

到这里才能确认：

日志新增 → 关键词命中 → Python 脚本 → 钉钉机器人

这一条链已经跑通。

6. 临时后台运行和 systemd 常驻，两种方式分开用

如果只是先跑一段时间，可以使用 nohup：

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

但这里的：

/path/to/log_monitor.py

只是路径占位形式，真正运行时必须对应脚本实际位置。

如果准备长期监控，我更倾向于交给 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

这样 systemd 会负责启动和异常重启。

不过这里同样有一处路径要核对。

前面的项目目录是：

/root/ceshi

而 systemd 的 ExecStart 使用：

/ceshi/log_monitor.py

两者并不一致。

代码按这套步骤保留，真正启用服务前要确认脚本到底放在哪里，否则服务文件本身可能找不到脚本。

7. 告警解决的是“知道出事了”，还缺“人在外面怎么处理”

到这里，服务器已经能在日志出现异常时给钉钉发消息。

但收到消息以后，如果机器在家庭 NAS、实验室或者公司内网，没有公网 IP，下一步仍然可能卡住：

我知道它出问题了，却进不去。

所以这篇后半段再补一条 SSH 远程入口。

cpolar 在这里不参与日志监控，也不发钉钉消息。

它只负责：

把服务器本地 SSH 22 端口提供成公网 TCP 入口。

这样整个流程才从“告警”继续往“处理”延伸。

8. 安装 cpolar

执行：

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

安装完成以后查看服务状态：

sudo systemctl status cpolar

状态正常后，通过：

http://ip:9200

打开 cpolar Web UI。

登录以后开始配置 SSH 隧道。

9. 先用随机 TCP 地址验证 SSH

进入 cpolar 隧道配置。

当前参数为：

• 隧道名称：ssh
• 协议：tcp
• 本地地址：22
• 端口类型：随机临时 TCP 端口
• 地区：China Top

创建成功以后，进入在线隧道列表。

这里会看到类似：

• 域名：2.tcp.cpolar.top
• 公网端口：12178

这样的 TCP 地址。

前面的场景描述里还出现过：

ssh -p 22222 root@192.xxx.xxx.xxx

这种占位式写法；真正后面测试使用的是下面这条命令：

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

这一步验证的是：

外部终端 → cpolar TCP 公网地址 → 内网服务器 SSH 22。

只要这条链能通，收到钉钉告警以后，就有了继续远程排查的入口。

10. 长期运维再换固定 TCP 地址

随机 TCP 端口适合先测试。

如果这台服务器以后长期作为远程运维目标，入口固定下来会省掉重复确认地址的麻烦。

进入 TCP 地址预留页面。

这里页面里出现了两个地区信息：

• 下拉菜单显示 China VIP；
• 后面的已保留记录写成 China Top。

已保留 TCP 地址示例为：

16.tcp.cpolar.top:14775

这两处地区信息按现有配置分别保留，真正操作时以自己账号页面实际显示的预留结果为准。

接着回到：

隧道管理 → 隧道列表

找到：

ssh

并编辑。

把：

• 端口类型改成固定 TCP 端口；
• 预留的 TCP 地址填写为前面保留成功的地址。

更新以后，再看在线隧道列表。

此时随机 TCP 地址已经替换成固定入口。

这样以后收到告警，再远程连回服务器时，不需要每次重新找随机端口。

11. 我会怎么理解这套“小型运维闭环”

真正让我觉得实用的不是 Python、钉钉或 cpolar 单独哪一个工具。

而是三件事各自只做自己那一层：

Python 脚本负责发现。
持续读取日志，匹配异常关键词。

钉钉机器人负责通知。
把主机、时间、日志路径和异常内容推到手机。

cpolar SSH 负责进入。
需要处理时，再从外部终端连回内网服务器。

这样遇到磁盘写满、服务报错之类的问题时，处理顺序就很明确：

日志出现异常 → 手机收到提醒 → 判断是否需要介入 → SSH 回服务器 → 清理、重启或继续排查。

总结

这套方案最适合我的地方，是它没有为了“监控”先引入一整套很重的平台，而是把最基本的发现、通知和远程处理先连起来。

这次实际跑通的主线是：

日志文件 → Python log_monitor.py → 关键词匹配 → 钉钉机器人 → 手动测试 → nohup / systemd 常驻 → cpolar → SSH 22 → 随机 TCP → 2.tcp.cpolar.top:12178 → 固定 TCP 16.tcp.cpolar.top:14775。

它当然不是完整监控体系：没有指标趋势、没有复杂规则引擎，也没有自动故障诊断。但如果只是想让一台内网服务器在出现关键日志时主动提醒，并且人在外面还能及时连回去处理，这条链已经把最重要的“发现”和“进入”接上了。对我来说，真正值得长期保留的不是“零值守”这个口号，而是异常发生时不用再等用户先来告诉我服务器出问题了。