paramiko：SSH 批量管理基础

paramiko 是 Python 实现的 SSHv2 协议库，是运维自动化的底层基石。当需要精细控制 SSH 连接、处理非标准场景时，paramiko 提供了最大灵活性。

基础连接与命令执行

import paramiko
import time

def ssh_exec(host, port, username, password, command):
    """基础 SSH 命令执行"""
    client = paramiko.SSHClient()
    # 自动添加主机密钥（生产环境建议使用 known_hosts）
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(host, port=port, username=username, password=password, timeout=10)
        stdin, stdout, stderr = client.exec_command(command)
        # 获取退出码
        exit_code = stdout.channel.recv_exit_status()
        output = stdout.read().decode().strip()
        error = stderr.read().decode().strip()
        return {
            'host': host,
            'exit_code': exit_code,
            'output': output,
            'error': error
        }
    finally:
        client.close()

# 使用示例
result = ssh_exec('10.0.1.10', 22, 'root', 'password', 'uname -r')
print(f"{result['host']}: exit={result['exit_code']}, output={result['output']}")

连接池与并发执行

生产环境中批量管理数百台服务器时，需要连接池和并发控制：

import paramiko
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock

class SSHConnectionPool:
    """SSH 连接池，复用连接减少握手开销"""

    def __init__(self, max_connections=20):
        self.pool = {}          # {host: SSHClient}
        self.lock = Lock()
        self.max_connections = max_connections

    def get_connection(self, host, port, username, password):
        with self.lock:
            if host in self.pool and self.pool[host].get_transport().is_active():
                return self.pool[host]
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.connect(host, port=port, username=username, password=password, timeout=10)
            self.pool[host] = client
            return client

    def close_all(self):
        with self.lock:
            for client in self.pool.values():
                client.close()
            self.pool.clear()

def batch_execute(hosts, command, max_workers=10):
    """并发批量执行命令"""
    pool = SSHConnectionPool()
    results = []

    def exec_on_host(host_info):
        host = host_info['host']
        try:
            client = pool.get_connection(
                host, host_info['port'],
                host_info['username'], host_info['password']
            )
            stdin, stdout, stderr = client.exec_command(command, timeout=30)
            return {
                'host': host,
                'exit_code': stdout.channel.recv_exit_status(),
                'output': stdout.read().decode().strip()
            }
        except Exception as e:
            return {'host': host, 'exit_code': -1, 'output': str(e)}

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(exec_on_host, h): h for h in hosts}
        for future in as_completed(futures):
            results.append(future.result())

    pool.close_all()
    return results

# 批量执行示例
hosts = [
    {'host': '10.0.1.10', 'port': 22, 'username': 'root', 'password': 'pass'},
    {'host': '10.0.1.11', 'port': 22, 'username': 'root', 'password': 'pass'},
    {'host': '10.0.1.12', 'port': 22, 'username': 'root', 'password': 'pass'},
]

results = batch_execute(hosts, 'df -h / && free -m')
for r in results:
    print(f"[{r['host']}] exit={r['exit_code']}")
    print(r['output'])
    print('-' * 60)

SFTP 文件传输

def sftp_upload(host, port, username, password, local_path, remote_path):
    """SFTP 文件上传"""
    transport = paramiko.Transport((host, port))
    try:
        transport.connect(username=username, password=password)
        sftp = paramiko.SFTPClient.from_transport(transport)
        sftp.put(local_path, remote_path)
        # 验证文件大小
        local_size = os.path.getsize(local_path)
        remote_size = sftp.stat(remote_path).st_size
        assert local_size == remote_size, f"Size mismatch: {local_size} vs {remote_size}"
        return True
    finally:
        transport.close()

Fabric：简化远程操作

Fabric 在 paramiko 之上封装了更高层的 API，将常见的 SSH 操作简化为函数调用：

from fabric import Connection, Config

# 批量部署配置
env_config = Config({
    'run': {'warn': True, 'echo': True, 'timeout': 30},
    'sudo': {'password': 'sudo-password'},
})

def deploy_nginx(c):
    """通过 Fabric 部署 Nginx"""
    c.run('yum install -y nginx')
    c.put('nginx.conf', '/etc/nginx/nginx.conf', use_sudo=True)
    c.sudo('systemctl enable nginx')
    c.sudo('systemctl restart nginx')
    c.run('nginx -t')

# 执行部署
for host in ['10.0.1.10', '10.0.1.11']:
    conn = Connection(host=host, user='root', config=env_config)
    deploy_nginx(conn)
    conn.close()

Fabric 的优势在于声明式 API——你描述"做什么"，连接管理、错误处理等细节由框架处理。适合中量级运维任务，但当规模超过百台机器时，维护脚本本身变得复杂。

Ansible：声明式自动化

当运维规模增长到数百上千台时，Ansible 成为更优选择。它不需要 Agent、使用 YAML 声明式描述、内置大量模块。

核心概念

概念说明类比
Inventory主机清单服务器列表
Module执行单元函数
Task调用模块的动作函数调用
Playbook任务集合脚本
Role可复用的任务包函数库
Variables参数化配置变量

Inventory 主机清单

# /etc/ansible/hosts
[webservers]
web-[01:03].sre.wang    # 通配符匹配 web-01, web-02, web-03

[databases]
db-01.sre.wang ansible_host=10.0.2.10
db-02.sre.wang ansible_host=10.0.2.11

