python paramiko 模块中长时间运行的 ssh 命令（以及如何结束它们）

Question

我想使用 python 的 paramiko 模块在远程计算机上运行 tail -f logfile 命令。 到目前为止，我一直在以以下方式尝试：

interface = paramiko.SSHClient()
#snip the connection setup portion
stdin, stdout, stderr = interface.exec_command("tail -f logfile")
#snip into threaded loop
print stdout.readline()

我希望命令在必要时运行，但我有两个问题：

1. 如何干净地停止此操作？ 我想过创建一个通道，然后在完成后在通道上使用 shutdown() 命令 - 但这看起来很混乱。 是否可以执行类似将 Ctrl-C 发送到通道的标准输入之类的操作？
2. readline() 会阻塞，如果我有一个非阻塞的获取输出的方法，我可以避免线程 - 有什么想法吗？

Answer 1

不要在客户端调用 exec_command，而是掌握传输并生成您自己的通道。 频道 可用于执行命令，并且您可以在 select 语句中使用它来确定何时可以读取数据：

#!/usr/bin/env python
import paramiko
import select
client = paramiko.SSHClient()
client.load_system_host_keys()
client.connect('host.example.com')
transport = client.get_transport()
channel = transport.open_session()
channel.exec_command("tail -f /var/log/everything/current")
while True:
  rl, wl, xl = select.select([channel],[],[],0.0)
  if len(rl) > 0:
      # Must be stdout
      print channel.recv(1024)

可以读取和写入通道对象，与远程命令的 stdout 和 stdin 连接。 您可以通过调用channel.makefile_stderr(...)来获取stderr。

我已将超时设置为 0.0 秒，因为请求了非阻塞解决方案。 根据您的需要，您可能希望使用非零超时进行阻止。

Answer 2

1）如果您愿意，您可以关闭客户端。 另一端的服务器将杀死尾部进程。

2）如果您需要以非阻塞方式执行此操作，则必须直接使用通道对象。 然后，您可以使用channel.recv_ready()和channel.recv_stderr_ready()来监视stdout和stderr，或者使用select.select。

Answer 3

只是对 Andrew Aylett 的解决方案进行了一个小更新。 以下代码实际上打破了循环并在外部进程完成时退出：

import paramiko
import select

client = paramiko.SSHClient()
client.load_system_host_keys()
client.connect('host.example.com')
channel = client.get_transport().open_session()
channel.exec_command("tail -f /var/log/everything/current")
while True:
    if channel.exit_status_ready():
        break
    rl, wl, xl = select.select([channel], [], [], 0.0)
    if len(rl) > 0:
        print channel.recv(1024)

Answer 4

要关闭该进程，只需运行：

interface.close()

就非阻塞而言，您无法获得非阻塞读取。 最好的方法是一次解析一个“块”，“stdout.read(1)”仅在缓冲区中没有剩余字符时才会阻塞。

Answer 5

仅供参考，有一个解决方案可以使用channel.get_pty() 来执行此操作。 有关更多详细信息，请查看：https://stackoverflow.com/a/11190727/1480181

Answer 6

我解决这个问题的方法是使用上下文管理器。 这将确保我的长时间运行的命令被中止。 关键逻辑是包装以模仿 SSHClient.exec_command，但捕获创建的通道并使用计时器，如果命令运行时间过长，该计时器将关闭该通道。

import paramiko
import threading


class TimeoutChannel:

    def __init__(self, client: paramiko.SSHClient, timeout):
        self.expired = False
        self._channel: paramiko.channel = None
        self.client = client
        self.timeout = timeout

    def __enter__(self):
        self.timer = threading.Timer(self.timeout, self.kill_client)
        self.timer.start()

        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Exited Timeout. Timed out:", self.expired)
        self.timer.cancel()

        if exc_val:
            return False  # Make sure the exceptions are re-raised

        if self.expired:
            raise TimeoutError("Command timed out")

    def kill_client(self):
        self.expired = True
        print("Should kill client")
        if self._channel:
            print("We have a channel")
            self._channel.close()

    def exec(self, command, bufsize=-1, timeout=None, get_pty=False, environment=None):
        self._channel = self.client.get_transport().open_session(timeout=timeout)
        if get_pty:
            self._channel.get_pty()
        self._channel.settimeout(timeout)
        if environment:
            self._channel.update_environment(environment)
        self._channel.exec_command(command)
        stdin = self._channel.makefile_stdin("wb", bufsize)
        stdout = self._channel.makefile("r", bufsize)
        stderr = self._channel.makefile_stderr("r", bufsize)
        return stdin, stdout, stderr

现在使用代码非常简单，第一个示例将抛出 TimeoutError

ssh = paramiko.SSHClient()
ssh.connect('hostname', username='user', password='pass')

with TimeoutChannel(ssh, 3) as c:
    ssh_stdin, ssh_stdout, ssh_stderr = c.exec("cat")    # non-blocking
    exit_status = ssh_stdout.channel.recv_exit_status()  # block til done, will never complete because cat wants input

此代码将正常工作（除非主机处于疯狂的负载之下！）

ssh = paramiko.SSHClient()
ssh.connect('hostname', username='user', password='pass')

with TimeoutChannel(ssh, 3) as c:
    ssh_stdin, ssh_stdout, ssh_stderr = c.exec("uptime")    # non-blocking
    exit_status = ssh_stdout.channel.recv_exit_status()     # block til done, will complete quickly
    print(ssh_stdout.read().decode("utf8"))                 # Show results