💡 这不是一篇"10 行脚本教你查看 CPU"的快餐文，而是一份完整的工程化项目复盘——从架构设计、psutil 系统采集、inode 监控暗坑、HTML 可视化报表、钉钉加签告警，到 crontab 定时调度，把所有坑都帮你踩了一遍。

---

一、为什么我要做这个工具？

故事的开头很朴素：线上一台 API 服务器在凌晨 3 点悄悄挂了，等我们早上发现时，已经丢了 6 小时的订单数据。事后排查发现，根因不是 CPU 也不是内存，而是 /var 分区 inode 耗尽——大量小日志文件把 inode 写满，mkdir 直接报 No space left on device，但 df -h 显示磁盘还有 30% 空间。

这件事让我意识到两个痛点：

1. 传统监控太重：Zabbix、Prometheus 部署复杂，小团队用不起
2. 人工巡检太低效：每天 SSH 登录十几台机器敲 top df -i ps，耗时且易漏检

市面上的方案无非两种：

• 云监控服务：按机器数收费，10 台服务器一年几千块
• Shell 脚本拼接：grep + awk 堆出来的"意大利面条"，无法生成可视化报表

作为一个 Python 开发者，我决定自己写一个。但要做得不只是脚本——我要做的是：

1. 用 psutil 一站式采集系统指标（官方定义它就是用来获取 CPU、内存、磁盘、网络、进程等系统利用率的跨平台库）
2. 配置与代码完全分离，阈值可在 YAML 里调
3. 生成 HTML 可视化报表，历史可追溯
4. 超过阈值 80% 时通过钉钉机器人实时告警
5. 支持 crontab 定时执行，无人值守

这就是这个项目的起点。下面我把整个技术实现拆给你看。

---

二、整体架构：五层解耦

很多新手做运维工具喜欢"一个 py 文件写到尾"。但我采用了五层解耦架构，这也是工业级 Python 项目的标准做法：

┌──────────────────────────────────────────────┐
│  入口层：main.py（CLI 入口，支持 crontab）       │
├──────────────────────────────────────────────┤
│  配置层：config_loader.py（YAML 解析+深度合并）  │
├──────────────────────────────────────────────┤
│  采集层：inspector.py（psutil 10 大类指标采集）  │
├──────────────────────────────────────────────┤
│  输出层：html_report.py（可视化报表生成）        │
├──────────────────────────────────────────────┤
│  告警层：dingtalk_alert.py（钉钉加签+抑制）      │
└──────────────────────────────────────────────┘

为什么要这么拆？

• 配置外置 → 改阈值不用动代码，重启即生效
• 采集与输出解耦 → 巡检引擎不关心报表格式，换 JSON/Excel 只需新增输出模块
• 告警独立 → 钉钉 webhook 挂了不影响巡检主流程

---

三、核心采集：psutil 的工程化用法

3.1 为什么选 psutil？

psutil 是 Python 生态中下载量排名前 100 的系统信息采集库，官方定位就是"用 Python 获取运行进程和系统利用率（CPU、内存、磁盘、网络、传感器）信息"，它实现了 ps、top、free、iostat、netstat 等 Unix 命令的功能。

跨平台支持 Linux/Windows/macOS/FreeBSD 等，支持 Python 3.6+。

3.2 CPU 采集的正确姿势

import psutil

# ✅ 正确：interval=1 阻塞采样，得到准确的 CPU 使用率
cpu_percent = psutil.cpu_percent(interval=1)

# ✅ 正确：获取每核使用率
cpu_per_core = psutil.cpu_percent(interval=1, percpu=True)

# ✅ 获取 CPU 时间分布（user/system/idle/iowait...）
cpu_times = psutil.cpu_times()

⚠️ 关键坑：cpu_percent(interval=None) 第一次调用永远返回 0.0（无意义值）。官方文档明确说明：第一次调用会与模块导入时的 CPU 时间做比较，建议两次调用间隔至少 0.1 秒。所以在巡检脚本里，务必用 interval=1 做阻塞采样，牺牲 1 秒换取准确性。

