Merge pull request #382 from xcp-ng/gln/optimize-ssh-tests-with-multiplexing-swro

ssh: use ssh multiplexing to speed up the tests

Author: stormi

lib/commands.py

@@ -1,5 +1,7 @@
 import base64
 import logging
+import os
+import platform
 import shlex
 import subprocess
 
@@ -63,7 +65,7 @@ def _ellide_log_lines(log):
     return "\n{}".format("\n".join(reduced_message))
 
 def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warnings,
-         background, decode, options) -> Union[SSHResult, SSHCommandFailed, str, bytes, None]:
+         background, decode, options, multiplexing) -> Union[SSHResult, SSHCommandFailed, str, bytes, None]:
     opts = list(options)
     opts.append('-o "BatchMode yes"')
     opts.append('-o "PubkeyAcceptedAlgorithms +ssh-rsa"')
@@ -74,6 +76,18 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
         opts.append('-o "StrictHostKeyChecking no"')
         opts.append('-o "LogLevel ERROR"')
         opts.append('-o "UserKnownHostsFile /dev/null"')
+    # ssh multiplexing is not always well supported on windows, so we disable it on that platform.
+    # It could work with git bash—we might want to check that instead.
+    # We use the pid in the control path to avoid a race condition on the master socket creation
+    # when running the tests in parallel. The socket is removed by the ssh client olding the master
+    # connection when it reaches the timeout.
+    if multiplexing and platform.system() != "Windows":
+        opts.append('-o "ControlMaster auto"')
+        opts.append(f'-o "ControlPath ~/.ssh/control-{os.getpid()}:%h:%p:%r"')
+        opts.append('-o "ControlPersist 10m"')
+        opts.append('-o "ServerAliveInterval 10s"')
+    else:
+        opts.append('-o "ControlMaster no"')
 
     if isinstance(cmd, str):
         command = cmd
@@ -143,47 +157,47 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
 def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
         simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-        decode: Literal[True] = True, options: List[str] = []) -> str:
+        decode: Literal[True] = True, options: List[str] = [], multiplexing=True) -> str:
     ...
 @overload
 def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
         simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-        decode: Literal[False], options: List[str] = []) -> bytes:
+        decode: Literal[False], options: List[str] = [], multiplexing=True) -> bytes:
     ...
 @overload
 def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
         simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-        decode: bool = True, options: List[str] = []) -> SSHResult:
+        decode: bool = True, options: List[str] = [], multiplexing=True) -> SSHResult:
     ...
 @overload
 def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
         simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[True],
-        decode: bool = True, options: List[str] = []) -> None:
+        decode: bool = True, options: List[str] = [], multiplexing=True) -> None:
     ...
 @overload
 def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check=True,
         simple_output: bool = True,
         suppress_fingerprint_warnings=True, background: bool = False,
-        decode: bool = True, options: List[str] = []) \
+        decode: bool = True, options: List[str] = [], multiplexing=True) \
         -> Union[str, bytes, SSHResult, None]:
     ...
 def ssh(hostname_or_ip, cmd, *, check=True, simple_output=True,
         suppress_fingerprint_warnings=True,
-        background=False, decode=True, options=[]):
+        background=False, decode=True, options=[], multiplexing=True):
     result_or_exc = _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warnings,
-                         background, decode, options)
+                         background, decode, options, multiplexing)
     if isinstance(result_or_exc, SSHCommandFailed):
         raise result_or_exc
     else:
         return result_or_exc
 
 def ssh_with_result(hostname_or_ip, cmd, suppress_fingerprint_warnings=True,
-                    background=False, decode=True, options=[]) -> SSHResult:
+                    background=False, decode=True, options=[], multiplexing=True) -> SSHResult:
     result_or_exc = _ssh(hostname_or_ip, cmd, False, False, suppress_fingerprint_warnings,
-                         background, decode, options)
+                         background, decode, options, multiplexing)
     if isinstance(result_or_exc, SSHCommandFailed):
         raise result_or_exc
     elif isinstance(result_or_exc, SSHResult):

lib/host.py

@@ -29,7 +29,6 @@
     strtobool,
     to_xapi_bool,
     wait_for,
-    wait_for_not,
 )
 from lib.netutil import wrap_ip
 from lib.pif import PIF
