🏠 首页›实战项目›用Python搭一个VPS自动巡检工具，宕机秒级告警
🚀

用Python搭一个VPS自动巡检工具，宕机秒级告警

手把手教你写一个服务器健康监控脚本，支持CPU/内存/磁盘/端口检测，异常时自动推送钉钉或微信告警。

👁️ - 次阅读

项目背景

跑着云服务器最怕什么？半夜宕机你不知道，第二天客户找你才发现问题。

手动SSH上去看状态太慢了。我试过用Prometheus+Grafana，太重了。后来自己写了个轻量级巡检脚本，200行代码搞定所有需求。

今天把这个方案分享出来，你直接复制就能用。

技术选型

组件选择理由
语言Python 3.8+跨平台，库多
系统信息psutil一行代码拿CPU/内存
网络检测requests + socket检查HTTP和端口
告警推送钉钉Webhook免费，配置简单
定时执行cron系统自带，零依赖

实现步骤

第一步：安装依赖

pip install psutil requests

就这两个库，没有别的。

第二步：编写巡检脚本

创建 vps_inspector.py：

#!/usr/bin/env python3
"""VPS自动巡检脚本 — 每5分钟运行一次，异常时推送告警"""

import os
import sys
import json
import socket
import subprocess
from datetime import datetime
from pathlib import Path

import psutil
import requests

# ============ 配置区 ============

# 告警推送地址（钉钉Webhook）
DINGTALK_WEBHOOK = os.getenv("DINGTALK_WEBHOOK", "")

# 需要监控的HTTP接口列表
MONITORED_URLS = [
    {"name": "主站", "url": "http://localhost:80"},
    {"name": "API服务", "url": "http://localhost:8080/api/health"},
]

# 需要监控的TCP端口
MONITORED_PORTS = [
    {"name": "SSH", "host": "localhost", "port": 22},
    {"name": "Nginx", "host": "localhost", "port": 80},
    {"name": "MySQL", "host": "localhost", "port": 3306},
]

# 阈值配置
CPU_WARN_THRESHOLD = 90      # CPU使用率超过90%告警
MEM_WARN_THRESHOLD = 85      # 内存使用率超过85%告警
DISK_WARN_THRESHOLD = 90     # 磁盘使用率超过90%告警
LOAD_AVG_THRESHOLD = 4       # 负载超过4告警

# 日志路径
LOG_DIR = Path("/var/log/vps-inspector")
LOG_DIR.mkdir(parents=True, exist_ok=True)
LOG_FILE = LOG_DIR / "inspector.log"

# ============ 核心检测函数 ============

def check_cpu():
    """检查CPU使用率和负载"""
    cpu_percent = psutil.cpu_percent(interval=1)
    load_avg = os.getloadavg()
    
    issues = []
    if cpu_percent > CPU_WARN_THRESHOLD:
        issues.append(f"CPU使用率过高: {cpu_percent}%")
    if load_avg[0] > LOAD_AVG_THRESHOLD:
        issues.append(f"系统负载过高: {load_avg[0]:.2f}")
    
    return {
        "metric": "cpu",
        "value": cpu_percent,
        "load_avg": load_avg[0],
        "issues": issues
    }


def check_memory():
    """检查内存使用情况"""
    mem = psutil.virtual_memory()
    
    issues = []
    if mem.percent > MEM_WARN_THRESHOLD:
        issues.append(f"内存使用率过高: {mem.percent}%")
    
    return {
        "metric": "memory",
        "used_percent": mem.percent,
        "total_gb": round(mem.total / 1024**3, 2),
        "available_gb": round(mem.available / 1024**3, 2),
        "issues": issues
    }


def check_disk():
    """检查磁盘空间"""
    disk = psutil.disk_usage("/")
    
    issues = []
    if disk.percent > DISK_WARN_THRESHOLD:
        issues.append(f"磁盘使用率过高: {disk.percent}%")
    
    return {
        "metric": "disk",
        "used_percent": disk.percent,
        "total_gb": round(disk.total / 1024**3, 2),
        "free_gb": round(disk.free / 1024**3, 2),
        "issues": issues
    }


