feat(webvirtcloud): a request waiting on a host holds one thread, not a worker

A host that is up but does not answer (a hung NFS mount, for one) keeps
a libvirt call waiting: libvirt's keepalive does not end it, since the
daemon still answers keepalives. Each such request held a whole sync
worker, and once they held every worker the panel stopped answering,
pages without libvirt too; F5 added another.

gunicorn now runs threaded workers, 4 threads each, and a worker takes
no more connections than it has threads, so one whose threads are all
busy leaves new requests to the others. A threaded worker keeps
reporting in while a thread waits, so gunicorn's timeout no longer ends
the request: the worker now exits itself when a request runs past the
timeout (600 s), and gunicorn starts a new one. Short claims expire on
that bound.

The address search runs one at a time in a process, as it did with one
request per worker.

README.md

@@ -319,7 +319,7 @@ Where to set them:
 
 - **Docker:** the `environment:` list of `docker-compose.override.yml`; every process in the container gets it.
 - **Installer and manual installation:** edit `webvirtcloud/settings.py`, or put them in `/etc/default/webvirtcloud` (root, mode 600), one `NAME='value'` per line (single quotes: systemd and the shell then read the same value); all four services read it. Then `systemctl restart webvirt-cloud webvirt-novncd webvirt-socketiod webvirt-worker`, and load the file before running `manage.py`: `set -a; . /etc/default/webvirtcloud; set +a`.
-- **gunicorn** runs at most 8 workers (fewer on hosts with under 4 CPUs); to change it, set `GUNICORN_CMD_ARGS=--workers N` in a drop-in of `webvirt-cloud` (`systemctl edit webvirt-cloud`: `[Service]` and `Environment="GUNICORN_CMD_ARGS=--workers N"`) or in `docker-compose.override.yml`.
+- **gunicorn** runs at most 8 workers (fewer on hosts with under 4 CPUs) of 4 threads each; to change it, set `GUNICORN_CMD_ARGS=--workers N` in a drop-in of `webvirt-cloud` (`systemctl edit webvirt-cloud`: `[Service]` and `Environment="GUNICORN_CMD_ARGS=--workers N"`) or in `docker-compose.override.yml`.
 
 ### HTTPS and reverse proxy
 

gunicorn.conf.py

@@ -85,10 +85,16 @@ def get_workers():
 
 
 workers = get_workers()
-# worker_class = 'egg:gunicorn#eventlet'
-worker_connections = 1000
+# threads of a worker: a request waiting on a host that does not answer
+# holds one thread, not the whole worker
+worker_class = 'webvirtcloud.gunicorn_worker.Worker'
+threads = 4
+# a worker takes no more connections than it has threads: one whose threads
+# are all busy leaves new requests to the others. nginx opens a connection
+# per request, so none is kept alive.
+worker_connections = threads
 timeout = 600
-keepalive = 2
+keepalive = 0
 # gunicorn 26's control socket is unused, and the service user cannot create it
 control_socket_disable = True
 

instances/addresses.py

