python小脚本,实时监测服务器是否宕机状态,并发送到指定Telegram群组

✍️Auth:运维笔记       Date:2024/09/11       Cat:python相关       👁️:257 次浏览

一,前言

众所周知,市面上监控软件很多,有Zabbix,Prometheus等,但对于相对简单的功能,需要第一时间发现问题,如服务器宕机,zabbix和Prometheus都需要等几分钟才会报警。

想到最原始的方法,也是最简单的方法,就是ping 服务器,ping断开就是宕机。所以用一个小脚本即可。

二,脚本内容

1,脚本说明:

  • 每个服务器都有对应的名称,保存在字典 servers 中。
  • 如果某个服务器连续 10 秒无法 ping 通,则向 TG群发送服务器的名称和 IP(根据自己需求添加发送指定的媒体)。
  • 可以根据需求调整 time.sleep(10) 的时间间隔。
  • 状态跟踪: server_status 字典保存了每个服务器的当前状态,True 表示服务器在线,False 表示服务器不在线。
  • 状态变化检测:
    如果服务器从“在线”变为“不在线”,就会被标记为挂掉,发送“服务器挂掉”通知。
    如果服务器从“不在线”变为“在线”,就会被标记为恢复,发送“服务器恢复”通知。
  • 状态更新: 状态在每次检测后更新为当前状态。
    这样就能在服务器挂掉时收到通知,在恢复后也会及时得到通知。
  • 每次检测时挂掉服务器都会发送通知: 即使之前已经挂掉过,只要服务器继续挂掉,脚本会在每次循环中发送挂掉信息,如果只发送一次,会误发或者忽略,所以重复发送可以肯定服务器状态。
  • 恢复的服务器只发送一次通知: 通过跟踪状态,只在服务器从“不在线”状态切换到“在线”状态时发送“恢复”通知。重复发的挂掉信息则会停止。

2,脚本内容

import os
import time
import requests

# 服务器列表,格式为 { '服务器名称': '服务器IP' }
servers = {
    '测试服务器1': '192.168.203.132',
    'servername2': '192.168.1.2',
    'servername3': '192.168.1.3'
}

# 记录每个服务器的状态,True 表示在线,False 表示不在线
server_status = {name: True for name, ip in servers.items()}

# TG 机器人 API Token 和群 ID
telegram_token = '5249774617:asdasdasd.....'
chat_id = '-459853...'

# 向 Telegram 群发送通知的函数
def send_telegram_message(message):
    url = f'https://api.telegram.org/bot{telegram_token}/sendMessage'
    data = {
        'chat_id': chat_id,
        'text': message
    }
    try:
        requests.post(url, data=data)
    except Exception as e:
        print(f"发送消息失败: {e}")

# 检查服务器状态的函数
def check_servers():
    down_servers = []
    recovered_servers = []

    for name, ip in servers.items():
        response = os.system(f"ping -c 5 {ip} > /dev/null 2>&1")
        is_online = (response == 0)

        # 如果服务器挂掉(即 ping 不通)
        if not is_online:
            # 如果之前是在线的,刚刚挂掉,标记为挂掉并发送通知
            if server_status[name]:
                server_status[name] = False
            # 每次服务器挂掉时都发送通知
            down_servers.append((name, ip))

        # 如果服务器恢复(即 ping 通),只发送一次恢复通知
        elif not server_status[name] and is_online:
            recovered_servers.append((name, ip))
            server_status[name] = True

    return down_servers, recovered_servers

# 主循环
while True:
    down_servers, recovered_servers = check_servers()

    # 处理挂掉的服务器
    for name, ip in down_servers:
        send_telegram_message(f"❌{name} -Server DOWN,IP:({ip})")

    # 处理恢复的服务器
    for name, ip in recovered_servers:
        send_telegram_message(f"✅{name} -Server UP,IP:({ip})")

    time.sleep(10)  # 每 10 秒检测一次

三,脚本运行测试

原脚本由于是持续运行,所以是死循环,只要后台运行一次即可。

nohup python3 monitor.py >dev/null 2>&1 &

模拟测试结果,完成。


后续可以继续完善,可以写个检测此脚本的小shell程序,定时检查此脚本,万一机器重启,可以重启自动重启。

参考我以前文章:grep在shell脚本中使用,判断进程是否存在

打赏作者

发表评论