服务器巡检平台

串联前七阶段能力：资产清单 → 并发 SSH 采集 → 规则判定 → 结果入库 → FastAPI 查询 / 定时调度。目标是替代「每台机器手工登录看磁盘、看进程」的重复劳动。

---

架构思路

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Scheduler  │────▶│   Executor   │────▶│  Reporter   │
│ APScheduler │     │ ThreadPool   │     │ SQLite/JSON │
│  / Cron     │     │ + Paramiko   │     │ + 告警推送   │
└─────────────┘     └──────────────┘     └─────────────┘
       ▲                    ▲
       │                    │
┌──────┴──────┐      ┌──────┴──────┐
│  FastAPI    │      │  assets.yaml │
│  手动触发    │      │  主机清单     │
└─────────────┘      └─────────────┘

层次	职责
配置层	assets.yaml：主机、端口、标签、巡检项开关
采集层	SSH 执行只读命令，超时与重试
规则层	磁盘 >85%、load > CPU 核数、服务非 active
存储层	每次巡检一条 run 记录 + 每台主机 detail
接口层	触发巡检、查历史、导出 CSV

---

目录与模块划分

server-inspection/
├── config/          # settings.py, assets.yaml
├── inspection/      # collectors, rules, runner
├── storage/         # repository.py (SQLite)
├── api/main.py
└── scheduler/jobs.py

---

核心代码骨架

assets.yaml

hosts:
  - name: web-01
    ip: 10.0.1.11
    tags: [web, prod]
    checks: [disk, load, nginx]
  - name: db-01
    ip: 10.0.1.21
    tags: [db, prod]
    checks: [disk, load, mysql]

collectors.py

"""单主机采集：返回结构化 dict，失败不抛到线程池外。"""
import paramiko
from dataclasses import dataclass

@dataclass
class HostTarget:
    name: str
    ip: str
    checks: list[str]

def collect_one(host: HostTarget, ssh_user: str, key_path: str, timeout: int = 15) -> dict:
    result = {"host": host.name, "ip": host.ip, "ok": False, "metrics": {}, "errors": []}
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(host.ip, username=ssh_user, key_filename=key_path, timeout=timeout)
        if "disk" in host.checks:
            _, stdout, _ = client.exec_command("df -h / | tail -1", timeout=10)
            result["metrics"]["disk_root"] = stdout.read().decode().strip()
        if "load" in host.checks:
            _, stdout, _ = client.exec_command("cat /proc/loadavg", timeout=10)
            result["metrics"]["loadavg"] = stdout.read().decode().strip()
        result["ok"] = True
    except Exception as e:
        result["errors"].append(str(e))
    finally:
        client.close()
    return result

runner.py

"""线程池批量巡检。"""
from concurrent.futures import ThreadPoolExecutor, as_completed
from inspection.collectors import collect_one, HostTarget
from inspection.rules import evaluate
from storage.repository import save_run

def run_inspection(hosts: list[HostTarget], max_workers: int = 20) -> str:
    details = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(collect_one, h, "ops", "/home/ops/.ssh/id_rsa"): h for h in hosts}
        for fut in as_completed(futures):
            row = fut.result()
            row["violations"] = evaluate(row)  # rules.py：返回违规列表
            details.append(row)
    run_id = save_run(details)
    return run_id

api/main.py

from fastapi import BackgroundTasks, Depends, FastAPI
from scheduler.jobs import load_hosts
from inspection.runner import run_inspection
# from api.auth import verify_api_key  # 复用第七阶段认证

app = FastAPI(title="Server Inspection Platform")

@app.post("/api/v1/inspection/run")
def trigger_run(background_tasks: BackgroundTasks):  # dependencies=[Depends(verify_api_key)]
    hosts = load_hosts()
    background_tasks.add_task(run_inspection, hosts)
    return {"message": "巡检已提交"}

@app.get("/api/v1/inspection/runs/{run_id}")
def get_run(run_id: str):
    # repository.get_run(run_id)
    return {"run_id": run_id, "status": "done"}

---

规则示例（rules.py）

def evaluate(row: dict) -> list[str]:
    violations = []
    disk = row.get("metrics", {}).get("disk_root", "")
    if disk and _parse_use_percent(disk) > 85:
        violations.append("disk_root > 85%")
    load = row.get("metrics", {}).get("loadavg", "")
    if load:
        one_min = float(load.split()[0])
        if one_min > 8:  # 可按 CPU 核数动态化
            violations.append(f"load1={one_min} 过高")
    return violations

---

扩展点

扩展	做法
新巡检项	在 collectors.py 增加命令块，并在 assets.yaml 的 checks 注册
新告警通道	reporter/notify.py 实现 send(run_id)，对接钉钉 / 邮件
多租户	assets.yaml 按 team 分组，API 带 team 参数过滤
大规模	主机 >500 时分片 + 消息队列；Executor 改为 Celery worker
可视化	导出 CSV 给 Grafana / 自建前端；或写入 Prometheus Pushgateway

落地顺序

先跑通「10 台主机 + SQLite + 手动 API 触发」，再加 Cron 与告警。