@@ -36,11 +36,12 @@ SOURCES = {
 # A read may outlast the search (a guest agent has AGENT_TIMEOUT seconds, a
 # host its connection's timeouts): its thread goes on after the search
 # answers, so a search starts only while its threads fit in MAX_RUNNING
-# search threads in this process.
+# search threads in this process, and one search at a time.
 MAX_WORKERS = 8
 MAX_RUNNING = 16
 _running = 0
 _running_lock = threading.Lock()
+_searching = threading.Lock()
 
 
 class Busy(Exception):
@@ -221,7 +222,7 @@ def _survey(targets, network, deadline, everything, with_os=False, mac=None):
     (late or failed). Returns the VMs' records, the observations, the agents'
     OS by vm, and the counts.
     """
-    if _running > MAX_RUNNING - MAX_WORKERS:
+    if _running > MAX_RUNNING - MAX_WORKERS or not _searching.acquire(blocking=False):
         raise Busy
     end = time.monotonic() + deadline
     records, observations, hosts_down, agent_reads, os_reads, systems = [], [], [], [], {}, {}
@@ -267,6 +268,7 @@ def _survey(targets, network, deadline, everything, with_os=False, mac=None):
                 systems[vm] = done.result()
     finally:
         pool.shutdown(wait=False, cancel_futures=True)
+        _searching.release()
     return records, observations, systems, {
         "checked": checked, "no_agent": no_agent, "unread": unread, "hosts_down": sorted(hosts_down),
         "hosts_unfinished": sorted(hosts[future].name for future in waiting)}

instances/tests/test_addresses.py

@@ -290,6 +290,16 @@ class FindTests(unittest.TestCase):
         finally:
             addresses._running = 0
 
+    def test_a_second_search_waits_for_none(self):
+        # threads of one worker search at once: a second search of the process is refused
+        addresses._searching.acquire()
+        try:
+            with self.assertRaises(addresses.Busy):
+                self.find(addresses.Target(compute("c1"), None))
+        finally:
+            addresses._searching.release()
+        self.find(addresses.Target(compute("c1"), None))  # the search lets the next one in
+
     def test_a_failed_agent_read_is_not_a_read(self):
         dom = FakeDomain({AGENT: libvirtError("guest-network-get-interfaces is disabled")})
         result = self.find(addresses.Target(compute("c1", dom), None))

webvirtcloud/gunicorn_worker.py

@@ -0,0 +1,38 @@
+"""gunicorn's threaded worker with the sync worker's request timeout."""
+
+import os
+import threading
+import time
+
+from gunicorn.workers.gthread import ThreadWorker
+
+
+class Worker(ThreadWorker):
+    """gunicorn kills a sync worker whose request runs past its timeout; a
+    threaded worker keeps reporting in while one of its threads waits on a
+    host that does not answer. This one exits when a request runs past the
+    timeout, and gunicorn starts a new one: a request ends by then, as the
+    short claims of operations.claims expect."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.started = {}  # thread: when its request started
+        self.started_lock = threading.Lock()
+
+    def handle(self, conn):
+        thread = threading.get_ident()
+        with self.started_lock:
+            self.started[thread] = time.monotonic()
+        try:
+            return super().handle(conn)
+        finally:
+            with self.started_lock:
+                del self.started[thread]
+
+    def notify(self):
+        with self.started_lock:
+            oldest = min(self.started.values(), default=None)
+        if oldest is not None and time.monotonic() - oldest > self.cfg.timeout:
+            self.log.critical("A request ran past %s s: the worker exits", self.cfg.timeout)
+            os._exit(1)  # its threads cannot be stopped, and exit() would wait for them
+        super().notify()

webvirtcloud/tests/test_gunicorn_conf.py

@@ -24,3 +24,9 @@ class GunicornConfTestCase(unittest.TestCase):
 
     def test_unknown_cpu_count_falls_back(self):
         self.assertEqual(load(-1)["workers"], 3)
+
+    def test_workers_run_requests_in_threads(self):
+        conf = load(2)
+        self.assertEqual((conf["worker_class"], conf["threads"]), ("webvirtcloud.gunicorn_worker.Worker", 4))
+        # a worker with every thread busy takes no new connection
+        self.assertEqual((conf["worker_connections"], conf["keepalive"]), (4, 0))

webvirtcloud/tests/test_gunicorn_worker.py

@@ -0,0 +1,56 @@
+"""The threaded worker exits when a request runs past gunicorn's timeout."""
+
+import threading
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from gunicorn.workers.gthread import ThreadWorker
+
+from webvirtcloud.gunicorn_worker import Worker
+
+
+def worker():
+    w = Worker.__new__(Worker)
+    w.cfg = SimpleNamespace(timeout=600)
+    w.log = MagicMock()
+    w.started, w.started_lock = {}, threading.Lock()
+    return w
+
+
+class WorkerTestCase(unittest.TestCase):
+    def test_a_request_is_timed_while_it_runs(self):
+        w = worker()
+        seen = []
+        with patch.object(ThreadWorker, "handle", lambda self, conn: seen.append(dict(self.started)) or "done"):
+            self.assertEqual(w.handle("conn"), "done")
+        self.assertEqual(list(seen[0]), [threading.get_ident()])
+        self.assertEqual(w.started, {})
+
+    def test_a_failed_request_is_no_longer_timed(self):
+        w = worker()
+        with patch.object(ThreadWorker, "handle", side_effect=OSError("reset")), self.assertRaises(OSError):
+            w.handle("conn")
+        self.assertEqual(w.started, {})
+
+    def test_the_worker_reports_in_while_its_requests_are_within_the_timeout(self):
+        w = worker()
+        w.started[1] = time.monotonic() - 599
+        with patch.object(ThreadWorker, "notify") as notify, patch("os._exit") as exit_:
+            w.notify()
+        notify.assert_called_once_with()
+        exit_.assert_not_called()
+
+    def test_the_worker_exits_when_a_request_runs_past_the_timeout(self):
+        w = worker()
+        w.started.update({1: time.monotonic(), 2: time.monotonic() - 601})
+        with patch.object(ThreadWorker, "notify") as notify, patch("os._exit", side_effect=SystemExit) as exit_:
+            with self.assertRaises(SystemExit):
+                w.notify()
+        exit_.assert_called_once_with(1)
+        notify.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()