fix(operations): a host that does not answer keeps the job worker going for the others

A job's libvirt call to a host that is up but stuck waits as long as
the host does. Measured with the worker's loop: with two jobs stuck on
one host, the other host's jobs ran but their progress was never read
(both progress threads waited on the stuck host); with four stuck, a
new job of another host stayed queued, every slot taken; cancelling
them took all four control threads, so cancels, recovery and
housekeeping stopped for every host.

Now a host's jobs take at most half of the worker's slots (at least
one; a migration counts on both its hosts), and the next queued job of
another host starts instead. A host has at most one progress read and
one control task (a cancel or a recovery inspection) at a time, and
the control tasks of hosts leave one thread to housekeeping.

README.md

@@ -599,6 +599,7 @@ A version from before v3.0.0 has no worker: `systemctl disable --now webvirt-wor
 - Changing one's own password is the "Can change password" permission (per user or group; new users have it); `SHOW_PROFILE_EDIT_PASSWORD` is gone.
 - The admin's first-login password change survives losing `data/admin_password` (migration `accounts.0007`); run `migrate` as a user who can read that file.
 - "VM Clone Auto Migrate" is removed; a clone stays on its source host.
+- The job worker runs at most half of its concurrency (Settings, at least one) of one host's jobs at a time; that host's other jobs wait in the queue, so a host that does not answer cannot take every slot.
 - A template VM is changed or deleted only by superusers and staff owners with the matching permission; other owners can still view, open the console and clone it. Power cycle no longer starts a template.
 - A VM that vanishes from its compute, or whose compute is deleted, keeps its owners for `INSTANCE_OWNERSHIP_RETENTION_DAYS` (30) in case the same UUID comes back; superusers list and remove these records on the "Removed VMs" admin page.
 - The log rotates (10 MB × 5) in `data/webvirtcloud.log`.

operations/tests/fakes.py

@@ -21,6 +21,8 @@ def reset():
     BARRIERS.clear()
     CANCELLED.clear()
     EXPIRED.clear()
+    STUCK.clear()
+    HOST_CALLS.clear()
 
 
 def start_worker(worker):
@@ -127,3 +129,46 @@ class RequestKind(NoopKind):
         if op.params["fail_expire"]:
             raise RuntimeError("expire failed")
         EXPIRED.append(str(op.id))
+
+
+STUCK = {}  # compute id -> threading.Event: test.host's libvirt calls on it wait until it is set
+HOST_CALLS = []  # (call, compute id) as test.host's progress and cancel begin
+
+
+@register
+class HostKind(NoopKind):
+    """A job on compute params["host"], whose calls wait while STUCK holds it,
+    as calls to a host that is up but does not answer do."""
+
+    name = "test.host"
+    label = "Host"
+    cancellable = True
+
+    def normalize(self, raw):
+        return {"host": int(raw["host"]), "n": int(raw["n"])}
+
+    def plan(self, user, params):
+        key = f"instance:host-{params['host']}-{params['n']}"
+        return Plan(claims={f"compute:{params['host']}": Claim.SHARED, key: Claim.EXCLUSIVE}, resource=key,
+                    label="c/vm")
+
+    def _call(self, name, op):
+        HOST_CALLS.append((name, op.params["host"]))
+        if (stuck := STUCK.get(op.params["host"])) is not None:
+            stuck.wait(30)
+
+    def run(self, op, ctx):
+        ctx.record_intent(step="effect")
+        self._call("run", op)
+        if (release := BARRIERS.get(op.submission_key)) is not None:
+            release.wait(30)
+        return {}
+
+    def progress(self, op):
+        self._call("progress", op)
+        return (1, 2)
+
+    def cancel(self, op):
+        self._call("cancel", op)
+        if (release := BARRIERS.get(op.submission_key)) is not None:
+            release.set()

operations/tests/test_recovery.py

@@ -20,6 +20,10 @@ class FakeWorker:
     def __init__(self):
         self.control = ThreadPoolExecutor(1)
 
+    def run_control(self, hosts, fn, *args):
+        self.control.submit(fn, *args)
+        return True
+
 
 class RecoveryTestCase(TransactionTestCase):
     def setUp(self):

operations/tests/test_worker.py

@@ -51,12 +51,43 @@ class WorkerTestCase(TransactionTestCase):
         self.thread = fakes.start_worker(self.worker)
 
     def tearDown(self):
-        for barrier in fakes.BARRIERS.values():
-            barrier.set()
+        for event in [*fakes.BARRIERS.values(), *fakes.STUCK.values()]:
+            event.set()
         if self.thread:
             fakes.stop_worker(self.worker, self.thread)
         connection.close()
 
+    def job(self, host, n):
+        return self.submit("test.host", {"host": host, "n": n}, f"key-h{host}-{n:04d}")
+
+    def test_a_stuck_host_takes_half_the_slots_and_one_progress_read(self):
+        fakes.STUCK[1] = threading.Event()
+        stuck = [self.job(1, n) for n in range(4)]
+        self.start(concurrency=4)
+        self.assertTrue(wait_for(lambda: sum(state(op) == Operation.RUNNING for op in stuck) == 2))
+        # the other host's job runs beside them, and its progress is read
+        healthy = self.submit("test.host", {"host": 2, "n": 0}, "key-h2-0000", barrier=True)
+        self.assertTrue(wait_for(lambda: ("progress", 2) in fakes.HOST_CALLS))
+        self.assertEqual(sum(state(op) == Operation.RUNNING for op in stuck), 2)
+        self.assertEqual(fakes.HOST_CALLS.count(("progress", 1)), 1)
+        fakes.BARRIERS["key-h2-0000"].set()
+        self.assertTrue(wait_for(lambda: state(healthy) == Operation.SUCCEEDED))
+
+    def test_a_stuck_host_takes_one_control_thread(self):
+        fakes.STUCK[1] = threading.Event()
+        # they run until cancelled
+        stuck = [self.submit("test.host", {"host": 1, "n": n}, f"key-h1-{n:04d}", barrier=True) for n in range(2)]
+        healthy = self.submit("test.host", {"host": 2, "n": 0}, "key-h2-0000", barrier=True)
+        self.start(concurrency=4)
+        self.assertTrue(wait_for(lambda: all(state(op) == Operation.RUNNING for op in [*stuck, healthy])))
+        for op in [*stuck, healthy]:
+            jobs.cancel(op.pk, self.user)
+        # the other host's cancel goes through; the stuck host's second cancel waits for its first
+        self.assertTrue(wait_for(lambda: ("cancel", 2) in fakes.HOST_CALLS))
+        self.assertEqual(fakes.HOST_CALLS.count(("cancel", 1)), 1)
+        fakes.STUCK[1].set()
+        self.assertTrue(wait_for(lambda: fakes.HOST_CALLS.count(("cancel", 1)) == 2))
+
     def submit(self, kind, params, key, barrier=False):
         if barrier:
             fakes.BARRIERS[key] = threading.Event()  # before the worker can see the job
@@ -316,3 +347,4 @@ class HostIdentityTestCase(TransactionTestCase):
             first = host_identity(self.dir)
         with patch("operations.worker.socket.gethostname", return_value="wvc-test-b"):
             self.assertEqual(host_identity(self.dir), first)
+

operations/worker.py

@@ -12,6 +12,7 @@ import socket
 import threading
 import time
 import uuid
+from collections import Counter
 from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
 from pathlib import Path
@@ -30,6 +31,7 @@ logger = logging.getLogger(__name__)
 
 MAX_SLOTS = 16
 PROGRESS_THREADS = 2
+CONTROL_THREADS = 4
 HOUSEKEEPING_EVERY = 600.0
 
 BACKOFF = (30, 60, 120, 300, 600)
@@ -89,8 +91,17 @@ class Worker:
         self.signalled = set()
         self.progress_pending = set()
         self.progress_asked = {}  # job -> monotonic time its progress was last asked
+        # A libvirt call to a host that is up but stuck waits as long as the
+        # host does: so a host's jobs take at most half of the slots, and it
+        # has at most one progress read and one control task (cancel or
+        # recovery) at a time, which leaves one control thread to housekeeping.
+        self.job_hosts = {}  # job -> its computes ("compute:<id>" claims), while it runs here
+        self.progress_hosts = set()
+        self.control_hosts = set()
+        self.control_tasks = 0  # control tasks for hosts that run
+        self.hosts_lock = threading.Lock()
         self.pool = ThreadPoolExecutor(MAX_SLOTS, thread_name_prefix="wvc-job")
-        self.control = ThreadPoolExecutor(4, thread_name_prefix="wvc-control")
+        self.control = ThreadPoolExecutor(CONTROL_THREADS, thread_name_prefix="wvc-control")
         self.progress_pool = ThreadPoolExecutor(PROGRESS_THREADS, thread_name_prefix="wvc-progress")
         self.recovery = Scheduler(self)
         self._last = {"heartbeat": 0.0, "progress": 0.0, "housekeeping": 0.0}
@@ -192,22 +203,34 @@ class Worker:
         for op_id, future in list(self.jobs.items()):
             if future.done():
                 del self.jobs[op_id]
+                self.job_hosts.pop(op_id, None)
                 self.signalled.discard(op_id)
                 self.progress_asked.pop(op_id, None)
                 if (err := future.exception()) is not None:
                     # execute raised before it could finish the job itself
                     logger.error("job %s ended unfinished: %s", op_id, utils.safe_text(err))
-                    self.control.submit(_guarded, self.recovery.inspect_now, op_id, utils.public_error(err))
+                    self.recovery.inspect_now(op_id, utils.public_error(err))
+
+    def _concurrency(self):
+        return self.fixed_concurrency or utils.concurrency()
 
     def _slots(self):
-        return (self.fixed_concurrency or utils.concurrency()) - len(self.jobs)
+        return self._concurrency() - len(self.jobs)
 
     def _claim(self):
-        while self._slots() > 0 and not self.stopping.is_set():
-            candidate = (Operation.objects.filter(state=Operation.QUEUED).order_by("created_at")
-                         .values_list("pk", flat=True).first())
-            if candidate is None:
+        """The oldest queued jobs, but none of a host that has half the slots
+        already (a job of two hosts counts on both): one that does not answer
+        leaves the others' jobs the rest."""
+        if self._slots() <= 0:
+            return
+        per_host = max(1, self._concurrency() // 2)
+        running = Counter(host for hosts in self.job_hosts.values() for host in hosts)
+        queued = Operation.objects.filter(state=Operation.QUEUED).order_by("created_at").values_list("pk", flat=True)
+        for candidate, hosts in hosts_of(list(queued)).items():
+            if self._slots() <= 0 or self.stopping.is_set():
                 return
+            if any(running[host] >= per_host for host in hosts):
+                continue
             try:
                 with fenced(self.identity):
                     now = timezone.now()
@@ -217,13 +240,36 @@ class Worker:
                 self._lose_lease()
                 return
             if taken:
+                self.job_hosts[candidate] = hosts
+                running.update(hosts)
                 self.jobs[candidate] = self.pool.submit(execute, self, candidate)
 
+    def run_control(self, hosts, fn, *args):
+        """fn on a control thread, unless one of hosts has a control task
+        still running, or it would take the thread left to housekeeping.
+        True if it started."""
+        with self.hosts_lock:
+            if not hosts.isdisjoint(self.control_hosts) or (hosts and self.control_tasks >= CONTROL_THREADS - 1):
+                return False
+            self.control_hosts |= hosts
+            self.control_tasks += bool(hosts)
+
+        def task():
+            try:
+                fn(*args)
+            finally:
+                with self.hosts_lock:
+                    self.control_hosts -= hosts
+                    self.control_tasks -= bool(hosts)
+
+        self.control.submit(task)
+        return True
+
     def _cancels(self):
         for op in Operation.objects.filter(pk__in=list(self.jobs), cancel_requested=True):
-            if op.pk not in self.signalled:
+            if op.pk not in self.signalled and self.run_control(
+                    self.job_hosts.get(op.pk, frozenset()), _guarded, jobs.get_kind(op.kind).cancel, op):
                 self.signalled.add(op.pk)
-                self.control.submit(_guarded, jobs.get_kind(op.kind).cancel, op)
 
     def _limits(self):
         now = timezone.now()
@@ -240,12 +286,16 @@ class Worker:
         for op in sorted(ops, key=lambda op: self.progress_asked.get(op.pk, 0.0)):
             if len(self.progress_pending) >= PROGRESS_THREADS:
                 return
-            if op.pk not in self.progress_pending:
-                self.progress_pending.add(op.pk)
-                self.progress_asked[op.pk] = time.monotonic()
-                self.progress_pool.submit(self._read_progress, op)
+            hosts = self.job_hosts.get(op.pk, frozenset())
+            with self.hosts_lock:
+                if op.pk in self.progress_pending or not hosts.isdisjoint(self.progress_hosts):
+                    continue  # one read a host: one that does not answer holds one thread
+                self.progress_hosts |= hosts
+            self.progress_pending.add(op.pk)
+            self.progress_asked[op.pk] = time.monotonic()
+            self.progress_pool.submit(self._read_progress, op, hosts)
 
-    def _read_progress(self, op):
+    def _read_progress(self, op, hosts):
         try:
             value = jobs.get_kind(op.kind).progress(op)
             if value is not None:
@@ -254,10 +304,21 @@ class Worker:
         except Exception as err:
             logger.debug("progress of job %s: %s", op.pk, utils.safe_text(err))
         finally:
+            with self.hosts_lock:
+                self.progress_hosts -= hosts
             self.progress_pending.discard(op.pk)
             close_old_connections()
 
 
+def hosts_of(op_ids):
+    """Each job's computes, by its "compute:<id>" claims, in the order of op_ids."""
+    hosts = {op_id: set() for op_id in op_ids}
+    for op_id, resource in Claim.objects.filter(operation_id__in=op_ids, resource__startswith="compute:") \
+            .values_list("operation_id", "resource"):
+        hosts[op_id].add(resource)
+    return {op_id: frozenset(found) for op_id, found in hosts.items()}
+
+
 def _guarded(fn, *args):
     try:
         fn(*args)
@@ -336,10 +397,11 @@ class Scheduler:
 
     def due(self):
         now = time.monotonic()
-        for op_id, at in list(self.next_try.items()):
-            if at <= now and op_id not in self.inflight:
-                self.inflight.add(op_id)
-                self.worker.control.submit(self._inspect, op_id)
+        due = [op_id for op_id, at in list(self.next_try.items()) if at <= now and op_id not in self.inflight]
+        for op_id, hosts in hosts_of(due).items():
+            self.inflight.add(op_id)  # before it starts: _inspect discards it when it ends
+            if not self.worker.run_control(hosts, self._inspect, op_id):
+                self.inflight.discard(op_id)  # its host's control task still runs: a later tick
 
     def _done(self, op_id):
         for table in (self.next_try, self.attempts, self.errors, self.stalled_since):