3.3 内存采集

mem = psutil.virtual_memory()
print(f"内存使用率: {mem.percent}%")
print(f"已用: {mem.used / 1024 / 1024 / 1024:.2f} GB")
print(f"可用: {mem.available / 1024 / 1024 / 1024:.2f} GB")

virtual_memory() 返回的 percent 字段直接是百分比，比自己算 used/total 更准（它考虑了 available 口径）。

3.4 磁盘分区与使用率

# 获取所有挂载分区
partitions = psutil.disk_partitions()
for part in partitions:
    usage = psutil.disk_usage(part.mountpoint)
    print(f"{part.mountpoint}: {usage.percent}%")

3.5 Inode 监控：最容易被忽视的暗坑

这是整个工具最有价值的部分。很多开发者监控磁盘只盯 disk_usage().percent（块使用率），但 inode 耗尽和块耗尽是两回事。

Linux 通过 statvfs() 系统调用获取文件系统信息，关键字段：

• f_files：文件系统 inode 总量
• f_ffree：空闲 inode 数
• f_favail：非特权用户可用的 inode 数

Python 的 os.statvfs() 直接封装了这个调用：

import os

def get_inode_usage(path="/"):
    """获取指定挂载点的 inode 使用率"""
    stat = os.statvfs(path)
    total_inodes = stat.f_files
    free_inodes = stat.f_favail
    
    if total_inodes == 0:
        # ⚠️ 某些文件系统（如 ZFS）f_files 可能返回 0 或伪造值
        return None
    
    used_inodes = total_inodes - free_inodes
    usage_percent = (used_inodes / total_inodes) * 100
    return round(usage_percent, 2)

💡 真实案例：我们的 /var 分区 inode 使用率到 100% 时，磁盘块使用率才 70%。如果只监控块使用率，永远发现不了这个问题。inode 耗尽的典型场景：大量小文件（如日志、邮件队列、Session 文件）。

⚠️ 进阶坑：ZFS 等现代文件系统没有固定 inode 数量，而是根据可用空间动态"编造"一个数字。所以在 ZFS 上监控 inode 使用率意义不大，代码里要做 fallback 处理。

3.6 网络丢包率

net = psutil.net_io_counters(pernic=True)
eth0 = net.get('eth0', net.get('ens33'))  # 兼容不同网卡名
if eth0:
    total_packets = eth0.packets_sent + eth0.packets_recv
    dropped = eth0.dropin + eth0.dropout
    loss_rate = (dropped / total_packets * 100) if total_packets > 0 else 0

3.7 僵尸进程检测

zombie_count = 0
for proc in psutil.process_iter(['pid', 'name', 'status']):
    try:
        if proc.info['status'] == psutil.STATUS_ZOMBIE:
            zombie_count += 1
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass

psutil.process_iter() 是官方推荐的进程遍历方式，比 psutil.pids() 更安全（避免竞态条件）。

3.8 关键进程存活性

def check_process_alive(process_names):
    """检查关键进程是否在运行"""
    results = {}
    for name in process_names:
        found = False
        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
            try:
                if name in proc.info['name'] or \
                   (proc.info['cmdline'] and name in ' '.join(proc.info['cmdline'])):
                    found = True
                    break
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        results[name] = found
    return results

# 检查 sshd/nginx/mysql/redis
key_processes = ['sshd', 'nginx', 'mysqld', 'redis-server']
alive_status = check_process_alive(key_processes)

3.9 关键端口监听状态

def check_port_listening(ports):
    """检查指定端口是否被监听"""
    listening_ports = set()
    for conn in psutil.net_connections(kind='tcp'):
        if conn.status == 'LISTEN':
            listening_ports.add(conn.laddr.port)
    
    results = {}
    for port in ports:
        results[port] = port in listening_ports
    return results

