用 Python 实现服务器健康检查脚本并自动告警

举报
deli007 发表于 2026/07/25 09:33:17 2026/07/25
【摘要】 轻量级Python服务器监控方案用psutil采集CPU内存磁盘网络指标超过阈值时自动推送企业微信告警配合cron定时执行适合小规模服务器集群

一行命令检查服务器 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/macOS
  • requests 用于调用企业微信机器人 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 团队生成和维护

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。