feat(vrtManager): a host that does not answer holds two requests of a worker

With threaded workers a request waiting on a stuck host holds one
thread, but each new request to that host (a VM list's poll every 5 s,
an F5) held another until the threads ran out.

In a process at most two web requests now use one host at a time,
counted by its name whatever the login. Another waits up to the connect
timeout (5 s) for one of them to end, and is then refused as the host
not answering: VM lists show the host as not reachable, pages show the
connection error. A request counts on the host it connected to last,
until it ends, so one that reads several hosts in turn holds none it
has left. Threads outside a request (the job worker's, the consoles')
are not counted: a job holds its host for as long as it runs.

vrtManager/connection.py

@@ -10,6 +10,7 @@ from urllib.parse import quote
 
 import libvirt
 from django.conf import settings
+from django.core.signals import request_finished, request_started
 from django.utils.functional import cached_property
 from libvirt import libvirtError
 from lxml import etree
@@ -33,6 +34,16 @@ TCP_PORT = 16509
 # not tried again for RETRY_AFTER seconds.
 CONNECT_TIMEOUT = getattr(settings, "LIBVIRT_CONNECT_TIMEOUT", 5)
 RETRY_AFTER = getattr(settings, "LIBVIRT_RETRY_AFTER", 30)
+# A call to a host that is up but stuck (a hung NFS mount, for one) waits as
+# long as the host does: libvirt has no call timeout, and keepalive does not
+# end it, since the daemon still answers keepalives. So in a process at most
+# HOST_REQUESTS web requests use one host (by its name, whatever the login)
+# at a time; another waits up to
+# CONNECT_TIMEOUT for one of them to end, and is then refused. A request
+# counts on the host it connected to last, until it ends. Threads outside a
+# request (the job worker's, the consoles') are not counted: a job holds its
+# host for as long as it runs.
+HOST_REQUESTS = getattr(settings, "LIBVIRT_HOST_REQUESTS", 2)
 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)
@@ -92,6 +103,41 @@ def _after_fork_in_child():
 os.register_at_fork(after_in_child=_after_fork_in_child)
 
 
+_request = threading.local()  # in a web request: the connection it counts on (HOST_REQUESTS)
+
+
+def _request_started(**kwargs):
+    _leave_host()  # a request that ended without request_finished (an error outside Django's handling)
+    _request.active = True
+
+
+def _request_finished(**kwargs):
+    _leave_host()
+    _request.active = False
+
+
+def _leave_host():
+    held = getattr(_request, "held", None)
+    if held is not None:
+        _request.held = None
+        held.requests.release()
+
+
+def _count_on(connection):
+    """A web request uses the host of connection from now on (HOST_REQUESTS)."""
+    if not getattr(_request, "active", False) or getattr(_request, "held", None) is connection:
+        return
+    _leave_host()
+    if not connection.requests.acquire(timeout=CONNECT_TIMEOUT):
+        raise util.ConnectionFailed(
+            f"Connection Failed: {connection.host} does not answer: {HOST_REQUESTS} requests still wait for it")
+    _request.held = connection
+
+
+request_started.connect(_request_started, dispatch_uid="vrtManager.connection.request_started")
+request_finished.connect(_request_finished, dispatch_uid="vrtManager.connection.request_finished")
+
+
 # libvirt's per-host directory of the VMs' UEFI variables (NVRAM)
 NVRAM_DIR = "/var/lib/libvirt/qemu/nvram"
 
@@ -112,6 +158,7 @@ class wvmConnection(object):
         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
+        self.requests = None  # the web requests using the host (HOST_REQUESTS), set by the manager
 
         # credentials
         self.host = host
@@ -293,6 +340,7 @@ class wvmConnectionManager(object):
         self._connections = dict()
         self._connections_lock = ReadWriteLock()
         self._inherited = []
+        self._host_requests = {}  # host: the web requests using it, whatever the login (HOST_REQUESTS)
 
     def forget_connections(self):
         """For a forked child: stop using the inherited connections (the
@@ -301,6 +349,7 @@ class wvmConnectionManager(object):
         self._inherited.append(self._connections)
         self._connections = dict()
         self._connections_lock = ReadWriteLock()
+        self._host_requests = {}
 
     def _search_connection(self, host, login, passwd, conn):
         """
@@ -343,10 +392,13 @@ class wvmConnectionManager(object):
                     # 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)
+                    connection.requests = self._host_requests.setdefault(
+                        host, threading.BoundedSemaphore(HOST_REQUESTS))
                     self._connections.setdefault(host, []).append(connection)
             finally:
                 self._connections_lock.release()
 
