💡 这不是一篇"10 行脚本教你查看 CPU"的快餐文，而是一份完整的工程化项目复盘——从架构设计、psutil 系统采集、inode 监控暗坑、HTML 可视化报表、钉钉加签告警，到 crontab 定时调度，把所有坑都帮你踩了一遍。
一、为什么我要做这个工具？
故事的开头很朴素：线上一台 API 服务器在凌晨 3 点悄悄挂了，等我们早上发现时，已经丢了 6 小时的订单数据。事后排查发现，根因不是 CPU 也不是内存，而是/var 分区 inode 耗尽——大量小日志文件把 inode 写满，mkdir 直接报 No space left on device，但 df -h 显示磁盘还有 30% 空间。
这件事让我意识到两个痛点：
市面上的方案无非两种：
-云监控服务：按机器数收费，10 台服务器一年几千块
-Shell 脚本拼接：grep + awk 堆出来的"意大利面条"，无法生成可视化报表
作为一个 Python 开发者，我决定自己写一个。但要做得不只是脚本——我要做的是：
这就是这个项目的起点。下面我把整个技术实现拆给你看。
二、整体架构：五层解耦
很多新手做运维工具喜欢"一个 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_favail：非特权用户可用的 inode 数
Python 的 os.statvfs() 直接封装了这个调用：
import osdef 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 = 0for 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/rediskey_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.yamlthresholds: cpu_percent: 80 memory_percent: 80 disk_percent: 80 inode_percent: 80 zombie_process_count: 5key_processes: - sshd - nginx - mysqld - redis-serverkey_ports: - 22 - 80 - 443 - 3306 - 6379disk_partitions: - / - /var - /datadingtalk: webhook: ”https://oapi.dingtalk.com/robot/send?access_token=xxx” secret: ”your-sign-secret” enable_sign: true suppress_interval: 1800# 同告警 30 分钟内不重复发送
配置加载器用"深度合并"模式，用户配置覆盖默认配置：
import yamlclass 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> 服务器巡检报表 - {inspection_data['hostname']} * {{ 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; }} 🖥️ {inspection_data['hostname']} 巡检报表 生成时间：{inspection_data['timestamp']} {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'])}””” with open(output_path, 'w', encoding='utf-8') as f: f.write(html)
报表设计要点：
- 单文件 HTML，无 CDN 依赖，邮件附件直接打开
- 颜色编码：绿色 OK / 橙色 Warning / 红色 Critical
六、钉钉告警：加签认证 + 告警抑制
6.1 钉钉机器人安全机制
钉钉自定义机器人支持三种安全策略：自定义关键词、加签（签名加密）、IP 地址段。生产环境强烈推荐"加签"模式——它与开发者双向进行安全认证。
6.2 加签算法实现
钉钉加签的规则是：把 timestamp + "\n" + secret 做 HmacSHA256 加密，再进行 Base64 编码，最后 URL Encode。
import timeimport hmacimport hashlibimport base64import urllib.parseimport requestsclass 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}×tamp={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/bashSCRIPT_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 核心知识点得到了实战演练：
九、抽象升华：从服务器巡检到通用监控框架
这个项目的真正价值，不在于"巡检 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+ 行代码，每一行都有详细注释。这个体量对于一个"能写进简历的工程级项目"来说，恰到好处——既体现了复杂度，又不会大到让人望而生畏。
写在最后
这个项目从需求提出到最终上线，我花了大约三周的业余时间。但三周学到的东西，比我过去一年看的运维教程都多——因为只有真正要解决一个具体生产问题，你才会去思考架构、性能、异常、用户体验。
技术栈：Python 3.8+ / psutil 5.9+ / PyYAML 6.0+ / Requests 2.28+ / HTML5 + CSS3
完整代码下载地址：
https://download.csdn.net/download/2301_76484015/93175666
10个月宝宝每天需要喝多少奶粉？