def check_ports():
    """检查指定端口是否开放"""
    results = []
    for target in MONITORED_PORTS:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(3)
            result = sock.connect_ex((target["host"], target["port"]))
            sock.close()
            status = "open" if result == 0 else "closed"
            if status == "closed":
                results.append({
                    "name": target["name"],
                    "status": "down",
                    "issue": f"{target['name']}端口({target['port']})不可达"
                })
            else:
                results.append({"name": target["name"], "status": "up"})
        except Exception as e:
            results.append({
                "name": target["name"],
                "status": "error",
                "issue": str(e)
            })
    return results


def check_http():
    """检查HTTP服务是否响应"""
    results = []
    for target in MONITORED_URLS:
        try:
            resp = requests.get(target["url"], timeout=5)
            if resp.status_code != 200:
                results.append({
                    "name": target["name"],
                    "status": "error",
                    "code": resp.status_code,
                    "issue": f"{target['name']}返回{resp.status_code}"
                })
            else:
                results.append({"name": target["name"], "status": "up", "code": resp.status_code})
        except requests.Timeout:
            results.append({
                "name": target["name"],
                "status": "timeout",
                "issue": f"{target['name']}请求超时"
            })
        except Exception as e:
            results.append({
                "name": target["name"],
                "status": "error",
                "issue": str(e)
            })
    return results


# ============ 告警推送 ============

def send_dingtalk_alert(title, content):
    """发送钉钉告警消息"""
    if not DINGTALK_WEBHOOK:
        print("[INFO] 未配置钉钉Webhook，跳过告警推送")
        return False
    
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": content
        }
    }
    
    try:
        resp = requests.post(DINGTALK_WEBHOOK, json=payload, timeout=10)
        return resp.json().get("errcode") == 0
    except Exception as e:
        print(f"[ERROR] 告警发送失败: {e}")
        return False


def format_alert_report(system_issues, port_results, http_results):
    """格式化告警报告"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    hostname = subprocess.getoutput("hostname")
    
    lines = [f"## 🚨 VPS巡检告警\n\n**时间**: {now}\n**主机**: {hostname}\n\n"]
    
    # 系统指标问题
    all_issues = system_issues.copy()
    for r in port_results + http_results:
        if "issue" in r:
            all_issues.append(r["issue"])
    
    if all_issues:
        lines.append("### 异常项\n")
        for i, issue in enumerate(all_issues, 1):
            lines.append(f"- {i}. {issue}\n")
    else:
        lines.append("### ✅ 全部正常\n\n所有检测项通过。\n")
    
    # 系统资源概览
    lines.append("\n### 系统资源\n")
    lines.append("| 指标 | 数值 |\n|------|------|\n")
    lines.append(f"| CPU使用率 | {system_issues[0]['value']}% |\n")
    lines.append(f"| 内存使用率 | {system_issues[0]['used_percent']}% |\n")
    lines.append(f"| 磁盘使用率 | {system_issues[0]['used_percent']}% |\n")
    
    # 端口状态
    lines.append("\n### 端口状态\n")
    for r in port_results:
        icon = "✅" if r.get("status") == "up" else "❌"
        lines.append(f"- {icon} {r['name']}: {r['status']}\n")
    
    # HTTP状态
    lines.append("\n### HTTP服务\n")
    for r in http_results:
        icon = "✅" if r.get("status") == "up" else "❌"
        lines.append(f"- {icon} {r['name']}: {r.get('status', 'unknown')}\n")
    
    return "".join(lines)


# ============ 主流程 ============

def run_inspection():
    """执行完整巡检"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # 收集所有检测结果
    cpu_result = check_cpu()
    mem_result = check_memory()
    disk_result = check_disk()
    port_results = check_ports()
    http_results = check_http()
    
    # 汇总所有问题
    all_issues = []
    for r in [cpu_result, mem_result, disk_result]:
        all_issues.extend(r.get("issues", []))
    for r in port_results + http_results:
        if "issue" in r:
            all_issues.append(r["issue"])
    
    # 记录日志
    log_entry = {
        "timestamp": timestamp,
        "cpu": cpu_result,
        "memory": mem_result,
        "disk": disk_result,
        "ports": port_results,
        "http": http_results,
        "issues_count": len(all_issues)
    }
    
    log_path = LOG_FILE.parent / f"report_{timestamp}.json"
    with open(log_path, "w") as f:
        json.dump(log_entry, f, indent=2, ensure_ascii=False)
    
    # 追加到总日志
    with open(LOG_FILE, "a") as f:
        f.write(f"[{timestamp}] 巡检完成，发现 {len(all_issues)} 个问题\n")
    
    # 有异常则推送告警
    if all_issues:
        report = format_alert_report(cpu_result, port_results, http_results)
        send_dingtalk_alert("🚨 VPS巡检告警", report)
        print(f"[ALERT] 发现 {len(all_issues)} 个问题，已推送告警")
    else:
        print(f"[OK] 巡检完成，一切正常")
    
    return all_issues


