用 Python 实现服务器健康检查脚本并自动告警
一行命令检查服务器 CPU/内存/磁盘/网络,异常自动推送企业微信告警
分类: ai-development
标签: Python, 运维, 监控
难度: 初级
读者: 运维工程师、后端开发者
背景
在管理华为云 ECS 服务器时,经常遇到 CPU 飙高、磁盘写满、内存泄漏等问题。手动 SSH 登录逐台检查效率低下,等到用户反馈时往往已经造成事故。
问题描述
传统监控方案(Prometheus、Zabbix)部署复杂,对于 2-3 台小规模服务器来说太重了。我们需要一个轻量级的 Python 脚本:
- 一键检查 CPU、内存、磁盘、网络状态
- 异常时自动推送企业微信机器人告警
- 支持 cron 定时执行
- 零依赖(仅用 Python 标准库 + requests)
适用场景
- 小规模服务器集群(1-10 台)
- 个人开发者或小团队的轻量监控
- 作为专业监控系统的补充检查
解决方案
方案概述
用 Python psutil 库采集服务器指标,与预设阈值比较,异常时通过企业微信机器人 Webhook 推送告警消息。脚本支持单次检查和守护进程两种模式,配合 cron 定时执行。
流程: 采集指标(CPU/内存/磁盘/网络) → 阈值比较(80%/85%/90%) → 超限则推送企业微信告警
实现步骤
步骤 1: 安装依赖
pip install psutil requests
关键说明:
psutil是跨平台的系统监控库,支持 Linux/Windows/macOSrequests用于调用企业微信机器人 Webhook- 无需 root 权限即可采集 CPU/内存/磁盘指标
步骤 2: 编写健康检查脚本
#!/usr/bin/env python3
"""服务器健康检查脚本 — 检查 CPU/内存/磁盘/网络,异常自动告警"""
import psutil
import requests
import json
import platform
import socket
from datetime import datetime
# ===== 配置区 =====
THRESHOLDS = {
"cpu_percent": 80, # CPU 使用率阈值 (%)
"memory_percent": 85, # 内存使用率阈值 (%)
"disk_percent": 90, # 磁盘使用率阈值 (%)
"disk_path": "/", # 检查的磁盘路径
}
# 企业微信机器人 Webhook(替换为你的)
WEBHOOK_URL = "YOUR_WEBHOOK_URL" # 替换为企业微信机器人 Webhook 地址
# ===== 采集函数 =====
def collect_metrics():
"""采集服务器各项指标"""
metrics = {}
# CPU
metrics["cpu_percent"] = psutil.cpu_percent(interval=1)
metrics["cpu_count"] = psutil.cpu_count()
# 内存
mem = psutil.virtual_memory()
metrics["mem_percent"] = mem.percent
metrics["mem_total_gb"] = round(mem.total / 1024**3, 1)
metrics["mem_used_gb"] = round(mem.used / 1024**3, 1)
# 磁盘
disk = psutil.disk_usage(THRESHOLDS["disk_path"])
metrics["disk_percent"] = disk.percent
metrics["disk_total_gb"] = round(disk.total / 1024**3, 1)
metrics["disk_used_gb"] = round(disk.used / 1024**3, 1)
# 网络
net = psutil.net_io_counters()
metrics["net_sent_mb"] = round(net.bytes_sent / 1024**2, 1)
metrics["net_recv_mb"] = round(net.bytes_recv / 1024**2, 1)
# 系统信息
metrics["hostname"] = socket.gethostname()
metrics["os"] = platform.platform()
metrics["uptime"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return metrics
def check_alerts(metrics):
"""检查是否有指标超过阈值"""
alerts = []
if metrics["cpu_percent"] > THRESHOLDS["cpu_percent"]:
alerts.append(f"⚠️ CPU 使用率 {metrics['cpu_percent']}% 超过阈值 {THRESHOLDS['cpu_percent']}%")
if metrics["mem_percent"] > THRESHOLDS["memory_percent"]:
alerts.append(f"⚠️ 内存使用率 {metrics['mem_percent']}% 超过阈值 {THRESHOLDS['memory_percent']}%")
if metrics["disk_percent"] > THRESHOLDS["disk_percent"]:
alerts.append(f"⚠️ 磁盘使用率 {metrics['disk_percent']}% 超过阈值 {THRESHOLDS['disk_percent']}%")
return alerts
def send_alert(alerts, metrics):
"""通过企业微信机器人推送告警"""
content = f"""【服务器健康告警】
主机: {metrics['hostname']}
时间: {metrics['uptime']}
{chr(10).join(alerts)}
当前状态:
- CPU: {metrics['cpu_percent']}% ({metrics['cpu_count']}核)
- 内存: {metrics['mem_used_gb']}/{metrics['mem_total_gb']}GB ({metrics['mem_percent']}%)
- 磁盘: {metrics['disk_used_gb']}/{metrics['disk_total_gb']}GB ({metrics['disk_percent']}%)
"""
payload = {
"msgtype": "text",
"text": {"content": content}
}
try:
resp = requests.post(WEBHOOK_URL, json=payload, timeout=10)
return resp.json().get("errcode") == 0
except Exception as e:
print(f"告警推送失败: {e}")
return False
def print_report(metrics):
"""打印检查报告"""
print(f"""
╔══════════════════════════════════════════╗
║ 服务器健康检查报告
╠══════════════════════════════════════════╣
║ 主机名: {metrics['hostname']}
║ 操作系统: {metrics['os']}
║ 检查时间: {metrics['uptime']}
╠══════════════════════════════════════════╣
║ CPU 使用率: {metrics['cpu_percent']}% ({metrics['cpu_count']}核)
║ 内存使用率: {metrics['mem_percent']}% ({metrics['mem_used_gb']}/{metrics['mem_total_gb']}GB)
║ 磁盘使用率: {metrics['disk_percent']}% ({metrics['disk_used_gb']}/{metrics['disk_total_gb']}GB)
║ 网络流量: 发送 {metrics['net_sent_mb']}MB / 接收 {metrics['net_recv_mb']}MB
╚══════════════════════════════════════════╝
""")
def main():
"""主函数: 采集 → 检查 → 告警"""
# 1. 采集指标
metrics = collect_metrics()
# 2. 打印报告
print_report(metrics)
# 3. 检查告警
alerts = check_alerts(metrics)
if alerts:
print(f"\n🚨 发现 {len(alerts)} 个告警:")
for alert in alerts:
print(f" {alert}")
# 4. 推送告警
success = send_alert(alerts, metrics)
if success:
print("✅ 告警已推送到企业微信")
else:
print("❌ 告警推送失败")
else:
print("\n✅ 所有指标正常")
if __name__ == "__main__":
main()
关键说明:
psutil.cpu_percent(interval=1)需要间隔 1 秒才能获得准确值- 企业微信机器人 Webhook 免费且配置简单,适合小团队
- 告警内容包含主机名和详细指标,方便快速定位问题
步骤 3: 配置 cron 定时执行
# 编辑 crontab
crontab -e
# 每 5 分钟检查一次
*/5 * * * * /usr/bin/python3 /opt/health-check/health_check.py >> /var/log/health-check.log 2>&1
# 每天早上 9 点发送日报
0 9 * * * /usr/bin/python3 /opt/health-check/health_check.py --report >> /var/log/health-check.log 2>&1
完整示例
将脚本保存为 health_check.py,直接运行:
$ python3 health_check.py
运行结果:
╔══════════════════════════════════════════╗
║ 服务器健康检查报告
╠══════════════════════════════════════════╣
║ 主机名: ecs-ai-ops-001
║ 操作系统: Linux-5.10.0-x86_64-with-glibc2.31
║ 检查时间: 2026-07-25 01:30:00
╠══════════════════════════════════════════╣
║ CPU 使用率: 12.5% (4核)
║ 内存使用率: 45.2% (3.6/8.0GB)
║ 磁盘使用率: 67.8% (33.9/50.0GB)
║ 网络流量: 发送 1256.3MB / 接收 3421.7MB
╚══════════════════════════════════════════╝
✅ 所有指标正常
当 CPU 超过 80% 时,企业微信收到告警:
【服务器健康告警】
主机: ecs-ai-ops-001
时间: 2026-07-25 01:35:00
⚠️ CPU 使用率 85.3% 超过阈值 80%
当前状态:
- CPU: 85.3% (4核)
- 内存: 3.6/8.0GB (45.2%)
- 磁盘: 33.9/50.0GB (67.8%)
注意事项
- ⚠️ psutil.cpu_percent() 第一次调用返回 0.0,需要
interval=1参数或先调用一次预热 - ⚠️ 企业微信 Webhook 有频率限制(每分钟 20 条),告警去重逻辑建议在脚本中实现
- ⚠️ 磁盘路径:Linux 用
/,Windows 用C:\\,根据系统调整THRESHOLDS["disk_path"] - ⚠️ 权限:普通用户即可运行,无需 root/sudo
最佳实践
- ✅ 告警去重:同一指标 5 分钟内只告警一次,避免刷屏
- ✅ 分级告警:CPU > 80% 为 warning,> 95% 为 critical,不同级别推送到不同群
- ✅ 历史记录:将每次检查结果写入 SQLite,方便回溯问题
- ✅ 多机检查:配合 SSH 或 API 批量检查多台服务器,汇总告警
常见问题
Q1: psutil 安装失败怎么办?
A: 尝试 pip install psutil --no-build-isolation,或用系统包管理器安装:apt install python3-psutil(Ubuntu)
Q2: 企业微信机器人怎么配置?
A: 企业微信管理后台 → 应用管理 → 自建应用 → 接收消息 → 设置 Webhook → 复制 URL 填入脚本
Q3: 能否用钉钉/飞书替代企业微信?
A: 可以。钉钉和飞书都有类似的机器人 Webhook,只需修改 send_alert 函数中的 payload 格式和 URL
本文由 developer-ecosystem 团队生成和维护
- 点赞
- 收藏
- 关注作者
评论(0)