服务器巡检脚本开发

本章综合 pathlib、subprocess、进程/服务、日志 做一个可扩展的单机巡检脚本：检查磁盘、内存、关键服务、近期 ERROR，输出 JSON 报告。

目标与结构

inspect_server.py
├── collect_disk()      # df 解析
├── collect_memory()    # /proc/meminfo 或 free
├── collect_services()  # systemctl
├── collect_log_errors()# 本地日志 ERROR 计数
└── main()              # 汇总、退出码

完整示例

#!/usr/bin/env python3
"""单机巡检：输出 JSON，异常时 exit 1。"""
from __future__ import annotations

import json
import subprocess
import sys
from dataclasses import dataclass, asdict
from pathlib import Path

SERVICES = ["nginx", "docker"]
ERROR_THRESHOLD = 50
DISK_WARN_PCT = 85

@dataclass
class InspectResult:
    host: str
    disk_ok: bool
    disk_usage_pct: float
    mem_available_mb: float
    services: dict[str, str]
    log_errors_1h: int
    log_source_ok: bool
    ok: bool

def run(cmd: list[str], timeout: int = 15) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)

def collect_disk() -> tuple[bool, float]:
    r = run(["df", "-P", "/"])  # POSIX 输出便于 split
    lines = r.stdout.strip().splitlines()
    if len(lines) < 2:
        return False, 100.0
    parts = lines[1].split()
    use_pct = float(parts[4].rstrip("%"))
    return use_pct < DISK_WARN_PCT, use_pct

def collect_memory() -> float:
    meminfo = Path("/proc/meminfo").read_text()
    kv = {}
    for line in meminfo.splitlines():
        if ":" in line:
            k, v = line.split(":", 1)
            kv[k.strip()] = int(v.strip().split()[0])  # kB
    avail_kb = kv.get("MemAvailable", kv.get("MemFree", 0))
    return avail_kb / 1024

def collect_services() -> dict[str, str]:
    out = {}
    for unit in SERVICES:
        r = run(["systemctl", "is-active", unit])
        out[unit] = r.stdout.strip() if r.returncode == 0 else "inactive"
    return out

def collect_log_errors_1h() -> tuple[bool, int]:
    """统计 journald 最近一小时 err 及以上级别日志。

    不再扫描整个 /var/log/syslog，否则字段名中的“1h”会产生误导。
    无权读取 journal 或主机未使用 systemd 时，将采集失败显式写入报告。
    """
    r = run(
        [
            "journalctl",
            "--since", "1 hour ago",
            "--priority", "err..alert",
            "--no-pager",
            "--output", "json",
        ],
        timeout=30,
    )
    if r.returncode != 0:
        return False, 0
    return True, sum(1 for line in r.stdout.splitlines() if line.strip())

def main() -> int:
    import socket
    disk_ok, disk_pct = collect_disk()
    mem_mb = collect_memory()
    services = collect_services()
    log_source_ok, err_count = collect_log_errors_1h()

    svc_ok = all(v == "active" for v in services.values())
    log_ok = log_source_ok and err_count <= ERROR_THRESHOLD
    ok = disk_ok and svc_ok and log_ok

    result = InspectResult(
        host=socket.gethostname(),
        disk_ok=disk_ok,
        disk_usage_pct=disk_pct,
        mem_available_mb=round(mem_mb, 1),
        services=services,
        log_errors_1h=err_count,
        log_source_ok=log_source_ok,
        ok=ok,
    )
    print(json.dumps(asdict(result), ensure_ascii=False, indent=2))
    return 0 if ok else 1

if __name__ == "__main__":
    sys.exit(main())

运行与扩展

chmod +x inspect_server.py
./inspect_server.py | tee /tmp/inspect.json
echo $?   # 0=健康, 1=有项失败

扩展方向	做法
多主机	第四阶段 SSH 批量执行，或 Ansible
定时	cron / systemd timer
告警	失败时调用 Webhook（见第七阶段 FastAPI）
配置外置	YAML 列出 `SERVICES`、日志查询条件、阈值

提示

巡检脚本要幂等、只读为主；写操作（清理日志、重启服务）单独脚本并加确认开关 --apply。

本例使用 journalctl 的优先级筛选来保证“最近一小时”字段真实可靠。若业务日志只写文件，应改为解析带年份的时间戳，并在无法解析或日志不可读时将该检查标记为失败，而不是返回零错误。

阶段小结

第三阶段掌握：路径与文件 → 安全执行命令 → 进程/服务 → 日志分析 → 综合巡检。下一阶段进入网络：端口探测、HTTP API、SSH 与第三方平台对接。

目标与结构​

完整示例​

运行与扩展​

阶段小结​

目标与结构

完整示例

运行与扩展

阶段小结