# 检查关键端口
key_ports = [22, 80, 443, 3306, 6379]
port_status = check_port_listening(key_ports)

---

四、配置与代码分离：YAML 驱动设计

硬编码是工程化的大敌。我把所有阈值、进程名、端口号都外置到 config.yaml：

# config.yaml
thresholds:
  cpu_percent: 80
  memory_percent: 80
  disk_percent: 80
  inode_percent: 80
  zombie_process_count: 5

key_processes:
  - sshd
  - nginx
  - mysqld
  - redis-server

key_ports:
  - 22
  - 80
  - 443
  - 3306
  - 6379

disk_partitions:
  - /
  - /var
  - /data

dingtalk:
  webhook: "https://oapi.dingtalk.com/robot/send?access_token=xxx"
  secret: "your-sign-secret"
  enable_sign: true
  suppress_interval: 1800  # 同告警 30 分钟内不重复发送

配置加载器用"深度合并"模式，用户配置覆盖默认配置：

import yaml

class ConfigLoader:
    def __init__(self, config_path="config.yaml"):
        self.config_path = config_path
        self.config = self._deep_merge(self._default_config(), self._load_user_config())
    
    def _load_user_config(self):
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or {}
        except FileNotFoundError:
            return {}
    
    def _deep_merge(self, default, user):
        """深度合并：用户配置覆盖默认配置"""
        result = default.copy()
        for key, value in user.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = self._deep_merge(result[key], value)
            else:
                result[key] = value
        return result

---

五、HTML 报表：让数据"看得见"

巡检数据如果只是打印到终端，价值减半。我设计了自包含的 HTML 报表（单文件，无外部依赖），用内联 CSS 实现卡片式布局：

