使用 Python Paramiko 在不同的 SSH 服务器中并行运行多个命令

Posted 2023-03-23

【中文标题】使用 Python Paramiko 在不同的 SSH 服务器中并行运行多个命令

【英文标题】：Run multiple commands in different SSH servers in parallel using Python Paramiko

【发布时间】：2021-05-17 08:20:25

【问题描述】：

我有一个SSH.py，目标是通过 SSH 连接到许多服务器以运行 Python 脚本 (worker.py)。我正在使用 Paramiko，但对它很陌生，并且在学习过程中不断学习。在我使用 ssh 连接的每台服务器上，我需要保持 Python 脚本运行——这是为了并行训练模型，因此脚本需要在所有机器上运行以共同更新模型参数/训练。服务器上的 Python 脚本需要运行，所以要么所有 SSH 连接都无法关闭，要么我必须想办法让服务器上的 Python 脚本在我关闭连接的情况下继续运行。

从广泛的谷歌搜索来看，您似乎可以使用 nohup 或：

client = paramiko.SSHClient()
client.connect(ip_address, username, password)
transport = client.get_transport()
channel = transport.open_session()
channel.exec_command("python worker.py > /logs/'command output' 2>&1")

但是，我不清楚我们如何关闭/退出所有 SSH 连接？我在cmd.exe 上运行SSH.py 文件，关闭cmd.exe 是否足以让所有进程远程关闭？

此外，我对client.close() 的使用是否符合我的目的？ 请在下面查看我的代码。

# SSH.py

import paramiko
import argparse
import os

path = "path"
python_script = "worker.py"

# definitions for ssh connection and cluster
ip_list = ['XXX.XXX.XXX.XXX', XXX.XXX.XXX.XXX', XXX.XXX.XXX.XXX']
port_list = [':XXXX', ':XXXX', ':XXXX']
user_list = ['user', 'user', 'user']
password_list = ['pass', 'pass', 'pass']
node_list = list(map(lambda x: f'-nodex + 1 ', list(range(len(ip_list)))))
cluster = ' '.join([node + ip + port for node, ip, port in zip(node_list, ip_list, port_list)])

# run script on command line of local machine
os.system(f"cd path && python python_script cluster -type worker -index 0 -batch 64 > path/logs/'command output'/ip_list[0].log 2>&1")

# loop for IP and password
for i, (ip, user, password) in enumerate(zip(ip_list[1:], user_list[1:], password_list[1:]), 1):
    try:
        print("Open session in: " + ip + "...")
        client = paramiko.SSHClient()
        client.connect(ip, user, password)
        transport = client.get_transport()
        channel = transport.open_session()
    except paramiko.SSHException:
        print("Connection Failed")
        quit()

    try:
        channel.exec_command(f"cd path && python python_script cluster -type worker -index i -batch 64 > path/logs/'command output'/ip_list[i].log 2>&1", timeout=30)
        client.close() # here I am closing connection but above command should be running, my question is can I safely close cmd.exe on which I am running SSH.py? 
    except paramiko.SSHException:
        print("Cannot run file. Continue with other IPs in list...")
        client.close()
        continue

代码基于Running process of remote SSH server in the background using Python Paramiko

编辑：好像 channel.exec_command() 没有执行命令

f"cd path && python python_script cluster -type worker -index i -batch 64 > path/logs/'command output'/ip_list[i].log 2>&1"

所以不知道是不是因为client.close()？如果我用client.close() 注释掉所有行会怎样？这会有帮助吗？这很危险吗？当我退出本地 Python 脚本时，这会关闭我所有的 SSH 连接，因此不需要client.close()？

而且我所有的机器都有 Windows 操作系统。

【参考方案1】：

确实，问题在于您关闭了 SSH 连接。由于远程进程未与终端分离，因此关闭终端会终止该进程。在 Linux 服务器上，您可以使用 nohup。我不知道什么是（如果有的话）Windows 等价物。

无论如何，看来您不需要关闭连接。我了解，您可以等待所有命令完成。

stdouts = []
clients = []

# Start the commands
commands = zip(ip_list[1:], user_list[1:], password_list[1:])
for i, (ip, user, password) in enumerate(commands, 1):
    print("Open session in: " + ip + "...")
    client = paramiko.SSHClient()
    client.connect(ip, user, password)
    command = \
        f"cd path && " + \
        f"python python_script cluster -type worker -index i -batch 64 " + \
        f"> path/logs/'command output'/ip_list[i].log 2>&1"
    stdin, stdout, stderr = client.exec_command(command)
    clients.append(client)
    stdouts.append(stdout)

# Wait for commands to complete
for i in range(len(stdouts)):
    stdouts[i].read()
    clients[i].close()

---

请注意，上述stdout.read() 的简单解决方案之所以有效，只是因为您将命令输出重定向到远程文件。你不是commands might deadlock。

没有它（或者如果您想在本地查看命令输出），您将需要这样的代码：

while any(x is not None for x in stdouts):
    for i in range(len(stdouts)):
        stdout = stdouts[i]
        if stdout is not None:
            channel = stdout.channel
            # To prevent losing output at the end, first test for exit,
            # then for output
            exited = channel.exit_status_ready()
            while channel.recv_ready():
                s = channel.recv(1024).decode('utf8')
                print(f"#i stdout: s")
            while channel.recv_stderr_ready():
                s = channel.recv_stderr(1024).decode('utf8')
                print(f"#i stderr: s")
            if exited:
                print(f"#i done")
                clients[i].close()
                stdouts[i] = None
    time.sleep(0.1)

如果不需要分离stdout和stderr，使用Channel.set_combine_stderr可以大大简化代码。见Paramiko ssh die/hang with big output。

---

关于你关于SSHClient.close的问题：如果你不调用它，当脚本完成时，当Python垃圾收集器清理挂起的对象时，连接将被隐式关闭。这是一个不好的做法。而且即使 Python 不这样做，本地操作系统也会终止本地 Python 进程的所有连接。这也是一种不好的做法。无论如何，这将终止远程进程。