diff --git a/README.md b/README.md index aa8dd313..82100e77 100644 --- a/README.md +++ b/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 diff --git a/gunicorn.conf.py b/gunicorn.conf.py index 04ced8b4..be0084af 100644 --- a/gunicorn.conf.py +++ b/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 diff --git a/instances/addresses.py b/instances/addresses.py index c6ea0edf..4c0f8737 100644 --- a/instances/addresses.py +++ b/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)} diff --git a/instances/tests/test_addresses.py b/instances/tests/test_addresses.py index 6f71995b..ba8c8623 100644 --- a/instances/tests/test_addresses.py +++ b/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)) diff --git a/webvirtcloud/gunicorn_worker.py b/webvirtcloud/gunicorn_worker.py new file mode 100644 index 00000000..98c3a990 --- /dev/null +++ b/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() diff --git a/webvirtcloud/tests/test_gunicorn_conf.py b/webvirtcloud/tests/test_gunicorn_conf.py index 978cfd86..11945f25 100644 --- a/webvirtcloud/tests/test_gunicorn_conf.py +++ b/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)) diff --git a/webvirtcloud/tests/test_gunicorn_worker.py b/webvirtcloud/tests/test_gunicorn_worker.py new file mode 100644 index 00000000..6bea26ac --- /dev/null +++ b/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()