fix(vrtManager): host_is_up returns a bool; slow hosts block no others

- host_is_up returned the exception object for a down host, which is
  truthy: check_user_quota then connected to it and failed, so a user
  with a VM on a down host could not open its page, clone or resize,
  and migrate_instance's "target is down" check never fired. It returns
  True or False now, closes its socket, and accepts IPv6 and host:port.
- The connection manager no longer holds its global lock while a new
  connection is opened, so one slow host does not delay new connections
  to the others.

Bounding the connection time itself (including SSH aliases and jump
hosts) is left for a separate change.

computes/test_quota_down_host.py

@@ -0,0 +1,25 @@
+import socket
+
+from accounts.models import UserAttributes, UserInstance
+from appsettings.settings import get_settings
+from computes.models import Compute
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from instances.models import Instance
+from instances.utils import check_user_quota
+
+
+class QuotaWithDownHostTestCase(TestCase):
+    def test_vm_on_an_unreachable_host_does_not_break_the_quota_check(self):
+        sock = socket.socket()
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+        sock.close()
+        user = get_user_model().objects.create_user("quota_down", password="pw")
+        UserAttributes.objects.create(user=user, max_instances=5, max_cpus=8, max_memory=8192, max_disk_size=100)
+        compute = Compute.objects.create(name="down", hostname=f"127.0.0.1:{port}", login="", password="", type=1)
+        vm = Instance.objects.create(compute=compute, name="vm", uuid="55555555-5555-5555-5555-555555555555")
+        UserInstance.objects.create(user=user, instance=vm)
+        get_settings()  # app settings are normally loaded by middleware
+
+        self.assertEqual(check_user_quota(user, 1, 1, 128, 1), "")

vrtManager/connection.py

@@ -65,9 +65,6 @@ class wvmConnection(object):
         self.passwd = passwd
         self.type = conn
 
-        # connect
-        self.connect()
-
     def connect(self):
         self.connection_state_lock.acquire()
         try:
@@ -281,19 +278,15 @@ class wvmConnectionManager(object):
                 # as the thread previously holding the write lock may have already added our connection
                 connection = self._search_connection(host, login, passwd, conn)
                 if connection is None:
-                    # create a new connection if a matching connection does not already exist
+                    # Only register it here: connecting under the global lock
+                    # would make every other host wait for a slow one.
                     connection = wvmConnection(host, login, passwd, conn)
-
-                    # add new connection to connection dict
-                    if host in self._connections:
-                        self._connections[host].append(connection)
-                    else:
-                        self._connections[host] = [connection]
+                    self._connections.setdefault(host, []).append(connection)
             finally:
                 self._connections_lock.release()
 
-        elif not connection.connected:
-            # try to (re-)connect if connection is closed
+        if not connection.connected:
+            # (re-)connect; the connection's own lock serializes concurrent attempts
             connection.connect()
 
         if connection.connected:
@@ -303,34 +296,33 @@ class wvmConnectionManager(object):
             # raise libvirt error
             raise util.OperationError(connection.last_error)
 
-    def host_is_up(self, conn_type, hostname):
-        """
-        returns True if the given host is up and we are able to establish
-        a connection using the given credentials.
-        """
+    def host_is_up(self, conn_type, hostname, timeout=1):
+        """True if the libvirt endpoint of hostname accepts a TCP (or Unix
+        socket) connection. This is a reachability probe only: it does not
+        check credentials or libvirt itself. hostname may carry a port
+        ("host:port", "[v6]:port")."""
+        if conn_type == CONN_SOCKET:
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            try:
+                sock.settimeout(timeout)
+                sock.connect("/var/run/libvirt/libvirt-sock")
+                return True
+            except OSError:
+                return False
+            finally:
+                sock.close()
+        default_port = {CONN_SSH: SSH_PORT, CONN_TCP: TCP_PORT, CONN_TLS: TLS_PORT}.get(conn_type)
+        if default_port is None:
+            return False
+        host, port = str(hostname), default_port
+        match = re.fullmatch(r"\[?([^\[\]]+?)\]?:(\d+)", host)
+        if match and (host.startswith("[") or host.count(":") == 1):
+            host, port = match.group(1), int(match.group(2))
         try:
-            socket_host = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            socket_host.settimeout(1)
-            if conn_type == CONN_SSH:
-                if ":" in hostname:
-                    libvirt_host, PORT = hostname.split(":")
-                    PORT = int(PORT)
-                else:
-                    PORT = SSH_PORT
-                    libvirt_host = hostname
-                socket_host.connect((libvirt_host, PORT))
-            if conn_type == CONN_TCP:
-                socket_host.connect((hostname, TCP_PORT))
-            if conn_type == CONN_TLS:
-                socket_host.connect((hostname, TLS_PORT))
-            if conn_type == CONN_SOCKET:
-                socket_host = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-                socket_host.connect("/var/run/libvirt/libvirt-sock")
-            socket_host.close()
-            return True
-        except Exception as err:
-            return err
-
+            with socket.create_connection((host.strip("[]"), port), timeout=timeout):
+                return True
+        except OSError:
+            return False
 
 connection_manager = wvmConnectionManager(
     settings.LIBVIRT_KEEPALIVE_INTERVAL if hasattr(settings, "LIBVIRT_KEEPALIVE_INTERVAL") else 5,

vrtManager/test_connection.py

@@ -0,0 +1,65 @@
+"""R-09: reachability probe and connections that do not block each other."""
+
+import socket
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+from django.conf import settings
+
+if not settings.configured:
+    settings.configure(MAC_OUI="52:54:10")
+
+from vrtManager.connection import CONN_TCP, wvmConnectionManager
+
+
+def closed_port():
+    sock = socket.socket()
+    sock.bind(("127.0.0.1", 0))
+    port = sock.getsockname()[1]
+    sock.close()
+    return port
+
+
+class HostIsUpTestCase(unittest.TestCase):
+    def setUp(self):
+        self.manager = wvmConnectionManager.__new__(wvmConnectionManager)
+
+    def test_reachable_host_is_true(self):
+        server = socket.socket()
+        server.bind(("127.0.0.1", 0))
+        server.listen(1)
+        self.addCleanup(server.close)
+        self.assertIs(self.manager.host_is_up(CONN_TCP, f"127.0.0.1:{server.getsockname()[1]}"), True)
+
+    def test_unreachable_host_is_false_not_an_exception(self):
+        self.assertIs(self.manager.host_is_up(CONN_TCP, f"127.0.0.1:{closed_port()}"), False)
+
+
+class ConnectionManagerTestCase(unittest.TestCase):
+    def setUp(self):
+        patcher = patch("vrtManager.connection.wvmEventLoop")
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.manager = wvmConnectionManager()
+
+    def test_a_slow_host_does_not_block_other_hosts(self):
+        def open_auth(uri, auth, flags):
+            if "slow" in uri:
+                time.sleep(1.5)
+            raise __import__("libvirt").libvirtError("refused")
+
+        with patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth):
+            slow = threading.Thread(target=lambda: self.assertRaises(Exception, self.manager.get_connection, "slow", "u", "p", CONN_TCP))
+            slow.start()
+            time.sleep(0.2)
+            started = time.monotonic()
+            with self.assertRaises(Exception):
+                self.manager.get_connection("fast", "u", "p", CONN_TCP)
+            self.assertLess(time.monotonic() - started, 1.0)
+            slow.join()
+
+
+if __name__ == "__main__":
+    unittest.main()