diff --git a/vrtManager/connection.py b/vrtManager/connection.py index d3a7c6b5..ccd2191f 100644 --- a/vrtManager/connection.py +++ b/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 diff --git a/vrtManager/tests/test_connection.py b/vrtManager/tests/test_connection.py index ad3115a1..54940e49 100644 --- a/vrtManager/tests/test_connection.py +++ b/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