@@ -92,38 +91,39 @@ def __str__(self):
     @overload
     def ssh(self, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
             suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-            decode: Literal[True] = True) -> str:
+            decode: Literal[True] = True, multiplexing=True) -> str:
         ...
 
     @overload
     def ssh(self, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[True] = True,
             suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-            decode: Literal[False]) -> bytes:
+            decode: Literal[False], multiplexing=True) -> bytes:
         ...
 
     @overload
     def ssh(self, cmd: Union[str, List[str]], *, check: bool = True, simple_output: Literal[False],
             suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
-            decode: bool = True) -> commands.SSHResult:
+            decode: bool = True, multiplexing=True) -> commands.SSHResult:
         ...
 
     @overload
     def ssh(self, cmd: Union[str, List[str]], *, check: bool = True, simple_output: bool = True,
             suppress_fingerprint_warnings: bool = True, background: Literal[True],
-            decode: bool = True) -> None:
+            decode: bool = True, multiplexing=True) -> None:
         ...
 
     @overload
     def ssh(self, cmd: Union[str, List[str]], *, check: bool = True, simple_output: bool = True,
-            suppress_fingerprint_warnings: bool = True, background: bool = False, decode: bool = True) \
+            suppress_fingerprint_warnings: bool = True, background: bool = False, decode: bool = True,
+            multiplexing=True) \
             -> Union[str, bytes, commands.SSHResult, None]:
         ...
 
     def ssh(self, cmd, *, check=True, simple_output=True, suppress_fingerprint_warnings=True,
-            background=False, decode=True):
+            background=False, decode=True, multiplexing=True):
         return commands.ssh(self.hostname_or_ip, cmd, check=check, simple_output=simple_output,
                             suppress_fingerprint_warnings=suppress_fingerprint_warnings,
-                            background=background, decode=decode)
+                            background=background, decode=decode, multiplexing=multiplexing)
 
     def ssh_with_result(self, cmd) -> commands.SSHResult:
         # doesn't raise if the command's return is nonzero, unless there's a SSH error
@@ -550,14 +550,11 @@ def yum_restore_saved_state(self):
 
     def reboot(self, verify=False):
         logging.info("Reboot host %s" % self)
-        try:
-            self.ssh(['reboot'])
-        except commands.SSHCommandFailed as e:
-            # ssh connection may get killed by the reboot and terminate with an error code
-            if "closed by remote host" not in e.stdout:
-                raise
+        # Running `reboot` directly immediately disconnects the ssh session and makes the ssh client return with an
+        # error code. Instead, we schedule the reboot a few seconds later to let the ssh command return properly.
+        self.ssh(['systemd-run --on-active=2s reboot'])
         if verify:
-            wait_for_not(self.is_enabled, "Wait for host down")
+            wait_for(lambda: os.system(f"ping -c1 {self.hostname_or_ip} > /dev/null 2>&1"), "Wait for host down")
             wait_for(lambda: not os.system(f"ping -c1 {self.hostname_or_ip} > /dev/null 2>&1"),
                      "Wait for host up", timeout_secs=10 * 60, retry_delay_secs=10)
             wait_for(lambda: not os.system(f"nc -zw5 {self.hostname_or_ip} 22"),

tests/storage/fsp/conftest.py

@@ -1,6 +1,9 @@
 import pytest
 
+import time
+
 # Explicitly import package-scoped fixtures (see explanation in pkgfixtures.py)
+from lib.host import Host
 from pkgfixtures import host_with_saved_yum_state_toolstack_restart
 
 FSP_REPO_NAME = 'runx'
@@ -18,9 +21,12 @@ def host_with_runx_repo(host_with_saved_yum_state_toolstack_restart):
     host.remove_xcpng_repo(FSP_REPO_NAME)
 
 @pytest.fixture(scope='package')
-def host_with_fsp(host_with_runx_repo):
+def host_with_fsp(host_with_runx_repo: Host):
     host = host_with_runx_repo
     host.yum_install(FSP_PACKAGES)
+    # fsp is not listed by xe sm-list until it's actually used, so just wait a few seconds instead
+    # wait_for(lambda: host.xe('sm-list', {'type': 'fsp'}).strip() != '', "Wait for fsp to be available")
+    time.sleep(3)
     yield host
     # teardown: nothing to do, done by host_with_saved_yum_state.