+        _count_on(connection)
         live = connection.alive_connection()
         if live is None:
             # (re-)connect within CONNECT_TIMEOUT; one open per connection at a time

vrtManager/tests/test_connection.py

@@ -8,6 +8,7 @@ import unittest
 from unittest.mock import MagicMock, patch
 
 from django.conf import settings
+from django.core.signals import request_finished, request_started
 
 if not settings.configured:
     settings.configure(MAC_OUI="52:54:10")
@@ -214,6 +215,91 @@ class BoundedOpenTestCase(unittest.TestCase):
             self.assertIn("ConnectTimeout", script.read())
 
 
+class HostRequestsTestCase(unittest.TestCase):
+    """At most HOST_REQUESTS web requests of a process use one host at a
+    time; another waits CONNECT_TIMEOUT and is refused."""
+
+    def setUp(self):
+        for name, value in (("wvmEventLoop", MagicMock()), ("CONNECT_TIMEOUT", 0.2)):
+            patcher = patch(f"vrtManager.connection.{name}", value)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        patcher = patch("vrtManager.connection.libvirt.openAuth", side_effect=lambda *args: alive_connection())
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.manager = wvmConnectionManager()
+        self.done = threading.Event()
+        self.threads = []
+        self.addCleanup(self.end_requests)
+
+    def end_requests(self):
+        self.done.set()
+        for thread in self.threads:
+            thread.join(5)
+        self.threads, self.done = [], threading.Event()
+
+    def request(self, *hosts, login="u"):
+        """A web request on its own thread: it connects to hosts in turn and
+        ends when self.done is set. Returns its errors, filled once connected."""
+        errors, connected, done = [], threading.Event(), self.done
+
+        def run():
+            request_started.send(sender=None)
+            try:
+                for host in hosts:
+                    self.manager.get_connection(host, login, "p", CONN_TCP)
+            except libvirt.libvirtError as err:
+                errors.append(str(err))
+            connected.set()
+            done.wait(5)
+            request_finished.send(sender=None)
+
+        thread = threading.Thread(target=run)
+        thread.start()
+        self.threads.append(thread)
+        connected.wait(5)
+        return errors
+
+    def test_a_third_request_to_a_host_is_refused_and_other_hosts_answer(self):
+        self.assertEqual(self.request("h") + self.request("h"), [])
+        started = time.monotonic()
+        self.assertEqual(self.request("h"), ["Connection Failed: h does not answer: 2 requests still wait for it"])
+        self.assertGreaterEqual(time.monotonic() - started, 0.2)
+        self.assertEqual(self.request("g"), [])
+
+    def test_a_host_counts_its_requests_whatever_the_login(self):
+        # two records of one host (another login, or credentials changed meanwhile)
+        self.assertEqual(self.request("h") + self.request("h", login="other"), [])
+        self.assertEqual(len(self.request("h", login="third")), 1)
+
+    def test_a_request_without_request_finished_leaves_its_host_at_the_next_one(self):
+        self.assertEqual(self.request("h"), [])
+        request_started.send(sender=None)  # this thread: a request that never got request_finished
+        try:
+            self.manager.get_connection("h", "u", "p", CONN_TCP)
+            request_started.send(sender=None)  # its next request on the thread
+            self.assertEqual(self.request("h"), [])
+        finally:
+            request_finished.send(sender=None)
+
+    def test_a_request_counts_once_on_a_host(self):
+        self.assertEqual(self.request("h", "h", "h") + self.request("h"), [])
+        self.assertEqual(len(self.request("h")), 1)
+
+    def test_a_request_counts_on_the_host_it_connected_to_last(self):
+        self.assertEqual(self.request("h", "g") + self.request("h") + self.request("h"), [])
+        self.assertEqual(len(self.request("g", "h")), 1)
+
+    def test_an_ended_request_lets_the_next_one_in(self):
+        self.assertEqual(self.request("h") + self.request("h"), [])
+        self.end_requests()
+        self.assertEqual(self.request("h") + self.request("h"), [])
+
+    def test_threads_outside_a_request_are_not_counted(self):
+        self.assertEqual(self.request("h") + self.request("h"), [])
+        self.manager.get_connection("h", "u", "p", CONN_TCP)
+
+
 class EventLoopTestCase(unittest.TestCase):
     """The libvirt event loop thread starts with the first connection of a
     process, not at import: novncd forks a child per console after importing