Python 运维自动化：并发批量管理远程主机

目标

写一个能在 100 台主机上并发执行命令、并汇总结果的工具。要求：

并发执行，单台超时不拖垮整体；
输出结构化（JSON），方便后续入库或告警；
失败可区分：连接失败 / 超时 / 命令返回非零。

为什么不用 subprocess 调 ssh

subprocess + ssh 命令当然可以，但：

每台机器 fork 一个进程，100 台就是 100 个进程；
密码/密钥交互难以程序化处理；
输出解析依赖 stdout 格式，脆弱。

Paramiko 是纯 Python 的 SSH2 实现，可以直接控制会话，更适合程序化场景。

核心实现

使用 concurrent.futures.ThreadPoolExecutor——SSH 是 IO 密集型，线程模型足够且简单。

import json
import socket
from concurrent.futures import ThreadPoolExecutor, as_completed

import paramiko


def run_on_host(host: str, command: str, timeout: int = 8) -> dict:
    """在单台主机执行命令，返回结构化结果。"""
    result = {"host": host, "ok": False, "code": None, "stdout": "", "error": ""}
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(
            hostname=host,
            username="ops",
            key_filename="/home/ops/.ssh/id_ed25519",
            timeout=timeout,
            banner_timeout=timeout,
            auth_timeout=timeout,
        )
        _, stdout, stderr = client.exec_command(command, timeout=timeout)
        result["stdout"] = stdout.read().decode(errors="replace").strip()
        result["code"] = stdout.channel.recv_exit_status()
        result["ok"] = result["code"] == 0
        if not result["ok"]:
            result["error"] = stderr.read().decode(errors="replace").strip()
    except (paramiko.SSHException, socket.error, OSError) as exc:
        result["error"] = f"{type(exc).__name__}: {exc}"
    finally:
        client.close()
    return result


def run_on_all(hosts: list[str], command: str, workers: int = 20) -> list[dict]:
    results = []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(run_on_host, h, command): h for h in hosts}
        for future in as_completed(futures):
            host = futures[future]
            try:
                results.append(future.result())
            except Exception as exc:  # 兜底，避免单个异常中断汇总
                results.append({"host": host, "ok": False, "error": repr(exc)})
    return sorted(results, key=lambda r: r["host"])


if __name__ == "__main__":
    hosts = [f"10.0.1.{i}" for i in range(10, 30)]
    out = run_on_all(hosts, "uptime && df -h / | tail -1")
    failed = [r for r in out if not r["ok"]]
    print(json.dumps({"total": len(out), "failed": len(failed), "detail": failed},
                     ensure_ascii=False, indent=2))

几个关键设计点

1. 超时必须显式设置

connect() 有三个超时参数，缺一不可：

参数	作用
`timeout`	TCP 连接超时
`banner_timeout`	等待服务端 banner
`auth_timeout`	认证超时

只设 timeout 时，网络通但 SSH 服务假死的主机仍会挂住线程。

2. workers 与目标主机数量的关系

SSH 握手是 CPU 加密操作，20~50 个 worker 在普通笔记本上已足够；
目标主机有 MaxStartups 限制（sshd 默认 10:30:100），并发过高会被拒连；
稳妥做法是 20 起步，按失败率动态调整。

3. 结构化输出的价值

返回 list[dict] 而不是拼接字符串，带来的好处：

可以直接 json.dump 供上层消费；
失败项可以按 error 类型分组统计；
天然适配后续写入 SQLite / 推送告警。

从脚本到工具

三点渐进式改造：

加配置文件：主机列表从 hosts.yaml 读取，不再硬编码。
加并发控制：用 as_completed 做流式输出，长任务也能实时看到进度。
加幂等性：执行的命令尽量设计成可重复执行，避免重试导致状态叠加。

小结

SSH 批量操作的本质是 IO 密集型并发，线程池是性价比最高的方案。
一定要区分「连不上」和「命令失败」——处理方式完全不同。
输出结构化，是这个脚本能否被复用的分水岭。