if __name__ == "__main__":
    issues = run_inspection()
    sys.exit(1 if issues else 0)

第三步：配置钉钉告警

打开钉钉群 → 设置 → 智能群助手 → 添加机器人 → 选择自定义（Webhook）。

复制生成的Webhook地址，设置环境变量：

echo 'export DINGTALK_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"' >> ~/.bashrc
source ~/.bashrc

第四步：设置定时任务

crontab -e

添加以下行，每5分钟执行一次：

*/5 * * * * /usr/bin/python3 /opt/scripts/vps_inspector.py >> /var/log/vps-inspector/cron.log 2>&1

也可以用systemd timer，更规范一些：

# /etc/systemd/system/vps-inspector.timer
[Unit]
Description=VPS Auto Inspector Timer

[Timer]
OnBootSec=5min
OnUnitActiveSec=5min

[Install]
WantedBy=timers.target

# /etc/systemd/system/vps-inspector.service
[Unit]
Description=VPS Auto Inspector

[Service]
Type=oneshot
EnvironmentFile=/etc/default/vps-inspector
ExecStart=/usr/bin/python3 /opt/scripts/vps_inspector.py

启动定时器：

sudo systemctl enable --now vps-inspector.timer

运行效果

正常巡检输出：

[2026-07-13 15:05:01] 巡检完成，一切正常

发现异常时的钉钉推送：

🚨 VPS巡检告警

时间: 2026-07-13 15:10:02
主机: prod-web-01

### 异常项
1. CPU使用率过高: 94%
2. MySQL端口(3306)不可达

### 系统资源
| 指标 | 数值 |
|------|------|
| CPU使用率 | 94% |
| 内存使用率 | 72% |
| 磁盘使用率 | 65% |

所有巡检结果会按时间保存到 /var/log/vps-inspector/report_*.json，方便后续分析趋势。

优化方向

这个版本已经能覆盖日常需求了，想进一步打磨可以加这些：

  1. 多实例管理：写个配置文件，一台机器上管理多个VPS的巡检。

  2. 历史趋势图表：把JSON数据导入InfluxDB，用Grafana画出CPU/内存趋势图。

  3. 自动恢复：检测到某个服务挂了，尝试自动重启（比如systemctl restart nginx），而不是只告警。

  4. 邮件备份告警：钉钉之外再加一封邮件，防止错过重要通知。

  5. 白名单机制：维护计划期间的正常停机不要触发告警，加个cron表达式判断。

运维这件事，自动化程度越高，你睡得越踏实。花一个下午搭好这套东西，以后半夜再也不用被告警电话吵醒了。

你现在的服务器是用什么方式监控的？有没有踩过坑？

这篇文章对你有帮助吗？