☰
从零搭建监控系统:用 WxPusher 实现微信告警
2026/9/27 20:25:53 网站建设 项目流程

从零搭建监控系统:用 WxPusher 实现微信告警

背景

之前做了个系统监控 API,能看 CPU、内存、磁盘,也能存历史、画图表。但它有个问题——只能"看",不能主动告诉你。CPU 跑满了,我得自己打开页面才知道。我想要的是:服务器出问题,微信直接收到通知。

试过的方案

一开始想用 PushPlus,注册简单,接口也简单,但实名认证要收 3.9 元。不是嫌贵,是觉得为一个告警功能花这个钱不值。

然后试企业微信的群机器人。注册了企业微信,建了群,加了机器人,结果发现个人版不支持群机器人的 Webhook,只能聊天,不能推送。折腾了一下午,放弃。

最后用了 WxPusher。免费、不用实名,接口跟 PushPlus 差不多。扫码登录,创建应用,拿到 AppToken 和 UID,就能推了。

实现过程

在 api_flask.py 里加了两个函数。

第一个是推送函数:

defsend_alert(title,content):url="https://wxpusher.zjiecode.com/api/send/message"payload={"appToken":WXPUSHER_TOKEN,"content":content,"summary":title,"contentType":1,"uids":[WXPUSHER_UID]}requests.post(url,json=payload,verify=False,timeout=10)

verify=False是必须的,因为 CentOS 7 的证书过期了,不加会报 SSL 错误。

第二个是告警检查函数:

defcheck_alerts(cpu_1min,mem_percent,disk_percent):...ifcpu_value>0.8:last=last_alert_time.get('cpu',0)ifnow-last>1800:send_alert("CPU 告警",f"CPU 负载过高:{cpu_value}")save_alert_to_db("cpu",f"CPU 负载过高:{cpu_value}")last_alert_time['cpu']=now

冷却机制:为什么不能报一次就完事

一开始我根本没想过冷却这件事。逻辑很简单:CPU 超过 0.8 就推一条微信,就这样。

上线第一天我就被自己坑了。

crontab 每 5 分钟调一次/report,/report里会检查告警。如果 CPU 一直高,那每 5 分钟就推一条。一小时 12 条,一晚上 100 多条。第二天早上打开微信,全是"CPU 负载过高"。

这在运维里有个专门的说法,叫告警风暴。听着挺吓人,其实原理很简单——你没告诉系统"什么时候该闭嘴"。

告警风暴的危害比"没告警"还大。因为一旦消息太多,人会本能地忽略所有通知。真正出事的那条,反而被淹了。老运维都懂一个道理:告警不是越多越好,是越准越好。

怎么加冷却

思路很直接:记住上次报警是什么时候,如果离现在太近,就先别报。

用字典存,因为我有三种告警,每种都要独立记时间:

last_alert_time={}# 全局字典,存每种告警上次推送的时间戳

判断的时候:

now=time_module.time()# 当前时间(秒)ifcpu_value>0.8:# 第一层:先看超没超阈值last=last_alert_time.get('cpu',0)# 取上次 CPU 报警的时间,没有就当 0ifnow-last>1800:# 第二层:离上次报警超过 30 分钟才报send_alert("CPU 告警",f"CPU 负载过高:{cpu_value}")save_alert_to_db("cpu",f"CPU 负载过高:{cpu_value}")last_alert_time['cpu']=now# 记下这次报警时间

为什么要两层判断

只判断阈值不行——CPU 一直高,就一直报。

只判断时间差也不行——CPU 明明正常,每 30 分钟也给你来一条。

两层合起来才是对的:真的超了,而且距离上次报警够久了,才报。

为什么用 time.time() 不用 datetime

因为要算时间差。time.time()返回的是秒数(比如 1758400000),两个秒数相减就是秒差,直接跟 1800 比。

datetime.now()返回的是日期对象,相减得到的是 timedelta 对象,还得再转成秒,麻烦。

为什么要用字典,不用一个变量

假设我只用一个last_alert_time变量:

CPU 刚报警 → last_alert_time = 现在 内存接着超阈值了 → 判断"现在 - last_alert_time",结果是 0,不报

内存告警被 CPU 的冷却时间误伤了。它们明明是两回事。

用字典分三个 key(cpu、memory、disk),三种告警各记各的,互不干扰。

1800 秒是怎么定的

30 分钟。这不是标准答案,是我自己拍脑袋定的。

太短:还是会被刷屏。
太长:真的持续出问题,你可能错过。

如果按运维的常规做法,一般会按"问题严重程度"分级:CPU 短暂飙升可以 15 分钟,磁盘满这种致命问题可能 5 分钟就报一次。我这边先用 30 分钟统一处理,够用。

告警记录入库:让告警留得住

微信推送是即时消息。推完就过去了,翻聊天记录得翻很久。想要"过去一周报了几次 CPU",光靠微信根本查不到。

所以我建了第二张表。

之前只有一张monitor_history,存的是每 5 分钟采集一次的原始数据。现在加一张alert_history:

CREATETABLEIFNOTEXISTSalert_history(idINTEGERPRIMARYKEYAUTOINCREMENT,timestampTEXT,alert_typeTEXT,messageTEXT)