[webservers:vars]
ansible_user=deploy
ansible_ssh_private_key_file=~/.ssh/id_ed25519
nginx_version=1.25.3

[databases:vars]
ansible_user=deploy
mysql_version=8.0.35

[production:children]
webservers
databases

Ansible 实战：批量部署 Nginx

---
# playbook: deploy-nginx.yml
- name: 批量部署 Nginx 并配置管理
  hosts: webservers
  become: yes
  vars:
    nginx_version: "1.25.3"
    nginx_worker_processes: "auto"
    nginx_worker_connections: "10240"

  tasks:
    - name: 安装 EPEL 仓库
      yum:
        name: epel-release
        state: present

    - name: 安装 Nginx
      yum:
        name: "nginx-{{ nginx_version }}"
        state: present
      notify: restart nginx

    - name: 创建 Nginx 配置目录
      file:
        path: /etc/nginx/conf.d
        state: directory
        mode: '0755'

    - name: 部署主配置文件
      template:
        src: nginx.conf.j2
        dest: /etc/nginx/nginx.conf
        backup: yes
        validate: nginx -t -c %s
      notify: reload nginx

    - name: 部署站点配置
      template:
        src: site.conf.j2
        dest: "/etc/nginx/conf.d/{{ item.name }}.conf"
      loop:
        - { name: 'app', port: 8080, server_name: 'app.sre.wang' }
        - { name: 'api', port: 8081, server_name: 'api.sre.wang' }
      notify: reload nginx

    - name: 确保 Nginx 已启动并设为开机自启
      systemd:
        name: nginx
        state: started
        enabled: yes

    - name: 开放防火墙端口
      firewalld:
        port: "{{ item }}/tcp"
        permanent: yes
        immediate: yes
        state: enabled
      loop: [80, 443]

  handlers:
    - name: restart nginx
      systemd:
        name: nginx
        state: restarted

    - name: reload nginx
      systemd:
        name: nginx
        state: reloaded

Jinja2 模板文件 nginx.conf.j2：

# {{ ansible_managed }}
worker_processes {{ nginx_worker_processes }};
worker_rlimit_nofile 65535;

events {
    worker_connections {{ nginx_worker_connections }};
    use epoll;
    multi_accept on;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    sendfile        on;
    tcp_nopush      on;
    tcp_nodelay     on;
    keepalive_timeout  65;
    server_tokens   off;

    gzip on;
    gzip_min_length 1k;
    gzip_types text/plain application/json application/javascript text/css;

    include /etc/nginx/conf.d/*.conf;
}

Role 组织结构

当 Playbook 膨大到数百行时，应拆分为 Role：

roles/nginx/
├── defaults/
│   └── main.yml         # 默认变量
├── vars/
│   └── main.yml         # 优先级更高的变量
├── tasks/
│   └── main.yml         # 主任务
├── handlers/
│   └── main.yml         # 触发器
├── templates/
│   ├── nginx.conf.j2
│   └── site.conf.j2
└── meta/
    └── main.yml         # 依赖声明

从 paramiko 迁移到 Ansible 的架构思考

演进路径

单台手动 SSH → paramiko 脚本 → Fabric 封装 → Ansible 声明式
    │              │              │              │
    │              │              │              │
  1-5台          5-50台        50-100台       100+台

三者对比

维度paramikoFabricAnsible
抽象层级协议级函数级声明式
Agent无无无
幂等性需自行实现需自行实现模块内置
模板系统无无Jinja2
Role 复用无弱强
学习曲线低低中
适用规模小中大
可维护性差中高

何时仍然需要 paramiko

Ansible 并非万能，以下场景 paramiko 仍然是更好的选择：

  • 非标准 SSH 场景：需要交互式会话、端口转发、SSH 隧道
  • 复杂条件逻辑：需要根据上一条命令的实时输出动态决策
  • 嵌入式环境：无法安装 Ansible 的轻量级设备
  • 性能敏感：需要细粒度控制超时、重连、连接复用

混合架构实践

生产环境推荐混合使用：

# 用 paramiko 做健康检查和预检
def pre_check(host):
    result = ssh_exec(host, 22, 'deploy', 'pass', 'uname -r')
    kernel = result['output']
    # 根据内核版本选择不同的 Ansible Playbook
    if '4.18' in kernel:
        return 'playbook-rhel8.yml'
    else:
        return 'playbook-rhel7.yml'

# 用 Ansible 做实际部署
import subprocess
for host in hosts:
    playbook = pre_check(host['host'])
    subprocess.run([
        'ansible-playbook', playbook,
        '-l', host['host'],
        '-e', f'target_host={host["host"]}'
    ])

总结

运维自动化不是工具选择问题，而是架构演进问题。paramiko 提供了 SSH 协议级控制力，Fabric 简化了常见操作，Ansible 用声明式模型解决了规模化管理的可维护性。

关键原则：小规模用 paramiko/Fabric 保持灵活，大规模用 Ansible 保证幂等和可维护，混合使用各取所长。无论选择哪个工具，最终目标都是：减少手动操作、提高一致性、让基础设施可审计可重现。

参考资料与致谢

本文在撰写过程中参考了以下资料，感谢原作者的贡献：

  1. paramiko — Docs，参考了paramiko相关内容
  2. Fabric — Docs，参考了Fabric相关内容