class HTMLReportGenerator:
    def generate(self, inspection_data, output_path):
        """生成自包含 HTML 报表"""
        html = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>服务器巡检报表 - {inspection_data['hostname']}</title>
    <style>
        * {{ margin: 0; padding: 0; box-sizing: border-box; }}
        body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI"; background: #f5f7fa; }}
        .header {{ background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); 
                  color: white; padding: 30px; }}
        .card-container {{ display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); 
                         gap: 20px; padding: 20px; }}
        .card {{ background: white; border-radius: 12px; padding: 20px; 
                box-shadow: 0 2px 12px rgba(0,0,0,0.08); }}
        .card.critical {{ border-left: 5px solid #e74c3c; }}
        .card.warning {{ border-left: 5px solid #f39c12; }}
        .card.ok {{ border-left: 5px solid #27ae60; }}
        .metric-value {{ font-size: 32px; font-weight: bold; margin: 10px 0; }}
        .critical .metric-value {{ color: #e74c3c; }}
    </style>
</head>
<body>
    <div class="header">
        <h1>🖥️ {inspection_data['hostname']} 巡检报表</h1>
        <p>生成时间：{inspection_data['timestamp']}</p>
    </div>
    <div class="card-container">
        {self._render_cpu_card(inspection_data['cpu'])}
        {self._render_memory_card(inspection_data['memory'])}
        {self._render_disk_cards(inspection_data['disks'])}
        {self._render_inode_cards(inspection_data['inodes'])}
    </div>
</body>
</html>
"""
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(html)

报表设计要点：

• 单文件 HTML，无 CDN 依赖，邮件附件直接打开
• 卡片式布局，响应式适配
• 颜色编码：绿色 OK / 橙色 Warning / 红色 Critical
• 超过 80% 阈值的指标自动标红 + 脉冲动画

---

六、钉钉告警：加签认证 + 告警抑制

6.1 钉钉机器人安全机制

钉钉自定义机器人支持三种安全策略：自定义关键词、加签（签名加密）、IP 地址段。生产环境强烈推荐"加签"模式——它与开发者双向进行安全认证。

6.2 加签算法实现

钉钉加签的规则是：把 timestamp + "\n" + secret 做 HmacSHA256 加密，再进行 Base64 编码，最后 URL Encode。

import time
import hmac
import hashlib
import base64
import urllib.parse
import requests

class DingTalkAlert:
    def __init__(self, webhook, secret):
        self.webhook = webhook
        self.secret = secret
        self.alert_cache = {}  # 告警抑制缓存
    
    def _sign(self):
        """生成钉钉加签"""
        timestamp = str(round(time.time() * 1000))
        string_to_sign = f"{timestamp}\n{self.secret}"
        hmac_code = hmac.new(
            self.secret.encode("utf-8"),
            string_to_sign.encode("utf-8"),
            digestmod=hashlib.sha256
        ).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return timestamp, sign
    
    def send_alert(self, title, content, is_critical=True):
        """发送钉钉告警"""
        timestamp, sign = self._sign()
        url = f"{self.webhook}&timestamp={timestamp}&sign={sign}"
        
        # 构造 Markdown 消息
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": f"## 🚨 {title}\n\n{content}\n\n> 时间：{time.strftime('%Y-%m-%d %H:%M:%S')}"
            }
        }
        
        try:
            resp = requests.post(url, json=data, timeout=5)
            result = resp.json()
            if result.get("errcode") == 0:
                print("✅ 钉钉告警发送成功")
            else:
                print(f"❌ 钉钉告警发送失败: {result}")
        except Exception as e:
            print(f"❌ 钉钉告警异常: {e}")

6.3 告警抑制：避免半夜被刷屏

def should_suppress(self, alert_key, suppress_interval=1800):
    """同告警 30 分钟内不重复发送"""
    now = time.time()
    last_sent = self.alert_cache.get(alert_key)
    if last_sent and (now - last_sent) < suppress_interval:
        return True
    self.alert_cache[alert_key] = now
    return False

# 使用示例
if cpu_percent > 80:
    alert_key = f"cpu_{hostname}"
    if not should_suppress(alert_key):
        dingtalk.send_alert(
            title="CPU 使用率超阈值",
            content=f"当前 CPU 使用率 **{cpu_percent}%**，超过阈值 80%"
        )

⚠️ 工程经验：没有抑制机制的告警系统，在持续超阈值时会每秒发一条消息，半夜能把运维人员逼疯。30 分钟抑制间隔是业界最佳实践。

---

七、crontab 定时执行：无人值守的关键

巡检脚本的价值在于"无人值守"。通过 crontab 每 5 分钟执行一次：

# 安装 crontab（install_cron.sh）
#!/bin/bash
SCRIPT_PATH="/opt/inspector/main.py"
PYTHON_BIN="/usr/bin/python3"

# 每 5 分钟执行一次巡检
CRON_JOB="*/5 * * * * $PYTHON_BIN $SCRIPT_PATH >> /var/log/inspector.log 2>&1"

# 检查是否已安装
(crontab -l 2>/dev/null | grep -F "$SCRIPT_PATH") || \
(crontab -l 2>/dev/null; echo "$CRON_JOB") | crontab -

echo "✅ crontab 安装完成：每 5 分钟巡检一次"

main.py 的退出码设计（让 crontab 能感知异常）：

def main():
    config = ConfigLoader().config
    inspector = Inspector(config)
    results = inspector.run_all_checks()
    
    # 生成 HTML 报表
    report_path = HTMLReportGenerator().generate(results, "/var/log/inspector_report.html")
    
    # 钉钉告警
    alert_manager = DingTalkAlert(config['dingtalk']['webhook'], config['dingtalk']['secret'])
    critical_count = results['summary']['critical']
    
    if critical_count > 0:
        alert_manager.send_alert(...)
        return 2  # CRITICAL
    elif results['summary']['warning'] > 0:
        return 1  # WARNING
    else:
        return 0  # OK

退出码语义化后，可以配合 cron + mail 实现"异常才邮件通知"的二级告警。

---

八、关键技术点总结

回顾整个项目，以下 Python 核心知识点得到了实战演练：

知识点	应用场景
psutil 系统采集	CPU/内存/磁盘/网络/进程一站式采集
cpu_percent interval 参数	避免首次调用返回 0.0 的坑
statvfs inode 监控	发现"磁盘未满但 inode 耗尽"的隐蔽故障
process_iter 安全遍历	避免进程遍历的竞态条件
YAML 配置深度合并	用户配置优雅覆盖默认配置
HTML 报表自包含生成	单文件、无 CDN 依赖、邮件可直接打开
HmacSHA256 加签	钉钉机器人安全认证
告警抑制算法	避免重复告警轰炸
退出码语义化	让 crontab 能感知脚本执行状态
crontab 自动化	实现无人值守巡检

---

九、抽象升华：从服务器巡检到通用监控框架

这个项目的真正价值，不在于"巡检 Linux"本身，而在于它提供了一个通用的"定时采集 → 判断阈值 → 多渠道告警"架构模板：

任何"周期性指标采集 + 阈值判断 + 通知"的场景都能套用：
┌─────────────────────────────────────────┐
│  场景                  │  采集源          │
├─────────────────────────────────────────┤
│  Linux 服务器巡检     │  psutil          │
│  MySQL 慢查询监控     │  information_schema│
│  Redis 内存告警       │  INFO 命令        │
│  Docker 容器健康检查  │  docker stats    │
│  API 接口存活探测     │  requests        │
│  网站 SSL 证书过期    │  ssl socket      │
└─────────────────────────────────────────┘

只要把 inspector.py 里的采集函数换成对应数据源，配置、报表、告警、crontab 调度全部可以复用。这就是为什么我强调"工程化架构"的重要性——一次设计，处处适用。

---

十、项目文件结构

最终交付的项目结构清晰、模块化：

inspector/
├── main.py              # 主入口，串联全流程
├── config.yaml          # 配置文件（阈值/进程/端口外置）
├── config_loader.py     # 配置加载器
├── inspector.py         # 核心引擎，10 大类巡检
├── html_report.py       # HTML 报表生成器
├── dingtalk_alert.py    # 钉钉告警模块
├── install_cron.sh      # 一键安装 crontab
├── requirements.txt     # psutil/PyYAML/requests
└── README.md            # 完整使用文档

总计 1500+ 行代码，每一行都有详细注释。这个体量对于一个"能写进简历的工程级项目"来说，恰到好处——既体现了复杂度，又不会大到让人望而生畏。

---

写在最后

这个项目从需求提出到最终上线，我花了大约三周的业余时间。但三周学到的东西，比我过去一年看的运维教程都多——因为只有真正要解决一个具体生产问题，你才会去思考架构、性能、异常、用户体验。

如果你也想做类似的项目，我的建议是：

1. 从一个真实痛点出发，不要为了做项目而做项目（我的出发点是那次 inode 耗尽的事故）
2. 先跑通最小可用版本（MVP），再逐步优化
3. 重视架构解耦，采集、配置、输出、告警一定要分离
4. 异常和边界条件处理到位，这是区分"玩具"和"工具"的关键
5. 定时调度，让脚本真正"活"在生产环境里

🎁 这个项目的完整源码（1500+ 行，带详细注释）、配置文件、HTML 报表模板、钉钉告警模块、crontab 安装脚本，我已经打包上传到 CSDN，搜索"Linux 服务器自动巡检 Python"即可找到。

如果你在实践过程中遇到任何问题，或者有更好的优化思路，欢迎在评论区交流。毕竟，Python 的魅力，就在于它让我们能用代码实实在在地解决问题，而不是停留在纸面上。

---

完整代码下载地址： Linux 服务器自动巡检系统

技术栈：Python 3.8+ / psutil 5.9+ / PyYAML 6.0+ / Requests 2.28+ / HTML5 + CSS3