四个字段:

  • id:主键,自动递增
  • timestamp:告警发生的时间
  • alert_type:类型(cpu / memory / disk)
  • message:告警内容

为什么不和 monitor_history 合一张表?

字段不一样。采集记录需要memory_total这种字段,告警记录不需要。硬塞一起,一半字段是空的,乱。数据库设计有个基本原则:一类数据一张表。

每次告警,做两件事

send_alert("CPU 告警",f"CPU 负载过高:{cpu_value}")# 推微信给人看save_alert_to_db("cpu",f"CPU 负载过高:{cpu_value}")# 存数据库给系统记

一个负责"实时通知",一个负责"历史可查"。

为什么 alert_type 存英文 cpu 不存中文

因为后面可能要根据类型筛选。SELECT * FROM alert_history WHERE alert_type = 'cpu'比WHERE alert_type = 'CPU告警'更稳。展示的时候再翻译成中文就行。

页面展示

在/dashboard的dashboard()函数里,多查一次告警表:

cursor.execute('SELECT timestamp, alert_type, message FROM alert_history ORDER BY id DESC LIMIT 10')alert_rows=cursor.fetchall()

ORDER BY id DESC让最新的排前面,LIMIT 10只取 10 条。

查出来的是元组列表,转成字典列表方便 HTML 里用:

alerts=[]forrowinalert_rows:alerts.append({"time":row[0],"type":row[1],"message":row[2]})

然后传给模板:

returnrender_template('index.html',...,alerts=alerts)

HTML 里用 Jinja2 循环生成表格:

{% for alert in alerts %}<tr><td>{{ alert.time }}</td><td>{{ alert.type }}</td><td>{{ alert.message }}</td></tr>{% endfor %}

每次循环生成一行,页面上就能看到最近 10 条告警。

踩的坑

PushPlus 实名要 3.9 元

一开始看 PushPlus 文档,接口简单,几乎零学习成本。结果调接口时报了个错:

{"code":905, "msg":"账户未进行实名认证"}

去实名页面一看,要收 3.9 元。不贵,但我觉得为个告警功能花这个钱不值,就换了。

企业微信个人版没有群机器人

注册了企业微信,建了群,加了机器人,看起来一切正常。但点机器人详情,发现只有一个"发消息"按钮和"添加到群聊",没有 Webhook 地址。

Webhook 是代码推送消息的入口,没有它就没法用。查了半天才确认:企业微信个人注册的账号,不支持群机器人的 Webhook 功能。折腾了一下午,放弃。

CentOS 7 的证书过期

WxPusher 的接口测试通了,但 Python 里调的时候报 SSL 错误:

SSL: CERTIFICATE_VERIFY_FAILED

原因是 CentOS 7 太老,系统里的根证书过期了,不认识 WxPusher 的证书。加个参数跳过验证:

requests.post(url,json=payload,verify=False,timeout=10)

verify=False意思是"别验证证书了,直接连"。这不是安全做法,只适合本地测试和老系统。生产环境应该更新证书,而不是关掉验证。

Token 硬编码被 GitGuardian 警告

最开始图方便,Token 直接写在代码里:

WXPUSHER_TOKEN="AT_Cde2Y..."

推到 GitHub 之后,第二天收到 GitGuardian 的邮件:检测到你的仓库里暴露了 Bearer 令牌。

那一刻挺懵的,因为之前完全没意识到这个问题。后来查资料才知道:API Key 等于账号密码,提交到公开仓库等于公开你的身份。

改法是改成从环境变量读:

WXPUSHER_TOKEN=os.environ.get('WXPUSHER_TOKEN','')

然后在自己机器的.bashrc里设置:

exportWXPUSHER_TOKEN="AT_Cde2Y..."

这样代码里只有变量名,值存在操作系统里,推到 GitHub 也不会泄露。以后再写任何项目,Token 都不能写死在代码里。

忘了加冷却,差点被自己刷屏

上面讲过了,第一天就被自己坑。一晚上 100 多条消息。当时的心理感受是:这东西比没告警还烦。

后来加了冷却才正常。

效果

现在系统在跑:

  • CPU 超过 0.8(单核负载)
  • 内存使用率超过 90%
  • 磁盘使用率超过 90%

任意一个超了,微信直接收到推送。30 分钟内不重复推同样的。

打开/dashboard,能看到三个卡片(实时状态)、一条 CPU 历史趋势图(最近 10 次)、一张告警历史表格(最近 10 条告警记录)。

整个项目从最早的check_system.py脚本,变成了一个能采集、能存库、能画图、能告警的小系统。

下一步

现在的告警还比较粗糙,能优化的方向:

  • 多级告警:轻微、严重、致命,不同级别不同的冷却时间和推送频率
  • 告警恢复通知:问题修好后发一条"已恢复"
  • 加登录:/dashboard现在谁都能看,应该加身份验证
  • 换 MySQL:SQLite 单机够用,但如果多人访问,SQLite 会锁表

先做到这一步,剩下的慢慢迭代。

GitHub:https://github.com/zzp-03/monitor

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询