From 6ca65e291ae9a577568471c51fee55fd256f52a6 Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 3 Nov 2026 00:22:19 +0300 Subject: [PATCH] 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 | 1 + operations/tests/fakes.py | 45 ++++++++++++++ operations/tests/test_recovery.py | 4 ++ operations/tests/test_worker.py | 36 +++++++++++- operations/worker.py | 98 +++++++++++++++++++++++++------ 5 files changed, 164 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 80f9017d..9b930022 100644 --- a/README.md +++ b/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`. diff --git a/operations/tests/fakes.py b/operations/tests/fakes.py index 51b04db4..3ddd272a 100644 --- a/operations/tests/fakes.py +++ b/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() diff --git a/operations/tests/test_recovery.py b/operations/tests/test_recovery.py index 775f4ca3..d93851c9 100644 --- a/operations/tests/test_recovery.py +++ b/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): diff --git a/operations/tests/test_worker.py b/operations/tests/test_worker.py index 8ad15a84..12c1411c 100644 --- a/operations/tests/test_worker.py +++ b/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) + diff --git a/operations/worker.py b/operations/worker.py index ca42b109..d7c17562 100644 --- a/operations/worker.py +++ b/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:" 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:" 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):