ToB企服应用市场:ToB评测及商务社交产业平台

标题: 哪吒面板联合微信机器人实时推送服务器告警消息 [打印本页]

作者: 拉不拉稀肚拉稀 时间: 2024-10-19 01:35
标题: 哪吒面板联合微信机器人实时推送服务器告警消息
大家好，我是雄雄，欢迎关注微信公众号：雄雄的小讲堂。

前言

服务器太多，管不外来怎么办？宝塔就退出了一款云监控面板，花里胡哨还挺有意思的，也比较实用，唯一的缺点就是能添加的服务器有限，假如多了，则需要开会员，这个钱，对于个人来说，能省就省了，省下来可以再买一台服务器…
今天，我们就来找一个能替代宝塔云监控的面板——哪吒面板。
实现效果

可以关照的消息如上所示。
前提条件

一个微信机器人，能通过api发送消息到微信群中
已经装好的哪吒监控面板
会点编程（我已python为例）

今天琢磨了一天，对于告警，可算是琢磨明白了，趁着如今还没忘记，就在此记载一下，回头忘记了之后，可以翻出来看看。
假设我们上面的三个前提条件都满足，那么接着往下看！
这是我的哪吒前台面板：

设置告警关照方式

下面，我们开始设置面板的告警关照方式。首先，我们进入面板的背景，找到告警，然后右上角添加【关照方式】，如下所示：

在面板消息中，占位符 #DATETIME# 代表事件发生的时间戳。当关照被触发时，面板会自动将 #DATETIME# 替换为事件的实际时间。
#NEZHA# 是面板消息占位符，面板触发关照时会自动用实际消息替换占位符。
Body 内容是 JSON 格式的：当请求类型为 FORM 时，值为 key:value 的形式，value 里面可放置占位符，关照时会自动替换。当请求类型为 JSON 时只会简朴举行字符串替换后直接提交到 URL。
URL 里面也可放置占位符，请求时会举行简朴的字符串替换。
填写的信息阐明:

名称：随便写个
分组：随便写个，留意，设置告警的时候，分组一定是如今写的这个，好比我的是fastapi
URL：就是你接口的地点（该接口实现了发送消息至微信中）
请求方式：post
请求类型：json
header：不消写（我的接口没有设置）
Body：也就是当到达告警之后，需要给服务器里面推送的消息是什么，直接复制我的：

{
"content": "#NEZHA#",
"server_name": "#SERVER.NAME#",
"server_ip": "#SERVER.IP#",
"server_ipv4": "#SERVER.IPV4#",
"server_ipv6": "#SERVER.IPV6#",
"cpu": "#SERVER.CPU#",
"mem": "#SERVER.MEM#",
"swap": "#SERVER.SWAP#",
"disk": "#SERVER.DISK#",
"net_in_speed": "#SERVER.NETINSPEED#",
"net_out_speed": "#SERVER.NETOUTSPEED#",
"transfer_in": "#SERVER.TRANSFERIN#",
"transfer_out": "#SERVER.TRANSFEROUT#",
"load1": "#SERVER.LOAD1#",
"load5": "#SERVER.LOAD5#",
"load15": "#SERVER.LOAD15#",
"tcp_conn_count": "#SERVER.TCPCONNCOUNT",
"udp_conn_count": "#SERVER.UDPCONNCOUNT"
}

复制代码

然后点击确定即可。
新增告警规则

点击下面的【新增告警规则】，我们就可以添加一条告警规则了。

如何添加？可以参考我的：

填写阐明：

名称：随便写个，到时候也会发到微信里面
规则，可以看看官方文档，https://nezha.wiki/guide/notifications.html#%E5%9F%BA%E6%9C%AC%E8%A7%84%E5%88%99

好比我设置的是：** CPU 持续 600 秒超过 50% **

[{"type":"cpu","max":50,"duration":600}]

复制代码

关照方式组：与你上面设置的关照方式的分组要一致，不然没法给你关照
关照触发模式：始终触发或者单次触发，按照你自己的需求选择就行
启用：勾选上就是正式启用见效了。

利用python实现自界说告警接口，发送至微信

首先，我们需要按照告警的字段，声明一个类：

# 消息实体
class CallBackDataEntity(BaseModel):
content: Optional[str] = None, # 通知事件名称
server_name: Optional[str] = None, # 服务器名称
server_ip: Optional[str] = None, # 服务器IP
server_ipv4: Optional[str] = None, # 服务器IPv4
server_ipv6: Optional[str] = None, # 服务器IPv6
cpu: Optional[str] = None, # CPU使用率
mem: Optional[str] = None, # 内存使用率
swap: Optional[str] = None, # 交换分区使用率
disk: Optional[str] = None, # 磁盘使用率
net_in_speed: Optional[str] = None, # 实时上传速度
net_out_speed: Optional[str] = None, # 实时下载速度
transfer_in: Optional[str] = None, # 总上传
transfer_out: Optional[str] = None, # 总下载
load1: Optional[str] = None, # 1分钟负载
load5: Optional[str] = None, # 5分钟负载
load15: Optional[str] = None, # 15分钟负载
tcp_conn_count: Optional[str] = None, # TCP连接数
udp_conn_count: Optional[str] = None, # UDP连接数
# 创建get方法
def get(self, param):
return getattr(self, param)

复制代码

我这都有解释，可以直接复制下来就行。
然后我们还需要再写个接口，来实现发送到微信群中：
[code]
# 处理通知
@nezha_router.post("/gaojing")
async def gaojing(data: CallBackDataEntity):
# 将收到的消息转换成json
data = data.dict()
print(f"触发了告警消息:{data}")
# 需要通知的消息
str = ''
# 获取content内容
content = data.get("content")
# 获取ip地址
ip = data.get("server_ip")
# 获取CPU使用率
cpu = float(data.get("cpu"))
# 获取内存使用情况,换算成已使用G的内存
mem = round(float(data.get("mem")) / (1024 ** 3), 2)
# 存储使用情况换算成已使用G的硬盘
disk = round(float(data.get("disk")) / (1024 ** 3), 2)
# 实时下载速度 Mbs
net_out_speed = round(float(data.get("net_out_speed")) / 1024, 2)
# 实时上传速度 Mbs
net_in_speed = round(float(data.get("net_in_speed")) / 1024, 2)
# 总上传
transfer_in = round(float(data.get("transfer_in")) / (1024 ** 3), 2)
# 总下载
transfer_out = round(float(data.get("transfer_out")) / (1024 ** 3), 2)
# 1分钟负载
load1 = float(data.get("load1"))
# 5分钟负载
load5 = float(data.get("load5"))
# 15分钟负载
load15 = float(data.get("load15"))
# TCP连接数
tcp_conn_count = data.get("tcp_conn_count")
# UDP连接数
udp_conn_count = data.get("udp_conn_count")
# 拼接通知消息
str = "⚠️⚠️⚠️哪吒面板告警⚠️⚠️⚠️\r\r" \
"

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)