feat(core/transport): keep one SSH connection per transport

CHANGELOG.md

@@ -10,6 +10,14 @@ Sections with "(Upcoming)" describe changes on the roadmap for CIJOE.
 Changes on the `master` branch, from the latest version tag up to and including
 HEAD can be subject to a git rebase.
 
+## 0.9.61 (Upcoming)
+
+The SSH transport keeps one connection for its run()/get()/put() instead of
+connecting and disconnecting around each, which cost a key exchange, about 0.4 s,
+per command. The connection has a 30 second keepalive, is made again
+on next use when it has dropped, e.g. as the remote rebooted, and is closed by
+the new `Cijoe.close()`, at the end of a pytest session, or as Python exits.
+
 ## 0.9.59
 
 Renamed the "workflow" concept to "task" across the codebase, docs, and example

docs/source/configs/index.rst

@@ -76,6 +76,11 @@ password login or key-based login. You can configure it as follows:
    key_filename = "path/to/private_key"
    passphrase   = "baz"
 
+The connection is made on first use and kept for the commands and transfers
+that follow, with a keepalive every 30 seconds. When it has dropped, for example
+because the remote rebooted, it is made again on next use. ``cijoe.close()``
+closes it; otherwise it is closed as Python exits.
+
 
 Configuring multiple transports
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

src/cijoe/core/command.py

@@ -263,6 +263,19 @@ def get(self, src, dst, transport_name=None):
 
         return False
 
+    def close(self, transport_name=None):
+        """
+        Close the connections the transports keep, or only the one of the named
+        transport; a later call connects again
+        """
+
+        if transport_name is not None:
+            self._get_transport(transport_name).close()
+            return
+
+        for tport in self.transports.values():
+            tport.close()
+
     def getconf(self, key: str, default: Any = None):
         """
         Return value for given key, and return default if no value is found.

src/cijoe/core/scripts/wait_for_transport.py

@@ -130,7 +130,10 @@ def main(args, cijoe):
             continue
 
         # TCP is open - verify with the trivial command to distinguish usable vs. unusable SSH.
+        # The transport keeps its connection between commands; closing it first makes the probe
+        # a fresh connect, which is what tells a host that went down or came up.
         try:
+            cijoe.close(transport_name)
             err, _ = cijoe.run("hostname", transport_name=transport_name)
             ssh_up = err == 0
         except Exception as exc:

src/cijoe/core/transport.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 import subprocess
+import weakref
 from abc import ABC, abstractmethod
 from pathlib import Path
 
@@ -25,6 +26,9 @@ def get(self, src, dst=None):
     def put(self, src, dst=None):
         pass
 
+    def close(self):
+        """Release what the transport holds open; it may be used again afterwards"""
+
 
 class Local(Transport):
     """Provide cmd/push/pull locally"""
@@ -83,6 +87,10 @@ def get(self, src, dst=None):
         return self.put(src, dst)
 
 
+# Keeps the NAT and firewall state of an idle kept connection alive
+KEEPALIVE_SECONDS = 30
+
+
 class SSH(Transport):
     """Provide cmd/push/pull over SSH"""
 
@@ -109,26 +117,42 @@ def __init__(self, config, output_path, transport_name):
         self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
         self.scp = None
 
-        # Not connecting at this point, since the remote side might be ready at the time
-        # the transport is initialized. For example, when a qemu-guest is booted, then
-        # cijoe.run_local() is used, and not until the guest is up will the cij.run() be
-        # used. Also, when a system reboots etc. then dropped connections must be
-        # handled, lastly connection close must happen as Python terminates, dangling
-        # connections will make it hang.
-        # Thus, __connect()/__disconnect() is called for each call to run()/get()/put().
-        # Current short-coming is of course that then these cannot happen in parallel.
+        # Not connecting here, the remote might not be up yet, e.g. a qemu guest that
+        # run_local() boots. The connection is made on first use and kept, rather than
+        # paying a key exchange per command. One that dropped, e.g. on a reboot of the
+        # remote, is made again on next use. Calls on one transport cannot overlap.
+        weakref.finalize(self, self.ssh.close)
 
         log.getLogger("paramiko.transport").setLevel(log.CRITICAL)
         paramiko.util.log_to_file(
             self.output_path / "paramiko.log", level=log.root.level
         )
 
     def __connect(self):
-        self.ssh.connect(**self.ssh_params)
-        self.scp = SCPClient(self.ssh.get_transport())
+        """Connect, unless the connection made before is still up"""
 
-    def __disconnect(self):
-        self.scp.close()
+        transport = self.ssh.get_transport()
+        if transport is not None and transport.is_active():
+            return
+
+        self.close()
+        try:
+            self.ssh.connect(**self.ssh_params)
+            transport = self.ssh.get_transport()
+            transport.set_keepalive(KEEPALIVE_SECONDS)
+            self.scp = SCPClient(transport)
+        except BaseException:
+            # paramiko starts the transport before it authenticates, so a failed
+            # connect can leave one active, which the next use must not take
+            self.close()
+            raise
+
+    def close(self):
+        """Close the connection; the next run()/get()/put() connects again"""
+
+        if self.scp is not None:
+            self.scp.close()
+            self.scp = None
         self.ssh.close()
 
     def run(self, cmd, cwd, env, cmd_output):
@@ -173,9 +197,9 @@ def run(self, cmd, cwd, env, cmd_output):
             cmd_output.flush()
 
             err = stdout.channel.recv_exit_status()
-
-            self.__disconnect()
+            stdout.channel.close()
         except paramiko.ssh_exception.SSHException as exc:
+            self.close()
             err = (
                 exc.errno
                 if hasattr(exc, "errno")
@@ -201,9 +225,11 @@ def put(self, src, dst=None):
         if not os.path.isabs(src):
             src = os.path.join(self.output_path, self.output_ident, src)
 
-        self.scp.put(src, dst, recursive=True)
-
-        self.__disconnect()
+        try:
+            self.scp.put(src, dst, recursive=True)
+        except Exception:
+            self.close()
+            raise
 
         return True
 
@@ -217,8 +243,10 @@ def get(self, src, dst=None):
         if not os.path.isabs(src):
             dst = os.path.join(self.output_path, self.output_ident, dst)
 
-        self.scp.get(src, dst, recursive=True)
-
-        self.__disconnect()
+        try:
+            self.scp.get(src, dst, recursive=True)
+        except Exception:
+            self.close()
+            raise
 
         return True

src/cijoe/pytest_plugin/hooks_and_fixtures.py

@@ -81,6 +81,13 @@ def pytest_configure(config):
         raise Exception("Failed instantiating cijoe")
 
 
+def pytest_unconfigure(config):
+    """Closes the connections of the cijoe instance made by ``pytest_configure``"""
+
+    if pytest.cijoe_instance is not None:
+        pytest.cijoe_instance.close()
+
+
 def pytest_terminal_summary(terminalreporter, exitstatus, config):
     """
     Prints out a notice that cijoe is beeing used along with the values of the