diff --git a/computes/models.py b/computes/models.py index 3f172540..a868ba85 100644 --- a/computes/models.py +++ b/computes/models.py @@ -17,12 +17,14 @@ class Compute(Model): @cached_property def status(self): - # return connection_manager.host_is_up(self.type, self.hostname) - # TODO: looks like socket has problems connecting via VPN - if isinstance(self.connection, virConnect): - return True - else: - return self.connection + """True if WebVirtCloud can connect to the host (within the connect + timeout; a host that failed recently is not tried again at once).""" + return isinstance(self.connection, virConnect) + + @cached_property + def connection_error(self): + """Why the host cannot be reached, or None.""" + return None if self.status else str(self.connection) @cached_property def connection(self): diff --git a/computes/test_ownership.py b/computes/test_ownership.py index 1cbe07f6..390b4bdd 100644 --- a/computes/test_ownership.py +++ b/computes/test_ownership.py @@ -34,6 +34,7 @@ class OwnershipTestCase(TestCase): def compute(self, name): compute = Compute.objects.create(name=name, hostname=f"{name}.example", login="", password="", type=1) compute.__dict__["proxy"] = MagicMock() + compute.__dict__["status"] = True # reachable self.on(compute, []) return compute diff --git a/computes/test_reconciliation.py b/computes/test_reconciliation.py index 4d234d51..082b575a 100644 --- a/computes/test_reconciliation.py +++ b/computes/test_reconciliation.py @@ -40,6 +40,7 @@ class InstanceReconciliationTestCase(TestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [dom1, dom2] self.compute.__dict__["proxy"] = mock_proxy + self.compute.__dict__["status"] = True # reachable refresh_instance_database(self.compute) @@ -61,6 +62,7 @@ class InstanceReconciliationTestCase(TestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [] self.compute.__dict__["proxy"] = mock_proxy + self.compute.__dict__["status"] = True # reachable refresh_instance_database(self.compute) @@ -82,6 +84,7 @@ class InstanceReconciliationTestCase(TestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [dom_renamed] self.compute.__dict__["proxy"] = mock_proxy + self.compute.__dict__["status"] = True # reachable refresh_instance_database(self.compute) @@ -105,6 +108,7 @@ class InstanceReconciliationTestCase(TestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.side_effect = Exception("Libvirt socket connection refused") self.compute.__dict__["proxy"] = mock_proxy + self.compute.__dict__["status"] = True # reachable refresh_instance_database(self.compute) @@ -124,6 +128,7 @@ class InstanceReconciliationTestCase(TestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [broken_dom] self.compute.__dict__["proxy"] = mock_proxy + self.compute.__dict__["status"] = True # reachable refresh_instance_database(self.compute) diff --git a/computes/tests.py b/computes/tests.py index bf034203..856147f2 100644 --- a/computes/tests.py +++ b/computes/tests.py @@ -214,6 +214,7 @@ class ComputeConcurrencyTestCase(TransactionTestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [mock_dom1, mock_dom2] compute.proxy = mock_proxy + compute.status = True # reachable exceptions = [] @@ -253,6 +254,7 @@ class ComputeConcurrencyTestCase(TransactionTestCase): mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [] compute.proxy = mock_proxy + compute.status = True # reachable # 1. When lock directory cannot be acquired with patch("computes.utils._get_lock_directory", return_value=None): diff --git a/instances/templates/allinstances.html b/instances/templates/allinstances.html index 9f10e1f8..23cb6fd8 100644 --- a/instances/templates/allinstances.html +++ b/instances/templates/allinstances.html @@ -18,7 +18,7 @@ {% for compute in computes %} {% if compute.status is not True %} {% endif %} diff --git a/vrtManager/connection.py b/vrtManager/connection.py index bb69fbf6..c4933a88 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -1,6 +1,9 @@ +import os import re import socket import threading +import time +from urllib.parse import quote import libvirt from django.conf import settings @@ -16,6 +19,20 @@ TLS_PORT = 16514 SSH_PORT = 22 TCP_PORT = 16509 +# libvirt has no timeout for opening a connection: an unreachable host +# blocks for the kernel's TCP connect timeout (about 2 minutes), a daemon +# that accepts but never answers blocks forever, and such an open cannot be +# cancelled. Connections are opened in a helper thread, at most one per +# connection at a time, so a stuck host holds one thread and nothing else; +# the caller waits at most CONNECT_TIMEOUT seconds, and a host that failed is +# not tried again for RETRY_AFTER seconds. +CONNECT_TIMEOUT = getattr(settings, "LIBVIRT_CONNECT_TIMEOUT", 5) +RETRY_AFTER = getattr(settings, "LIBVIRT_RETRY_AFTER", 30) +CLOSE_REASONS = {0: "error", 1: "end of file", 2: "keepalive timeout", 3: "client closed"} +# ssh with a connect timeout for qemu+ssh (libvirt runs the "command" binary +# in place of ssh and never passes one itself) +SSH_COMMAND = os.path.join(os.path.dirname(os.path.abspath(__file__)), "libvirt-ssh") + class wvmEventLoop(threading.Thread): """ Event Loop Class""" @@ -58,6 +75,8 @@ class wvmConnection(object): self.connection_state_lock = threading.Lock() self.connection = None self.last_error = None + self._opening = None # threading.Event of the open in progress + self._retry_at = 0.0 # monotonic time before which no new open is tried # credentials self.host = host @@ -66,39 +85,70 @@ class wvmConnection(object): self.type = conn def connect(self): - self.connection_state_lock.acquire() - try: - # recheck if we have a connection (it may have been - if not self.connected: - if self.type == CONN_TCP: - self.__connect_tcp() - elif self.type == CONN_SSH: - self.__connect_ssh() - elif self.type == CONN_TLS: - self.__connect_tls() - elif self.type == CONN_SOCKET: - self.__connect_socket() - else: - raise ValueError(f'"{self.type}" is not a valid connection type') + """Open the connection if needed, waiting at most CONNECT_TIMEOUT. + On failure last_error says why and connected stays False.""" + with self.connection_state_lock: + if self.connected: + return + if self._opening is not None and self._opening.abandoned: + # A caller already waited for this open in vain; nobody waits + # again until it ends (one thread per connection, however long). + return + if time.monotonic() < self._retry_at: + return # failed recently: answer with the last error at once + if self._opening is None: + self._opening = threading.Event() + self._opening.abandoned = False + threading.Thread(target=self._open, args=(self._opening,), daemon=True).start() + opening = self._opening - if self.connected: - # do some preprocessing of the connection: - # * set keep alive interval - # * set connection close/fail handler - try: - self.connection.setKeepAlive( - connection_manager.keepalive_interval, connection_manager.keepalive_count - ) - try: - self.connection.registerCloseCallback(self.__connection_close_callback, None) - except Exception: - # Temporary fix for libvirt > libvirt-0.10.2-41 - pass - except libvirtError as e: - # hypervisor driver does not seem to support persistent connections - self.last_error = str(e) + if not opening.wait(CONNECT_TIMEOUT): + with self.connection_state_lock: + if self._opening is opening: # still running: give up waiting, not the open + self.last_error = f"Connection Failed: {self.host} did not answer within {CONNECT_TIMEOUT} s" + opening.abandoned = True + + def _open(self, opening): + """Helper thread: open, configure and publish the connection.""" + connection = error = None + try: + connection = self._open_transport() + if not connection.isAlive(): # closed again before it was published + raise util.OperationError("the connection closed right after opening") + # Drivers without keepalive or close callbacks work without them + try: + connection.setKeepAlive(connection_manager.keepalive_interval, connection_manager.keepalive_count) + except libvirtError: + pass + try: + connection.registerCloseCallback(self.__connection_close_callback, None) + except libvirtError: + pass + except Exception as e: # libvirtError, or anything else from the transport + error = f"Connection Failed: {str(e)}" finally: - self.connection_state_lock.release() + with self.connection_state_lock: + if error is None: + self.connection = connection + self.last_error = None + self._retry_at = 0.0 + else: + self.connection = None + self.last_error = error + self._retry_at = time.monotonic() + RETRY_AFTER + self._opening = None + opening.set() + + def _open_transport(self): + if self.type == CONN_TCP: + return self.__connect_tcp() + if self.type == CONN_SSH: + return self.__connect_ssh() + if self.type == CONN_TLS: + return self.__connect_tls() + if self.type == CONN_SOCKET: + return self.__connect_socket() + raise ValueError(f'"{self.type}" is not a valid connection type') @property def connected(self): @@ -121,65 +171,30 @@ class wvmConnection(object): return 0 def __connection_close_callback(self, connection, reason, opaque=None): - self.connection_state_lock.acquire() - try: - # on server shutdown libvirt module gets freed before the close callbacks are called - # so we just check here if it is still present - if libvirt is not None: - self.last_error = reason - + with self.connection_state_lock: + # A late callback of a connection that was already replaced must + # not drop its successor. + if connection is not self.connection: + return + self.last_error = f"Connection closed ({CLOSE_REASONS.get(reason, reason)})" # prevent other threads from using the connection (in the future) self.connection = None - finally: - self.connection_state_lock.release() def __connect_tcp(self): flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE] auth = [flags, self.__libvirt_auth_credentials_callback, None] - uri = f"qemu+tcp://{self.host}/system" - - try: - self.connection = libvirt.openAuth(uri, auth, 0) - self.last_error = None - - except libvirtError as e: - self.last_error = f"Connection Failed: {str(e)}" - self.connection = None + return libvirt.openAuth(f"qemu+tcp://{self.host}/system", auth, 0) def __connect_ssh(self): - uri = "qemu+ssh://%s@%s/system" % (self.login, self.host) - - try: - self.connection = libvirt.open(uri) - self.last_error = None - - except libvirtError as e: - self.last_error = f"Connection Failed: {str(e)} --- " + repr(libvirt.virGetLastError()) - self.connection = None + return libvirt.open(f"qemu+ssh://{self.login}@{self.host}/system?command={quote(SSH_COMMAND)}") def __connect_tls(self): flags = [libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE] auth = [flags, self.__libvirt_auth_credentials_callback, None] - uri = "qemu+tls://%s@%s/system" % (self.login, self.host) - - try: - self.connection = libvirt.openAuth(uri, auth, 0) - self.last_error = None - - except libvirtError as e: - self.last_error = f"Connection Failed: {str(e)}" - self.connection = None + return libvirt.openAuth(f"qemu+tls://{self.login}@{self.host}/system", auth, 0) def __connect_socket(self): - uri = "qemu:///system" - - try: - self.connection = libvirt.open(uri) - self.last_error = None - - except libvirtError as e: - self.last_error = f"Connection Failed: {str(e)}" - self.connection = None + return libvirt.open("qemu:///system") def close(self): """ @@ -286,7 +301,7 @@ class wvmConnectionManager(object): self._connections_lock.release() if not connection.connected: - # (re-)connect; the connection's own lock serializes concurrent attempts + # (re-)connect within CONNECT_TIMEOUT; one open per connection at a time connection.connect() if connection.connected: diff --git a/vrtManager/libvirt-ssh b/vrtManager/libvirt-ssh new file mode 100755 index 00000000..16c48715 --- /dev/null +++ b/vrtManager/libvirt-ssh @@ -0,0 +1,12 @@ +#!/bin/sh +# ssh for libvirt qemu+ssh connections (the URI's "command" parameter). +# libvirt passes no connect timeout, so an unreachable host would keep this +# ssh process waiting for the kernel's TCP timeout. BatchMode makes ssh fail +# instead of waiting for a password or host key prompt nobody can answer. +# The timeout is fixed: libvirt passes this script no environment or extra +# arguments. It bounds only ssh's TCP connect; WebVirtCloud stops waiting +# after LIBVIRT_CONNECT_TIMEOUT on its own, so an abandoned ssh lives at most +# this long. +# Everything else comes from the user's ssh configuration (aliases, +# ProxyJump, keys). +exec ssh -o ConnectTimeout=10 -o BatchMode=yes "$@" diff --git a/vrtManager/test_connection.py b/vrtManager/test_connection.py index 8dc82ae8..dcdafeb7 100644 --- a/vrtManager/test_connection.py +++ b/vrtManager/test_connection.py @@ -1,17 +1,19 @@ """Reachability probe and connections that do not block each other.""" +import os import socket import threading import time import unittest -from unittest.mock import patch +from unittest.mock import MagicMock, patch from django.conf import settings if not settings.configured: settings.configure(MAC_OUI="52:54:10") -from vrtManager.connection import CONN_TCP, wvmConnectionManager +import libvirt +from vrtManager.connection import CONN_SSH, CONN_TCP, SSH_COMMAND, wvmConnectionManager def closed_port(): @@ -61,5 +63,148 @@ class ConnectionManagerTestCase(unittest.TestCase): slow.join() + +def alive_connection(): + conn = MagicMock() + conn.isAlive.return_value = 1 + return conn + + +class BoundedOpenTestCase(unittest.TestCase): + """Opening waits at most CONNECT_TIMEOUT; a failed host is not retried + for RETRY_AFTER; one open per connection at a time.""" + + def setUp(self): + for name, value in (("wvmEventLoop", MagicMock()), ("CONNECT_TIMEOUT", 0.3), ("RETRY_AFTER", 1.0)): + patcher = patch(f"vrtManager.connection.{name}", value) + patcher.start() + self.addCleanup(patcher.stop) + self.manager = wvmConnectionManager() + self.release = threading.Event() + self.addCleanup(self.release.set) + self.calls = 0 + + def open_auth(self, result): + def open_auth(uri, auth, flags): + self.calls += 1 + self.release.wait(5) + if isinstance(result, Exception): + raise result + return result + return patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth) + + def get(self, host="h"): + started = time.monotonic() + try: + return self.manager.get_connection(host, "u", "p", CONN_TCP), time.monotonic() - started + except libvirt.libvirtError as err: + return err, time.monotonic() - started + + def test_a_hanging_open_is_given_up_after_the_timeout(self): + with self.open_auth(alive_connection()): + result, took = self.get() + self.assertIsInstance(result, libvirt.libvirtError) + self.assertIn("did not answer within", str(result)) + self.assertLess(took, 1.0) + + def test_within_retry_after_the_last_error_comes_at_once(self): + with self.open_auth(alive_connection()): + self.get() + result, took = self.get() + self.assertIn("did not answer within", str(result)) + self.assertLess(took, 0.1) + self.assertEqual(self.calls, 1) + + def test_a_late_success_is_used(self): + conn = alive_connection() + with self.open_auth(conn): + self.get() + self.release.set() + time.sleep(0.1) + result, took = self.get() + self.assertIs(result, conn) + self.assertEqual(self.calls, 1) + + def test_a_failed_host_is_tried_again_after_retry_after(self): + self.release.set() + with self.open_auth(libvirt.libvirtError("refused")): + self.get() + self.get() + self.assertEqual(self.calls, 1) + time.sleep(1.1) + self.get() + self.assertEqual(self.calls, 2) + + def test_concurrent_requests_share_one_open(self): + with self.open_auth(alive_connection()): + threads = [threading.Thread(target=self.get) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(self.calls, 1) + + def test_a_stuck_host_does_not_keep_other_hosts_from_connecting(self): + healthy = alive_connection() + + def open_auth(uri, auth, flags): + if "stuck" in uri: + self.release.wait(5) # an open that never returns + return healthy + + with patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth): + for i in range(12): + self.get(f"stuck{i}") + result, took = self.get("healthy") + self.assertIs(result, healthy) + + def test_an_abandoned_open_is_not_waited_for_again_nor_duplicated(self): + conn = alive_connection() + with self.open_auth(conn): + self.get() + time.sleep(1.1) # past RETRY_AFTER, the open still runs + result, took = self.get() + self.assertIsInstance(result, libvirt.libvirtError) + self.assertLess(took, 0.1) + self.release.set() # the host answers at last + time.sleep(0.1) + result, _ = self.get() + self.assertIs(result, conn) + self.assertEqual(self.calls, 1) + + def test_a_late_close_callback_does_not_drop_the_new_connection(self): + self.release.set() + old, new = alive_connection(), alive_connection() + with self.open_auth(old): + self.get() + entry = self.manager._search_connection("h", "u", "p", CONN_TCP) + old.isAlive.return_value = 0 # seen dead, reconnected + with self.open_auth(new): + self.assertIs(self.get()[0], new) + entry._wvmConnection__connection_close_callback(old, 1) + self.assertIs(self.get()[0], new) + entry._wvmConnection__connection_close_callback(new, 2) + self.assertIsNone(entry.connection) + self.assertEqual(entry.last_error, "Connection closed (keepalive timeout)") + + def test_a_connection_closed_before_it_is_published_is_an_error(self): + self.release.set() + dead = alive_connection() + dead.isAlive.return_value = 0 + with self.open_auth(dead): + result, _ = self.get() + self.assertIn("closed right after opening", str(result)) + + def test_ssh_runs_ssh_with_a_connect_timeout(self): + self.release.set() + with patch("vrtManager.connection.libvirt.open", return_value=alive_connection()) as libvirt_open: + self.manager.get_connection("host", "root", "", CONN_SSH) + uri = libvirt_open.call_args[0][0] + self.assertTrue(uri.startswith("qemu+ssh://root@host/system?command=")) + self.assertTrue(os.access(SSH_COMMAND, os.X_OK)) + with open(SSH_COMMAND) as script: + self.assertIn("ConnectTimeout", script.read()) + + if __name__ == "__main__": unittest.main() diff --git a/vrtManager/test_live_connection.py b/vrtManager/test_live_connection.py new file mode 100644 index 00000000..9e0974af --- /dev/null +++ b/vrtManager/test_live_connection.py @@ -0,0 +1,249 @@ +"""Live checks of bounded connection opens against real libvirt. + +Runs only with TEST_LIBVIRT_HOST set (a TCP compute, see instances/livetest.py); +TEST_LIBVIRT_SSH_HOST/_LOGIN add a reachable ssh compute. 192.0.2.1 is +TEST-NET-1: SYN packets to it are dropped, as for a host that went away +without a trace. + +TEST_LIBVIRT_SSH_ALIASES=1 adds the ssh tests that need these entries in the +ssh configuration of the user running the tests (SSH_HOST is the ssh compute): + + Host wvc-alias + HostName SSH_HOST + Host wvc-relay + HostName 127.0.0.1 + Port 2222 + HostKeyAlias SSH_HOST + Host wvc-silent + HostName 127.0.0.1 + Port 2223 + +TEST_LIBVIRT_SSH_JUMP=1 adds a ProxyJump test through + Host wvc-jump + HostName SSH_HOST + ProxyJump user@another-host +""" + +import os +import shutil +import socket +import subprocess +import threading +import time +import unittest + +import libvirt +from vrtManager.connection import CONN_SSH, CONN_TCP, CONNECT_TIMEOUT, connection_manager + +UNREACHABLE = "192.0.2.1" +KEEPALIVE_DEATH = 45 # keepalive (5, 5) closes a silent connection after ~30 s + + +def timed(*args): + started = time.monotonic() + try: + return connection_manager.get_connection(*args), time.monotonic() - started + except libvirt.libvirtError as err: + return err, time.monotonic() - started + + +def ssh_processes(host): + """ssh client processes (exact name) whose arguments mention host.""" + out = subprocess.run(["pgrep", "-a", "-x", "ssh"], capture_output=True, text=True).stdout + return [line for line in out.splitlines() if host in line] + + +class Relay: + """TCP relay that can hold all traffic (frozen) and pass it again (thawed); + a frozen relay looks like a host that went silent.""" + + def __init__(self, target, port=0, frozen=False): + self.forwarding = threading.Event() + if not frozen: + self.forwarding.set() + self.server = socket.socket() + self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server.bind(("127.0.0.1", port)) + self.server.listen(5) + self.port = self.server.getsockname()[1] + self.target = target + self.closed = False + self.sockets = [] + threading.Thread(target=self.accept, daemon=True).start() + + def accept(self): + while True: + try: + client, _ = self.server.accept() + except OSError: + return + upstream = socket.create_connection(self.target) + self.sockets += [client, upstream] + for a, b in ((client, upstream), (upstream, client)): + threading.Thread(target=self.pump, args=(a, b), daemon=True).start() + + def pump(self, src, dst): + try: + while not self.closed: + if not self.forwarding.wait(0.2): + continue + src.settimeout(0.2) + try: + data = src.recv(65536) + except socket.timeout: + continue + if not data: + break + dst.sendall(data) + except OSError: + pass + + def freeze(self): + self.forwarding.clear() + time.sleep(0.5) # let a recv that was already waiting run out + + def thaw(self): + self.forwarding.set() + + def close(self): + """Stop, and end the relayed connections, as a host that went away would.""" + self.closed = True + self.forwarding.set() + try: + self.server.shutdown(socket.SHUT_RDWR) # wakes the accept; close alone does not + except OSError: + pass + self.server.close() + for sock in self.sockets: + sock.close() + + +@unittest.skipUnless(os.environ.get("TEST_LIBVIRT_HOST"), "Set TEST_LIBVIRT_HOST to run the live libvirt tests") +class LiveBoundedOpenTestCase(unittest.TestCase): + def setUp(self): + self.host = os.environ["TEST_LIBVIRT_HOST"] + self.login = os.environ.get("TEST_LIBVIRT_LOGIN", "") + self.password = os.environ.get("TEST_LIBVIRT_PASSWORD", "") + self.ssh_host = os.environ.get("TEST_LIBVIRT_SSH_HOST") + self.ssh_login = os.environ.get("TEST_LIBVIRT_SSH_LOGIN", "root") + + # Opening + + def test_unreachable_tcp_host_fails_within_the_timeout_then_at_once(self): + args = (UNREACHABLE, "u", "p", CONN_TCP) + first, took = timed(*args) + self.assertIsInstance(first, libvirt.libvirtError) + self.assertLess(took, CONNECT_TIMEOUT + 1) + second, took = timed(*args) + self.assertIsInstance(second, libvirt.libvirtError) + self.assertLess(took, 0.5) + + @unittest.skipUnless(shutil.which("ssh"), "no ssh client") + def test_unreachable_ssh_host_fails_within_the_timeout_and_leaves_no_ssh(self): + result, took = timed(UNREACHABLE, "root", "", CONN_SSH) + self.assertIsInstance(result, libvirt.libvirtError) + self.assertLess(took, CONNECT_TIMEOUT + 1) + time.sleep(12) # the wrapper's ConnectTimeout is 10 s + self.assertEqual(ssh_processes(UNREACHABLE), [], "an ssh process outlived its connect timeout") + + def test_a_daemon_that_never_answers_fails_within_the_timeout(self): + silent = socket.socket() + silent.bind(("127.0.0.1", 0)) + silent.listen(5) # accepts (kernel backlog), never answers + self.addCleanup(silent.close) + result, took = timed(f"127.0.0.1:{silent.getsockname()[1]}", "u", "p", CONN_TCP) + self.assertIsInstance(result, libvirt.libvirtError) + self.assertLess(took, CONNECT_TIMEOUT + 1) + + def check_slow_open_completes_later(self, args, relay): + first, _ = timed(*args) + self.assertIsInstance(first, libvirt.libvirtError) + relay.thaw() + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + conn, _ = timed(*args) + if not isinstance(conn, libvirt.libvirtError): + break + time.sleep(0.5) + self.assertNotIsInstance(conn, libvirt.libvirtError) + self.assertTrue(conn.getVersion()) + + def test_a_slow_tcp_open_that_completes_later_is_used(self): + relay = Relay((self.host, 16509), frozen=True) + self.addCleanup(relay.close) + self.check_slow_open_completes_later((f"127.0.0.1:{relay.port}", self.login, self.password, CONN_TCP), relay) + + # A connection whose host goes silent + + def check_silent_host_is_noticed_and_replaced(self, args, relay): + old = connection_manager.get_connection(*args) + relay.freeze() + outcome = [] + + def call(): + try: + outcome.append(old.getVersion()) + except libvirt.libvirtError as err: + outcome.append(err) + + caller = threading.Thread(target=call, daemon=True) + caller.start() + caller.join(KEEPALIVE_DEATH) # keepalive must end the call by then + if caller.is_alive(): + relay.close() # cut the connection so the stuck call returns + self.fail(f"a call on a silent connection still waits after {KEEPALIVE_DEATH} s") + self.assertIsInstance(outcome[0], libvirt.libvirtError) + relay.thaw() + new, _ = timed(*args) + self.assertNotIsInstance(new, libvirt.libvirtError) + self.assertIsNot(new, old) + self.assertTrue(new.getVersion()) + + def test_a_silent_tcp_host_is_noticed_and_reconnected(self): + relay = Relay((self.host, 16509)) + self.addCleanup(relay.close) + self.check_silent_host_is_noticed_and_replaced((f"127.0.0.1:{relay.port}", self.login, self.password, CONN_TCP), relay) + + # ssh + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_HOST"), "Set TEST_LIBVIRT_SSH_HOST (and _LOGIN) for an ssh compute") + def test_a_reachable_ssh_host_connects_through_the_wrapper(self): + conn, _ = timed(self.ssh_host, self.ssh_login, "", CONN_SSH) + self.assertNotIsInstance(conn, libvirt.libvirtError) + self.assertIn("command=", conn.getURI()) + self.assertTrue(conn.getVersion()) + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)") + def test_an_ssh_config_alias_is_used(self): + conn, _ = timed("wvc-alias", self.ssh_login, "", CONN_SSH) + self.assertNotIsInstance(conn, libvirt.libvirtError) + self.assertTrue(conn.getVersion()) + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)") + def test_an_ssh_port_that_never_answers_fails_within_the_timeout(self): + silent = socket.socket() + silent.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + silent.bind(("127.0.0.1", 2223)) + silent.listen(5) # no ssh banner ever comes + self.addCleanup(silent.close) + result, took = timed("wvc-silent", self.ssh_login, "", CONN_SSH) + self.assertIsInstance(result, libvirt.libvirtError) + self.assertLess(took, CONNECT_TIMEOUT + 1) + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)") + def test_a_slow_ssh_open_that_completes_later_is_used(self): + relay = Relay((self.ssh_host, 22), port=2222, frozen=True) + self.addCleanup(relay.close) + self.check_slow_open_completes_later(("wvc-relay", self.ssh_login, "", CONN_SSH), relay) + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_ALIASES"), "Set TEST_LIBVIRT_SSH_ALIASES=1 (see the module docstring)") + def test_a_silent_ssh_host_is_noticed_and_reconnected(self): + relay = Relay((self.ssh_host, 22), port=2222) + self.addCleanup(relay.close) + self.check_silent_host_is_noticed_and_replaced(("wvc-relay", self.ssh_login, "", CONN_SSH), relay) + + @unittest.skipUnless(os.environ.get("TEST_LIBVIRT_SSH_JUMP"), "Set TEST_LIBVIRT_SSH_JUMP=1 (see the module docstring)") + def test_ssh_through_a_jump_host(self): + conn, _ = timed("wvc-jump", self.ssh_login, "", CONN_SSH) + self.assertNotIsInstance(conn, libvirt.libvirtError) + self.assertTrue(conn.getVersion())