python监控系统资源使用率并钉钉报警脚本

监控系统资源使用率并钉钉报警

当Linux系统 cpu 内存 SWap交换空间 使用率超过80% 给钉钉发送报警。

#python3版本
[root@localhost 11:35:16]# python3 --version
Python 3.6.8

#pip3版本
[root@localhost 11:35:22]# pip3 --version
pip 9.0.3 from /usr/lib/python3.6/site-packages (python 3.6)

#安装requests模块
pip3 install requests

#创建脚本文件
vim monitor.py

#赋予脚本执行权限
chmod +x monitor.py

#运行脚本：
python3 monitor.py

#设置定时任务
#打开 crontab 编辑器：
crontab -e
#添加以下内容，每分钟运行一次脚本：
* * * * * /usr/bin/python3 /path/to/monitor.py
#将 /path/to/monitor.py 替换为脚本的实际路径。

monitor.py脚本内容

import os
import requests
import json
import socket

# 钉钉 Webhook URL
webhook_url = "钉钉 Webhook URL"

# 获取主机名称
def get_hostname():
    return socket.gethostname()

# 获取 IP 地址
def get_ip_address():
    try:
        # 使用 ifconfig 获取 IP 地址
        cmd = "ifconfig ens160 | grep 'inet ' | awk '{ print $2 }'"
        ip_address = os.popen(cmd).read().strip()
        if ip_address:
            return ip_address
        else:
            return "未知"
    except Exception as e:
        print(f"获取 IP 地址失败: {e}")
        return "未知"

# 获取 CPU 使用率
def get_cpu_usage():
    cmd = "top -bn1 | grep '%Cpu' | awk '{print $2}'"
    result = os.popen(cmd).read().strip()
    print(f"CPU 使用率命令输出: {result}")  # 调试信息
    return float(result)

# 获取内存使用率
def get_memory_usage():
    cmd = "free -m | grep 'Mem'"
    result = os.popen(cmd).read().split()
    memory_total = float(result[1])
    memory_used = float(result[2])
    memory_usage = (memory_used / memory_total) * 100
    print(f"内存使用率: {memory_usage:.2f}%")  # 调试信息
    return memory_usage

# 获取 SWAP 使用率
def get_swap_usage():
    cmd = "free -m | grep 'Swap'"
    result = os.popen(cmd).read().split()
    swap_total = float(result[1])
    swap_used = float(result[2])
    swap_usage = (swap_used / swap_total) * 100
    print(f"SWAP 使用率: {swap_usage:.2f}%")  # 调试信息
    return swap_usage

# 发送钉钉报警
def send_dingding_alert(message):
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": f"报警：{message}"  # 加入关键词 "报警"
        }
    }
    print(f"发送消息到钉钉: {message}")  # 调试信息
    response = requests.post(webhook_url, headers=headers, data=json.dumps(data))
    print(f"钉钉响应状态码: {response.status_code}")  # 调试信息
    print(f"钉钉响应内容: {response.text}")  # 调试信息
    return response.status_code

# 主函数
def main():
    print("脚本开始运行...")  # 调试信息
    hostname = get_hostname()
    ip_address = get_ip_address()
    cpu_usage = get_cpu_usage()
    memory_usage = get_memory_usage()
    swap_usage = get_swap_usage()

    if cpu_usage > 80:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ CPU 使用率超过 80%: {cpu_usage:.2f}%"
        send_dingding_alert(message)

    if memory_usage > 80:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ 内存使用率超过 80%: {memory_usage:.2f}%"
        send_dingding_alert(message)

    if swap_usage > 80:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ SWAP 使用率超过 80%: {swap_usage:.2f}%"
        send_dingding_alert(message)

if __name__ == "__main__":
    main()

注意事项1：钉钉Webhook URL

注意事项2：网络接口名称

 注意事项3：报警关键字。

 注意事项4：报警控制 资源利用率。

 运行结果：

python3 monitor.py

 

 Windows 在pycharm 上面执行脚本。

import os
import requests
import json
import socket
import psutil

# 钉钉 Webhook URL
webhook_url = " 钉钉 Webhook URL"

# 获取主机名称
def get_hostname():
    return socket.gethostname()

# 获取 IP 地址
def get_ip_address():
    try:
        hostname = socket.gethostname()
        ip_address = socket.gethostbyname(hostname)
        return ip_address
    except Exception as e:
        print(f"获取 IP 地址失败: {e}")
        return "未知"

# 获取 CPU 使用率
def get_cpu_usage():
    return psutil.cpu_percent(interval=1)

# 获取内存使用率
def get_memory_usage():
    memory_info = psutil.virtual_memory()
    return memory_info.percent

# 获取 SWAP 使用率
def get_swap_usage():
    swap_info = psutil.swap_memory()
    return swap_info.percent

# 发送钉钉报警
def send_dingding_alert(message):
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": f"报警：{message}"  # 加入关键词 "报警"
        }
    }
    print(f"发送消息到钉钉: {message}")  # 调试信息
    try:
        response = requests.post(webhook_url, headers=headers, data=json.dumps(data), timeout=10)
        print(f"钉钉响应状态码: {response.status_code}")  # 调试信息
        print(f"钉钉响应内容: {response.text}")  # 调试信息
        return response.status_code
    except requests.exceptions.RequestException as e:
        print(f"发送消息到钉钉失败: {e}")
        return None

# 主函数
def main():
    print("脚本开始运行...")  # 调试信息
    hostname = get_hostname()
    ip_address = get_ip_address()
    cpu_usage = get_cpu_usage()
    memory_usage = get_memory_usage()
    swap_usage = get_swap_usage()

    print(f"CPU 使用率: {cpu_usage}%")
    print(f"内存使用率: {memory_usage}%")
    print(f"SWAP 使用率: {swap_usage}%")

    if cpu_usage > 50:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ CPU 使用率超过 50%: {cpu_usage:.2f}%"
        send_dingding_alert(message)

    if memory_usage > 50:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ 内存使用率超过 50%: {memory_usage:.2f}%"
        send_dingding_alert(message)

    if swap_usage > 50:
        message = f"主机: {hostname} (IP: {ip_address})\n⚠️ SWAP 使用率超过 50%: {swap_usage:.2f}%"
        send_dingding_alert(message)

if __name__ == "__main__":
    main()

运行结果：