网站延迟检测报警

---

网站延迟检测与钉钉报警

介绍

这个 Python 脚本用于监控指定主机的延迟，并在检测到高延迟或 Ping 超时时，通过钉钉 webhook 发送报警信息。脚本会对主机进行定期 Ping 操作，并将结果记录到本地文件中，同时检测延迟的情况。如果发现延迟过高或 Ping 超时，则会发送钉钉报警。

功能

1. Ping 主机：对指定的主机进行 Ping 操作，记录延迟数据。
2. 记录结果：将延迟数据记录到本地文件中。
3. 钉钉报警：在检测到高延迟或 Ping 超时的情况下，发送钉钉报警。

代码

#!/usr/bin/python3
import subprocess
from operator import itemgetter
from time import strftime
import threading
import re
import requests

# 主机名称
hostname = "自己webhook关键词+主机唯一表示自己认识就行"

# 发送钉钉报警
def send_webhook_alert(message):
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=3d3c3a0e7f45bd760b6d0dde83fe5b193783a50a9db5a5f92dbXXXXXXX"
    headers = {
        "Content-Type": "application/json"
    }
    payload = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    try:
        response = requests.post(webhook_url, json=payload, headers=headers)
        if response.status_code == 200:
            print("Webhook警报发送成功")
        else:
            print(f"无法发送Webhook警报，状态码: {response.status_code}")
    except Exception as e:
        print(f"发送Webhook警报时出错: {e}")

# Ping 主机
def ping_host(host):
    print(f"{strftime('%Y-%m-%d %H:%M:%S')} 正在Ping {host}:\n")
    result = []
    timeout_count = 0

    for i in range(10):
        try:
            output = subprocess.check_output(f"ping -c 1 {host}", shell=True, stderr=subprocess.STDOUT)
            lines = output.splitlines()
            line = lines[1].decode()  # 包含ping结果的行
            match = re.search(r'time=(\d+\.?\d*)', line)  # 在行中查找延迟信息
            if match:
                delay = float(match.group(1))
                result.append((i, delay))
                print(f"数据包 {i}，延迟 {delay} 毫秒\n")
            else:
                print(f"数据包 {i}，未找到延迟信息。\n")
        except subprocess.CalledProcessError as e:
            print(f"数据包 {i}，Ping失败。{e.output.decode()}")
            timeout_count += 1
        except Exception as e:
            print(f"Ping过程中出现意外错误: {e}")
            timeout_count += 1

    if timeout_count >= 3:
        send_webhook_alert(f"{hostname}: {host} 的 Ping 超时 {timeout_count} 次")

    result.sort(key=itemgetter(1), reverse=True)
    top3 = result[:3]

    with open("/opt/shell/ping.txt", "a") as f:
        f.write(f"{strftime('%Y-%m-%d %H:%M:%S')} 正在Ping {host}:\n")
        for i, delay in top3:
            f.write(f"数据包 {i}，延迟 {delay} 毫秒\n")
            if delay > 500:
                send_webhook_alert(f"{hostname}: 检测到高延迟：{delay} 毫秒，主机：{host}")
        f.write("\n")

# Ping 主机列表
def ping_hosts(hosts):
    for host in hosts:
        ping_host(host)

# 主机列表
hosts = [
    'kr-third-dispatch.自己域名.co',
    'sg-third-dispatch.自己域名.co',
    'static.jianweidata.com',
    'api.chandler.bet',
    'sentry.自己域名.it',
    'dispatch-uk.自己域名.co',
    'dispatch-us.自己域名.co',
    'dispatch-aus.自己域名.co',
    'track-aus.自己域名.co',
    'aus-gateway.自己域名.co',
    'track-eu.自己域名.co',
    'api.mapbox.com',
    'www.baidu.com',
    'www.google.com',
    'dispatch-ca.自己域名.co',
    'ca-gateway.自己域名.co',
    'dispatch-uk.自己域名.co',
    'uk-gateway.自己域名.co',
    'dispatch-eur.自己域名.co',
    'dispatch-nzd.自己域名.co',
    'nzd-gateway.自己域名.co',
    'usa-gateway.自己域名.co',
    'dispatch-jp.自己域名.co',
    'jp-gateway.自己域名.co',
    'dispatch-kr.自己域名.co',
    'track-kr.自己域名.co',
    'kr-gateway.自己域名.co',
    'dispatch-sg.自己域名.co',
    'track-sg.自己域名.co',
    'sg-gateway.自己域名.co',
    'fr-gateway.自己域名.co',
    '8.8.8.8'
]

if __name__ == "__main__":
    threading.Thread(target=ping_hosts, args=(hosts,)).start()