diff --git a/computes/tests/test_api_security.py b/computes/tests/test_api_security.py index b572d4e9..c47765be 100644 --- a/computes/tests/test_api_security.py +++ b/computes/tests/test_api_security.py @@ -5,6 +5,7 @@ from computes.models import Compute from django.contrib.auth import get_user_model from django.test import TestCase from instances.models import Flavor, Instance +from operations import claims from rest_framework.test import APIClient User = get_user_model() @@ -178,12 +179,13 @@ class APISecurityTestCase(TestCase): def test_busy_instance_api_does_not_report_success(self): self.client.force_login(self.regular_user) - with patch("instances.views.libvirt_instance_lock", side_effect=TimeoutError("busy")), patch( + with patch("instances.views.libvirt_instance_lock", side_effect=claims.Busy("instance:wvc-test")), patch( "instances.models.wvmInstance" ) as proxy: response = self.client.post( f"/api/v1/computes/{self.compute.pk}/instances/{self.instance1.pk}/poweron/", {} ) self.assertEqual(response.status_code, 409) - self.assertIn("busy", response.json()["detail"]) + self.assertIn("short operation", response.json()["detail"]) + self.assertIsNone(response.json()["operation"]) proxy.assert_not_called() diff --git a/computes/tests/test_operation_locks.py b/computes/tests/test_operation_locks.py index e9e50622..1aa20045 100644 --- a/computes/tests/test_operation_locks.py +++ b/computes/tests/test_operation_locks.py @@ -1,18 +1,20 @@ import threading from types import SimpleNamespace -from django.test import SimpleTestCase +from django.db import connection +from django.test import TransactionTestCase from computes.utils import libvirt_compute_lock, libvirt_instance_lock +from operations import claims -class OperationLocksTestCase(SimpleTestCase): +class OperationLocksTestCase(TransactionTestCase): def setUp(self): self.compute = SimpleNamespace(pk=987654321) - self.first = SimpleNamespace(pk=987654321, compute=self.compute) - self.second = SimpleNamespace(pk=987654322, compute=self.compute) + self.first = SimpleNamespace(pk=987654321, uuid="wvc-test-uuid-1", compute=self.compute) + self.second = SimpleNamespace(pk=987654322, uuid="wvc-test-uuid-2", compute=self.compute) - def test_sibling_vm_can_run_while_same_vm_and_reconcile_are_excluded(self): + def test_sibling_vm_can_run_while_same_vm_and_vm_set_changes_are_refused(self): acquired = threading.Event() release = threading.Event() errors = [] @@ -21,37 +23,39 @@ class OperationLocksTestCase(SimpleTestCase): try: with libvirt_instance_lock(self.first): acquired.set() - if not release.wait(5): - raise TimeoutError("Test did not release the VM lock") + if not release.wait(10): + raise AssertionError("Test did not release the VM lock") except Exception as err: errors.append(err) + finally: + connection.close() worker = threading.Thread(target=hold_first_vm) worker.start() try: - self.assertTrue(acquired.wait(2)) - with libvirt_instance_lock(self.second, timeout=0.1): + self.assertTrue(acquired.wait(10)) + with libvirt_instance_lock(self.second): pass - with self.assertRaises(TimeoutError): - with libvirt_instance_lock(self.first, timeout=0.05): + with self.assertRaises(claims.Busy): + with libvirt_instance_lock(self.first): pass - with self.assertRaises(TimeoutError): - with libvirt_compute_lock(self.compute, timeout=0.05): + with self.assertRaises(claims.Busy): + with libvirt_compute_lock(self.compute): pass finally: release.set() - worker.join(5) + worker.join(10) self.assertFalse(worker.is_alive()) self.assertEqual(errors, []) - with libvirt_compute_lock(self.compute, timeout=0.1): + with libvirt_compute_lock(self.compute): pass - def test_instance_lock_is_reentrant_without_releasing_the_outer_lock(self): + def test_instance_lock_nests_and_cannot_be_upgraded(self): with libvirt_instance_lock(self.first): with libvirt_instance_lock(self.first): pass - with self.assertRaisesRegex(RuntimeError, "Cannot upgrade"): - with libvirt_compute_lock(self.compute, timeout=0.05): + with self.assertRaisesRegex(RuntimeError, "upgrade"): + with libvirt_compute_lock(self.compute): pass def test_compute_exclusive_owner_can_take_an_instance_lock(self): diff --git a/computes/tests/test_views.py b/computes/tests/test_views.py index 7d6b0683..ae8ab767 100644 --- a/computes/tests/test_views.py +++ b/computes/tests/test_views.py @@ -236,85 +236,144 @@ class ComputeConcurrencyTestCase(TransactionTestCase): self.assertEqual(exceptions, []) self.assertEqual(Instance.objects.filter(compute=compute).count(), 2) - def test_refresh_fails_closed_on_lock_error(self): - from unittest.mock import patch, MagicMock + def test_refresh_skips_a_busy_compute(self): + from unittest.mock import MagicMock from computes.utils import refresh_instance_database from instances.models import Instance + from operations import claims + from operations.models import Claim compute = Compute.objects.create( - name="lock-fail-test", - hostname="127.0.0.1", + name="lock-busy-test", + hostname="192.0.2.10", login="test", password="pwd", type=CONN_SOCKET, ) - Instance.objects.create(compute=compute, name="existing-vm", uuid="existing-uuid-1234") + Instance.objects.create(compute=compute, name="wvc-test-existing", uuid="existing-uuid-1234") - # Libvirt returns empty list - if it did not fail closed, existing-vm would be deleted! + # Libvirt returns an empty list: a refresh that ran would delete the VM mock_proxy = MagicMock() mock_proxy.wvm.listAllDomains.return_value = [] compute.proxy = mock_proxy - compute.status = True # reachable + compute.status = True - # 1. When lock directory cannot be acquired - with patch("computes.utils._get_lock_directory", return_value=None): - refresh_instance_database(compute) - self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + claims.admit({claims.compute_key(compute.id): Claim.EXCLUSIVE}, holder="other") + refresh_instance_database(compute) + self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + mock_proxy.wvm.listAllDomains.assert_not_called() - # 2. When lock file is a symlink - with patch("os.path.islink", return_value=True): - refresh_instance_database(compute) - self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + def _reconcile_compute(self, name, domains): + from unittest.mock import MagicMock - # 3. When lock file cannot be opened - with patch("os.open", side_effect=OSError("Permission denied")): - refresh_instance_database(compute) - self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + compute = Compute.objects.create(name=name, hostname="192.0.2.10", login="test", password="pwd", + type=CONN_SOCKET) + compute.proxy = MagicMock() + compute.proxy.wvm.listAllDomains.side_effect = domains + compute.status = True + return compute - # 4. When flock times out - curr_time = [0.0] - def advance_time(): - curr_time[0] += 20.0 - return curr_time[0] + def test_a_vm_operation_runs_while_a_reconcile_holds_the_compute(self): + import threading + from types import SimpleNamespace + from django.db import connection + from computes.utils import libvirt_instance_lock, refresh_instance_database - with patch("fcntl.flock", side_effect=BlockingIOError("Resource temporarily unavailable")): - with patch("time.monotonic", side_effect=advance_time): + outcome = [] + + def vm_operation(): + try: + with libvirt_instance_lock(SimpleNamespace(uuid="wvc-test-uuid-1", compute=compute)): + outcome.append("ran") + except Exception as err: + outcome.append(err) + finally: + connection.close() + + def list_domains(): + thread = threading.Thread(target=vm_operation) + thread.start() + thread.join(10) + return [] + + compute = self._reconcile_compute("reconcile-shared-test", list_domains) + refresh_instance_database(compute) + self.assertEqual(outcome, ["ran"]) + + def test_a_second_reconcile_of_the_compute_is_skipped_meanwhile(self): + """Two reconciles would read libvirt at different times and could + commit in reverse order, an older listing undoing a newer one.""" + import threading + from django.db import connection + from computes.utils import refresh_instance_database + + calls = [] + + def second_reconcile(): + try: refresh_instance_database(compute) - self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + finally: + connection.close() + + def list_domains(): + calls.append("list") + if len(calls) == 1: + thread = threading.Thread(target=second_reconcile) + thread.start() + thread.join(10) + return [] + + compute = self._reconcile_compute("reconcile-twice-test", list_domains) + refresh_instance_database(compute) + self.assertEqual(calls, ["list"]) + + def test_a_reconcile_runs_while_a_job_holds_the_compute_shared(self): + from computes.utils import refresh_instance_database + from instances.models import Instance + from operations import claims + from operations.models import Claim, Operation + + dom = type("Dom", (), {"UUIDString": lambda self: "33333333-4444-5555-6666-777777777777", + "name": lambda self: "wvc-test-vm"})() + compute = self._reconcile_compute("reconcile-job-test", lambda: [dom]) + op = Operation.objects.create(kind="test.job", username="wvc-test-user", submission_key="key-00000001", + state=Operation.RUNNING) + Claim.objects.create(operation=op, resource=claims.compute_key(compute.id), mode=Claim.SHARED) + refresh_instance_database(compute) + self.assertTrue(Instance.objects.filter(compute=compute, name="wvc-test-vm").exists()) + + def test_libvirt_compute_lock_held_by_another_holder_is_busy(self): + from computes.utils import libvirt_compute_lock + from operations import claims + from operations.models import Claim - def test_libvirt_compute_lock_thread_timeout(self): - from computes.utils import libvirt_compute_lock, _get_compute_thread_lock compute = Compute.objects.create( name="thread-lock-test", - hostname="127.0.0.1", + hostname="192.0.2.10", login="test", password="pwd", type=CONN_SOCKET, ) - tlock = _get_compute_thread_lock(compute.id) - tlock.acquire() - try: - with self.assertRaises(TimeoutError): - with libvirt_compute_lock(compute, timeout=0.05): - pass - finally: - tlock.release() + claims.admit({claims.compute_key(compute.id): Claim.EXCLUSIVE}, holder="other") + with self.assertRaises(claims.Busy): + with libvirt_compute_lock(compute): + pass def test_libvirt_compute_lock_reentrant(self): from computes.utils import libvirt_compute_lock compute = Compute.objects.create( name="reentrant-lock-test", - hostname="127.0.0.1", + hostname="192.0.2.10", login="test", password="pwd", type=CONN_SOCKET, ) executed = [] - with libvirt_compute_lock(compute, timeout=1.0): + with libvirt_compute_lock(compute): executed.append("outer") - with libvirt_compute_lock(compute, timeout=1.0): + with libvirt_compute_lock(compute): executed.append("inner") - with libvirt_compute_lock(compute, timeout=1.0): + with libvirt_compute_lock(compute): executed.append("deep") self.assertEqual(executed, ["outer", "inner", "deep"]) diff --git a/computes/utils.py b/computes/utils.py index 557dccf1..d5f2e10a 100644 --- a/computes/utils.py +++ b/computes/utils.py @@ -1,11 +1,5 @@ from contextlib import contextmanager -import fcntl import logging -import os -import stat -import tempfile -import threading -import time from datetime import timedelta from accounts.models import UserInstance @@ -15,142 +9,40 @@ from django.db import transaction from django.utils import timezone from computes.models import Compute from instances.models import Instance, InstanceTombstone +from operations import claims +from operations.models import Claim logger = logging.getLogger(__name__) -_COMPUTE_THREAD_LOCKS = {} -_COMPUTE_THREAD_LOCKS_GUARD = threading.Lock() - - -def _get_compute_thread_lock(compute_pk): - with _COMPUTE_THREAD_LOCKS_GUARD: - if compute_pk not in _COMPUTE_THREAD_LOCKS: - _COMPUTE_THREAD_LOCKS[compute_pk] = threading.Lock() - return _COMPUTE_THREAD_LOCKS[compute_pk] - - -def _get_lock_directory(): - candidates = [ - "/srv/webvirtcloud/data/locks", - os.path.join(tempfile.gettempdir(), f"webvirtcloud_locks_{os.geteuid()}"), - ] - for candidate in candidates: - try: - if os.path.islink(candidate): - continue - if not os.path.exists(candidate): - os.makedirs(candidate, mode=0o700, exist_ok=True) - if not os.path.isdir(candidate) or os.path.islink(candidate): - continue - - stat_info = os.stat(candidate) - # Ensure candidate directory is owned by the current process user - if stat_info.st_uid != os.geteuid(): - continue - # Ensure candidate directory has private permissions (mode 0700) - if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0: - try: - os.chmod(candidate, 0o700) - except OSError: - continue - stat_info = os.stat(candidate) - if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0: - continue - - test_file = os.path.join(candidate, f".test_write_{os.getpid()}_{threading.get_ident()}") - with open(test_file, "w") as tf: - tf.write("1") - os.unlink(test_file) - return candidate - except (OSError, PermissionError): - continue - return None - - -_thread_local = threading.local() - - -def _get_thread_held_locks(): - if not hasattr(_thread_local, "compute_locks"): - _thread_local.compute_locks = {} - return _thread_local.compute_locks - @contextmanager -def _libvirt_lock(key, filename, timeout=15.0, shared=False): - """Use flock for both threads and processes; exclusive locks also use a mutex.""" - held_locks = _get_thread_held_locks() - if key in held_locks: - if held_locks[key] and not shared: - raise RuntimeError("Cannot upgrade a shared compute lock inside a VM operation") - yield - return - - thread_lock = None if shared else _get_compute_thread_lock(key) - start_time = time.monotonic() - if thread_lock is not None and not thread_lock.acquire(timeout=timeout): - raise TimeoutError(f"Timeout waiting for lock on {key}") - try: - lock_dir = _get_lock_directory() - if not lock_dir: - raise OSError(f"Could not acquire secure lock directory for {key}") - path = os.path.join(lock_dir, filename) - if os.path.islink(path): - raise PermissionError(f"Lock file {path} is a symlink") - fd = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) - acquired = False - try: - mode = fcntl.LOCK_SH if shared else fcntl.LOCK_EX - while True: - try: - fcntl.flock(fd, mode | fcntl.LOCK_NB) - acquired = True - break - except BlockingIOError: - if time.monotonic() - start_time >= timeout: - raise TimeoutError(f"Timeout waiting for lock on {key}") - time.sleep(0.05) - held_locks[key] = shared - try: - yield - finally: - held_locks.pop(key, None) - finally: - try: - if acquired: - fcntl.flock(fd, fcntl.LOCK_UN) - finally: - os.close(fd) - finally: - if thread_lock is not None: - thread_lock.release() - - -@contextmanager -def libvirt_compute_lock(compute, timeout=15.0, *, shared=False): - """Exclude reconciliation/migration while allowing independent VM operations.""" +def libvirt_compute_lock(compute, *, shared=False): + """The compute for the block: exclusive keeps out every VM operation and + reconcile on it; shared lets sibling VMs and reconciles go on. Busy if it is taken.""" compute_pk = getattr(compute, "pk", getattr(compute, "id", None)) if not compute_pk: yield return - with _libvirt_lock(compute_pk, f"sync_compute_{compute_pk}.lock", timeout, shared): + with claims.hold({claims.compute_key(compute_pk): Claim.SHARED if shared else Claim.EXCLUSIVE}): yield @contextmanager -def user_quota_lock(user, timeout=15.0): - """One user's quota check and the change it allows run one at a time, on every compute. - Take it before any compute or VM lock.""" - with _libvirt_lock(("quota", user.pk), f"quota_user_{user.pk}.lock", timeout): +def user_quota_lock(user): + """One user's quota check and the change it allows run one at a time.""" + with claims.hold({claims.quota_key(user.pk): Claim.EXCLUSIVE}): yield @contextmanager -def libvirt_instance_lock(instance, timeout=15.0): - """Serialize one VM and share the compute barrier with other VM operations.""" - with libvirt_compute_lock(instance.compute, timeout, shared=True): - with _libvirt_lock(("instance", instance.pk), f"instance_{instance.pk}.lock", timeout): - yield +def libvirt_instance_lock(instance): + """One VM for the block, sharing its compute with sibling VMs.""" + compute = instance.compute + with claims.hold({ + claims.compute_key(getattr(compute, "pk", getattr(compute, "id", None))): Claim.SHARED, + claims.instance_key(instance.uuid): Claim.EXCLUSIVE, + }): + yield def _move_ownership(source, target): @@ -219,11 +111,16 @@ def refresh_instance_database(compute): return try: - with libvirt_compute_lock(compute, timeout=15.0): + # The compute shared: VM operations and jobs on it go on; changes of + # its VM set (create, clone, delete, migration, define) hold it + # exclusive and stay out until the end. One reconcile of a compute at + # a time: two would read libvirt at different times and could commit + # in reverse order, an older listing undoing a newer one. + with claims.hold({claims.compute_key(compute_pk): Claim.SHARED, + claims.reconcile_key(compute_pk): Claim.EXCLUSIVE}): # libvirt is asked before the transaction: a silent host can keep a # call waiting for 30 s, and on SQLite an open transaction makes every - # other write of the app fail meanwhile. The compute lock still keeps - # create/delete/migration of this compute out until the end. + # other write of the app fail meanwhile. if hasattr(compute, "status") and compute.status is False: return @@ -294,6 +191,6 @@ def refresh_instance_database(compute): ) if created and not Instance.objects.filter(uuid=uuid).exclude(compute=compute).exists(): _restore(inst) - except (OSError, PermissionError, TimeoutError) as e: - logger.error("Could not synchronize instances for compute %s: %s; failing closed", compute_pk, e) + except claims.Busy: + logger.debug("Compute %s is busy; not synchronized now", compute_pk) return diff --git a/instances/api/viewsets.py b/instances/api/viewsets.py index 8d53d3fb..aab52c88 100644 --- a/instances/api/viewsets.py +++ b/instances/api/viewsets.py @@ -311,8 +311,9 @@ class CreateInstanceViewSet(viewsets.ViewSet): raise return Response({"status": str(err)}, status=status.HTTP_400_BAD_REQUEST) # After the definition: the VM uses the volume, so no rollback here. - with utils.libvirt_compute_lock(compute): - utils.refresh_instance_database(compute) + # The refresh claims the compute itself and skips it when busy; the + # VM exists either way, so a busy compute must not refuse the request. + utils.refresh_instance_database(compute) msg = f"Instance {serializer.validated_data['name']} is created" return Response({"status": msg}) else: diff --git a/instances/tests/test_clone_permission.py b/instances/tests/test_clone_permission.py index 5007f422..f49b7cb4 100644 --- a/instances/tests/test_clone_permission.py +++ b/instances/tests/test_clone_permission.py @@ -176,7 +176,7 @@ class ClonePermissionTestCase(TestCase): calls = [] @contextmanager - def record(compute, timeout=15.0, *, shared=False): + def record(compute, *, shared=False): calls.append(shared) yield diff --git a/instances/tests/test_destroy.py b/instances/tests/test_destroy.py index 1afd3505..cc90627b 100644 --- a/instances/tests/test_destroy.py +++ b/instances/tests/test_destroy.py @@ -58,7 +58,7 @@ class DestroyOrderTestCase(TestCase): calls = [] @contextmanager - def record(compute, timeout=15.0, *, shared=False): + def record(compute, *, shared=False): calls.append(shared) yield @@ -86,7 +86,7 @@ class DestroyOrderTestCase(TestCase): calls = [] @contextmanager - def record(compute, timeout=15.0, *, shared=False): + def record(compute, *, shared=False): calls.append(shared) yield diff --git a/instances/tests/test_moved_while_waiting.py b/instances/tests/test_moved_while_waiting.py index 12151ca4..ebd413b3 100644 --- a/instances/tests/test_moved_while_waiting.py +++ b/instances/tests/test_moved_while_waiting.py @@ -1,9 +1,8 @@ """ -A VM operation waits for the VM's compute lock while a migration holds it. -When the migration moves the VM to another compute meanwhile, the waiting -operation holds the old compute's lock, not the new one's: it must not run -on the new host but answer "busy, retry", and the retry locks the right -compute. +A migration may move a VM between a request reading it and taking its +compute's claim: the request then holds the old compute's claim, not the +new one's. It must not run on the new host but answer "moved, try again", +and the retry claims the right compute. """ from contextlib import contextmanager from unittest.mock import PropertyMock, patch @@ -35,7 +34,7 @@ class MovedWhileWaitingTestCase(TestCase): def test_a_vm_moved_while_the_request_waited_is_not_touched(self): @contextmanager - def lock_after_a_migration(instance, timeout=15.0): + def lock_after_a_migration(instance): # the migration that held the lock moved the VM before releasing it Instance.objects.filter(pk=instance.pk).update(compute=self.target) yield @@ -44,11 +43,11 @@ class MovedWhileWaitingTestCase(TestCase): self.assertFalse(touched) self.assertEqual(response.status_code, 302) messages = [str(m) for m in response.wsgi_request._messages] - self.assertIn("Instance operation is busy. Please retry.", messages) + self.assertIn("The VM was moved meanwhile; try again.", messages) def test_a_vm_that_stayed_is_changed(self): @contextmanager - def plain_lock(instance, timeout=15.0): + def plain_lock(instance): yield response, touched = self.poweroff(plain_lock) @@ -58,7 +57,7 @@ class MovedWhileWaitingTestCase(TestCase): # destroy holds the whole compute; the migration moved the VM before # releasing it @contextmanager - def compute_lock_after_a_migration(compute, timeout=15.0, *, shared=False): + def compute_lock_after_a_migration(compute, *, shared=False): Instance.objects.filter(pk=self.vm.pk).update(compute=self.target) yield @@ -69,11 +68,11 @@ class MovedWhileWaitingTestCase(TestCase): self.assertFalse(wvm.return_value.delete.called) self.assertTrue(Instance.objects.filter(pk=self.vm.pk).exists()) messages = [str(m) for m in response.wsgi_request._messages] - self.assertIn("Instance operation is busy. Please retry.", messages) + self.assertIn("The VM was moved meanwhile; try again.", messages) def clone(self, user, change): @contextmanager - def compute_lock_meanwhile(compute, timeout=15.0, *, shared=False): + def compute_lock_meanwhile(compute, *, shared=False): change() yield @@ -101,7 +100,7 @@ class MovedWhileWaitingTestCase(TestCase): owner, lambda: Instance.objects.filter(pk=self.vm.pk).update(compute=self.target) ) self.assertFalse(cloned) - self.assertIn("Instance operation is busy. Please retry.", [str(m) for m in response.wsgi_request._messages]) + self.assertIn("The VM was moved meanwhile; try again.", [str(m) for m in response.wsgi_request._messages]) def test_a_template_unmarked_while_the_request_waited_is_not_cloned_by_a_viewer(self): from django.contrib.auth.models import Permission @@ -128,7 +127,7 @@ class MovedWhileWaitingTestCase(TestCase): wvm.return_value.get_memory.return_value = 512 @contextmanager - def compute_lock_after_a_resize(compute, timeout=15.0, *, shared=False): + def compute_lock_after_a_resize(compute, *, shared=False): wvm.return_value.get_vcpu.return_value = 8 # resized meanwhile yield diff --git a/instances/tests/test_security_mutations.py b/instances/tests/test_security_mutations.py index 47fc76fd..f3977234 100644 --- a/instances/tests/test_security_mutations.py +++ b/instances/tests/test_security_mutations.py @@ -3,12 +3,14 @@ from django.core.exceptions import PermissionDenied from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.http import Http404 -from django.test import TestCase +from django.db import connection +from django.test import TestCase, TransactionTestCase from django.urls import reverse from accounts.models import UserInstance from computes.models import Compute from instances.models import Instance +from operations import claims from instances.views import get_instance @@ -214,89 +216,6 @@ class InstanceSecurityMutationsTestCase(TestCase): res = self.client.get(reverse(f"instances:{ep}", args=[self.instance.id])) self.assertEqual(res.status_code, 405, f"{ep} allowed GET request instead of 405") - def test_opposite_direction_migrations_no_deadlock(self): - import threading - import time - from instances.utils import migrate_instance - - c1 = MagicMock() - c1.id = 101 - c1.pk = 101 - c1.hostname = "127.0.0.1" - c1.type = 1 - c1.login = "root" - c1.password = "" - - c2 = MagicMock() - c2.id = 102 - c2.pk = 102 - c2.hostname = "127.0.0.2" - c2.type = 1 - c2.login = "root" - c2.password = "" - - inst1 = MagicMock() - inst1.id = "inst1" - inst1.compute = c1 - inst1.name = "inst1" - inst1.autostart = False - inst1.proxy = MagicMock() - inst1.uuid = "u1" - - inst2 = MagicMock() - inst2.id = "inst2" - inst2.compute = c2 - inst2.name = "inst2" - inst2.autostart = False - inst2.proxy = MagicMock() - inst2.uuid = "u2" - - barrier = threading.Barrier(2) - errors = [] - - def worker_1_to_2(): - try: - barrier.wait(timeout=5) - migrate_instance(c2, inst1, self.superuser) - except Exception as e: - errors.append(("worker_1_to_2", e)) - - def worker_2_to_1(): - try: - barrier.wait(timeout=5) - migrate_instance(c1, inst2, self.superuser) - except Exception as e: - errors.append(("worker_2_to_1", e)) - - def slow_moveto(*args, **kwargs): - time.sleep(0.05) - - with patch("instances.utils.wvmInstances") as mock_instances_cls, \ - patch("instances.utils.wvmInstance"), \ - patch("instances.utils.transaction.atomic"), \ - patch("instances.utils.addlogmsg"), \ - patch("instances.utils.Instance") as mock_instance_model: - - mock_instances_cls.return_value.moveto.side_effect = slow_moveto - # under the locks each VM is still on its compute; the - # destination has no record of it - def rows(**query): - found = MagicMock(compute_id={"inst1": 101, "inst2": 102}[query["id"]]) if "id" in query else None - return MagicMock(**{"first.return_value": found}) - - mock_instance_model.objects.filter.side_effect = rows - - t1 = threading.Thread(target=worker_1_to_2) - t2 = threading.Thread(target=worker_2_to_1) - t1.start() - t2.start() - t1.join(timeout=10) - t2.join(timeout=10) - - self.assertFalse(t1.is_alive(), "Worker 1 to 2 timed out / deadlocked") - self.assertFalse(t2.is_alive(), "Worker 2 to 1 timed out / deadlocked") - self.assertEqual(errors, [], f"Unexpected errors during concurrent migrations: {errors}") - def test_resize_disk_empty_selection_warns_cleanly(self): self.client.force_login(self.change_user) with patch("instances.models.wvmInstance") as mock_wvm_cls: @@ -411,9 +330,112 @@ class InstanceSecurityMutationsTestCase(TestCase): def test_busy_instance_redirects_without_executing_the_operation(self): self.client.force_login(self.owner) - with patch("instances.views.libvirt_instance_lock", side_effect=TimeoutError("busy")), patch( + with patch("instances.views.libvirt_instance_lock", side_effect=claims.Busy("instance:wvc-test")), patch( "instances.models.wvmInstance" ) as proxy: response = self.client.post(reverse("instances:poweron", args=[self.instance.pk]), {}) self.assertRedirects(response, reverse("instances:instance", args=[self.instance.pk]), fetch_redirect_response=False) proxy.assert_not_called() + + +class OppositeMigrationsTestCase(TransactionTestCase): + """Two migrations between the same computes in opposite directions: a + claim is refused at once instead of waited for, so one may be refused but + none deadlocks.""" + + def setUp(self): + self.superuser = get_user_model().objects.create_superuser("migrate_admin", "a@example.com", "x") + + def test_opposite_direction_migrations_never_deadlock(self): + import threading + import time + from instances.utils import migrate_instance + + c1 = MagicMock() + c1.id = 101 + c1.pk = 101 + c1.hostname = "192.0.2.1" + c1.type = 1 + c1.login = "root" + c1.password = "" + + c2 = MagicMock() + c2.id = 102 + c2.pk = 102 + c2.hostname = "192.0.2.2" + c2.type = 1 + c2.login = "root" + c2.password = "" + + inst1 = MagicMock() + inst1.id = "inst1" + inst1.compute = c1 + inst1.name = "inst1" + inst1.autostart = False + inst1.proxy = MagicMock() + inst1.uuid = "u1" + + inst2 = MagicMock() + inst2.id = "inst2" + inst2.compute = c2 + inst2.name = "inst2" + inst2.autostart = False + inst2.proxy = MagicMock() + inst2.uuid = "u2" + + barrier = threading.Barrier(2) + errors, refused = [], [] + + def worker_1_to_2(): + try: + barrier.wait(timeout=5) + migrate_instance(c2, inst1, self.superuser) + except claims.Busy: + refused.append("worker_1_to_2") + except Exception as e: + errors.append(("worker_1_to_2", e)) + finally: + connection.close() + + def worker_2_to_1(): + try: + barrier.wait(timeout=5) + migrate_instance(c1, inst2, self.superuser) + except claims.Busy: + refused.append("worker_2_to_1") + except Exception as e: + errors.append(("worker_2_to_1", e)) + finally: + connection.close() + + def slow_moveto(*args, **kwargs): + time.sleep(0.05) + + with patch("instances.utils.wvmInstances") as mock_instances_cls, \ + patch("instances.utils.wvmInstance"), \ + patch("instances.utils.record_migration"), \ + patch("instances.utils.addlogmsg"), \ + patch("instances.utils.Instance") as mock_instance_model: + + mock_instances_cls.return_value.moveto.side_effect = slow_moveto + # under the locks each VM is still on its compute; the + # destination has no record of it + def rows(**query): + found = MagicMock(compute_id={"inst1": 101, "inst2": 102}[query["id"]]) if "id" in query else None + return MagicMock(**{"first.return_value": found}) + + mock_instance_model.objects.filter.side_effect = rows + + t1 = threading.Thread(target=worker_1_to_2) + t2 = threading.Thread(target=worker_2_to_1) + t1.start() + t2.start() + t1.join(timeout=10) + t2.join(timeout=10) + + self.assertFalse(t1.is_alive(), "Worker 1 to 2 timed out / deadlocked") + self.assertFalse(t2.is_alive(), "Worker 2 to 1 timed out / deadlocked") + self.assertEqual(errors, [], f"Unexpected errors during concurrent migrations: {errors}") + # a refusal is immediate; at least one of them migrates + self.assertLess(len(refused), 2) + self.assertEqual(mock_instances_cls.return_value.moveto.call_count, 2 - len(refused)) diff --git a/instances/views.py b/instances/views.py index 02f85686..97c8762c 100755 --- a/instances/views.py +++ b/instances/views.py @@ -38,6 +38,8 @@ from libvirt import (VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, VIR_DOMAIN_UNDEFINE_NVRAM, libvirtError) from logs.views import addlogmsg +from operations import claims +from operations.utils import busy_message, summary from webvirtcloud.middleware import error_text from vrtManager import util from vrtManager.create import wvmCreate @@ -446,16 +448,25 @@ def get_instance(user, pk, perm_type="view"): return instance -def _busy(request, pk): +def _busy(request, pk, busy): if hasattr(request, "accepted_renderer"): - return JsonResponse({"detail": "Instance operation is busy. Please retry."}, status=409) - messages.error(request, _("Instance operation is busy. Please retry.")) + blocker = busy.blocker + return JsonResponse({"detail": busy_message(busy, request.user), + "operation": summary(blocker, request.user) if blocker else None}, status=409) + messages.error(request, busy_message(busy, request.user)) + return get_safe_redirect(request, default=reverse("instances:instance", args=[pk])) + + +def _moved(request, pk): + if hasattr(request, "accepted_renderer"): + return JsonResponse({"detail": "The VM was moved meanwhile; try again."}, status=409) + messages.error(request, _("The VM was moved meanwhile; try again.")) return get_safe_redirect(request, default=reverse("instances:instance", args=[pk])) def _moved_meanwhile(pk, inst): - """A migration may have moved the VM while the request waited for its - compute's lock: the lock held is then the old compute's. The retry locks + """A migration may have moved the VM before the request took its + compute's claim: the claim held is then the old compute's. The retry locks the new one (or finds the record merged away).""" return not Instance.objects.filter(pk=pk, compute_id=inst.compute_id).exists() @@ -468,10 +479,10 @@ def serialize_instance_mutation(func): try: with libvirt_instance_lock(inst): if _moved_meanwhile(pk, inst): - return _busy(request, pk) + return _moved(request, pk) return func(request, pk, *args, **kwargs) - except TimeoutError: - return _busy(request, pk) + except claims.Busy as busy: + return _busy(request, pk, busy) return wrapper @@ -494,8 +505,8 @@ def serialize_user_quota(func): try: with user_quota_lock(request.user): return func(request, pk, *args, **kwargs) - except TimeoutError: - return _busy(request, pk) + except claims.Busy as busy: + return _busy(request, pk, busy) return wrapper @@ -510,10 +521,10 @@ def serialize_compute_mutation(func): try: with libvirt_compute_lock(inst.compute): if _moved_meanwhile(pk, inst): - return _busy(request, pk) + return _moved(request, pk) return func(request, pk, *args, **kwargs) - except TimeoutError: - return _busy(request, pk) + except claims.Busy as busy: + return _busy(request, pk, busy) return wrapper @@ -728,7 +739,9 @@ def migrate(request, pk): try: # migrate_instance logs the outcome utils.migrate_instance(target_host, instance, request.user, live, unsafe, offline, autoconverge, compress) - except (libvirtError, OSError) as err: # OSError includes a lock timeout + except claims.Busy as busy: + return _busy(request, instance.id, busy) + except (libvirtError, OSError) as err: messages.error(request, err) return get_safe_redirect( @@ -1945,11 +1958,11 @@ def clone(request, pk): # The whole compute: the destination name and disk names must not be # taken by a concurrent clone of another VM while volumes are copied. with libvirt_compute_lock(instance.compute): - # While this request waited, a migration may have moved the + # Before this request took the claim, a migration may have moved the # source, or its template mark, size or the user's rights changed: # the clone is checked against the source as it is now. if _moved_meanwhile(pk, instance): - return _busy(request, pk) + return _moved(request, pk) instance = _clone_source(request.user, pk) clone_data, refusal = _clone_plan(request, instance) if refusal: @@ -1978,7 +1991,9 @@ def clone(request, pk): ) return redirect(reverse("instances:instance", args=[new_instance.id])) - except (libvirtError, ValueError, OSError) as e: # OSError includes a lock timeout + except claims.Busy as busy: + return _busy(request, pk, busy) + except (libvirtError, ValueError, OSError) as e: messages.error(request, error_text(request, e)) return get_safe_redirect(request, default=reverse("instances:instance", args=[pk])) diff --git a/operations/claims.py b/operations/claims.py index 00a98d76..2488f667 100644 --- a/operations/claims.py +++ b/operations/claims.py @@ -5,6 +5,7 @@ claim conflicts; otherwise it is refused at once (Busy).""" import os import socket import threading +from contextlib import contextmanager from datetime import timedelta from django.db import connection, transaction @@ -31,6 +32,10 @@ def compute_key(compute_id): return f"compute:{compute_id}" +def reconcile_key(compute_id): + return f"reconcile:{compute_id}" + + def instance_key(uuid): return f"instance:{uuid}" @@ -90,3 +95,51 @@ def admit_job(op, wanted, existing, extra_locks=()): op.save() Claim.objects.bulk_create([Claim(operation=op, resource=k, mode=m) for k, m in sorted(wanted.items())]) return op + + +_local = threading.local() + + +def _held(): + if not hasattr(_local, "held"): + _local.held = {} + return _local.held + + +@contextmanager +def hold(wanted): + """Transient claims for the block. Keys this thread holds are reused; + exclusive covers shared; shared cannot become exclusive inside its block.""" + held = _held() + new = {} + for key, mode in wanted.items(): + if key in held: + if held[key] == Claim.SHARED and mode == Claim.EXCLUSIVE: + raise RuntimeError(f"Cannot upgrade a shared claim on {key}") + continue + new[key] = mode + if not new: + yield + return + ids = admit(new, holder=holder_id()) + held.update(new) + try: + yield + finally: + for key in new: + held.pop(key, None) + Claim.objects.filter(pk__in=ids).delete() + + +@contextmanager +def adopted(op): + """The job thread holds its job's claims, so code it calls that takes a + lock helper does not refuse its own job.""" + held = _held() + own = dict(Claim.objects.filter(operation=op).values_list("resource", "mode")) + held.update(own) + try: + yield + finally: + for key in own: + held.pop(key, None) diff --git a/operations/jobs.py b/operations/jobs.py new file mode 100644 index 00000000..fd822fbc --- /dev/null +++ b/operations/jobs.py @@ -0,0 +1,12 @@ +"""Kinds of jobs by name (filled by each app's jobs.py via register).""" + +KINDS = {} + + +def get_kind(name): + return KINDS[name] + + +def label_of(name): + kind = KINDS.get(name) + return kind.label if kind is not None else name diff --git a/operations/tests/test_hold.py b/operations/tests/test_hold.py new file mode 100644 index 00000000..696a25d9 --- /dev/null +++ b/operations/tests/test_hold.py @@ -0,0 +1,68 @@ +"""Short operations hold transient claims for their duration; a thread may +nest them; a job thread holds its job's claims.""" + +import threading + +from django.contrib.auth import get_user_model +from django.db import connection +from django.test import TransactionTestCase + +from operations import claims +from operations.models import Claim, Operation + +S, X = Claim.SHARED, Claim.EXCLUSIVE + + +class HoldTestCase(TransactionTestCase): + def test_claims_live_for_the_block(self): + with claims.hold({"instance:u1": X}): + self.assertTrue(Claim.objects.filter(resource="instance:u1").exists()) + self.assertFalse(Claim.objects.exists()) + + def test_another_thread_is_refused_at_once(self): + inside, release, seen = threading.Event(), threading.Event(), [] + + def holder(): + try: + with claims.hold({"instance:u1": X}): + inside.set() + release.wait(10) + finally: + connection.close() + + t = threading.Thread(target=holder) + t.start() + self.assertTrue(inside.wait(10)) + try: + with claims.hold({"instance:u1": X}): + pass + except claims.Busy as busy: + seen.append(busy.blocker) + finally: + release.set() + t.join(10) + self.assertEqual(seen, [None]) + + def test_nesting_in_one_thread_reuses_held_claims(self): + with claims.hold({"compute:1": X}): + with claims.hold({"compute:1": S, "instance:u1": X}): + self.assertEqual(Claim.objects.count(), 2) + self.assertEqual(Claim.objects.count(), 1) + + def test_a_shared_claim_cannot_be_upgraded(self): + with claims.hold({"compute:1": S}): + with self.assertRaisesRegex(RuntimeError, "upgrade"): + with claims.hold({"compute:1": X}): + pass + + def test_a_job_thread_holds_its_own_claims(self): + user = get_user_model().objects.create_user("hold_user", "h@example.com", "pw") + op = Operation(kind="test.noop", user=user, username="hold_user", submission_key="k", + resource="instance:u1", resource_label="c/vm") + claims.admit_job(op, {"compute:1": S, "instance:u1": X}, existing=lambda: None) + with claims.adopted(op): + with claims.hold({"compute:1": S, "instance:u1": X}): + pass + with self.assertRaises(claims.Busy): + with claims.hold({"instance:u1": X}): + pass diff --git a/operations/utils.py b/operations/utils.py index ef67fb23..e781ec64 100644 --- a/operations/utils.py +++ b/operations/utils.py @@ -1,7 +1,15 @@ -"""Secrets never reach a job record, a response or a log line.""" +"""Secrets never reach a job record, a response or a log line. What a refused +request says about the job holding the resource: details for its owner and +superusers, a minimal summary for everyone else.""" import re +from django.utils import timezone +from django.utils.timesince import timesince +from django.utils.translation import gettext as _ + +from . import jobs + _NAMES = r"(?:pass|secret|token|key)\w*" _PATTERNS = [ (re.compile(r"(://[^:/@\s]+:)[^@\s]+@"), r"\1***@"), @@ -30,3 +38,25 @@ def check_params(params): elif isinstance(params, (list, tuple)): for value in params: check_params(value) + + +def may_see(op, user): + return user.is_superuser or (op.user_id is not None and op.user_id == user.pk) + + +def summary(op, user): + base = {"kind": op.kind, "label": str(_(jobs.label_of(op.kind))), "state": op.state, "since": op.created_at} + if may_see(op, user): + base.update({"id": str(op.id), "user": op.username}) + return base + + +def busy_message(busy, user): + op = busy.blocker + if op is None: + return _("A short operation is in progress on this resource; try again in a moment.") + values = {"label": _(jobs.label_of(op.kind)), "state": op.state, "since": timesince(op.created_at, timezone.now())} + if may_see(op, user): + return _("%(label)s is %(state)s on this resource (started by %(user)s, %(since)s ago); " + "see the Jobs page.") % {**values, "user": op.username} + return _("%(label)s is %(state)s on this resource (since %(since)s).") % values diff --git a/storages/tests/test_views.py b/storages/tests/test_views.py index e21226d6..5c8847f6 100644 --- a/storages/tests/test_views.py +++ b/storages/tests/test_views.py @@ -11,6 +11,7 @@ from django.urls import reverse from django.contrib.auth import get_user_model from computes.models import Compute +from operations import claims from storages.upload import ( handle_uploaded_file, _acquire_upload_lock, @@ -958,12 +959,12 @@ class VolumeDeleteTests(TestCase): def test_a_busy_compute_is_a_message(self, mock_wvm): conn = self.refused(mock_wvm) url = reverse("storage", args=[self.compute.id, "default"]) - with patch("storages.views.libvirt_compute_lock", side_effect=TimeoutError): + with patch("storages.views.libvirt_compute_lock", side_effect=claims.Busy("compute:1")): response = self.client.post(url, {"del_volume": "1", "volname": "disk.qcow2"}) self.assertRedirects(response, url, fetch_redirect_response=False) conn.del_volume.assert_not_called() messages = [str(m) for m in get_messages(response.wsgi_request)] - self.assertTrue(any("Please retry" in m for m in messages), messages) + self.assertTrue(any("short operation" in m for m in messages), messages) @patch("storages.api.viewsets.wvmStorage") def test_the_api_refuses_with_the_reason(self, mock_wvm): diff --git a/storages/views.py b/storages/views.py index e7085c16..13149440 100644 --- a/storages/views.py +++ b/storages/views.py @@ -11,6 +11,8 @@ from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.utils.translation import gettext_lazy as _ from libvirt import libvirtError +from operations import claims +from operations.utils import busy_message from vrtManager.connection import CONN_TCP, CONN_TLS from vrtManager.util import OperationError @@ -240,8 +242,8 @@ def storage(request, compute_id, pool): conn.del_volume(volname) except OperationError as err: messages.error(request, str(err)) - except TimeoutError: - messages.error(request, _("A VM operation on this host is running. Please retry.")) + except claims.Busy as busy: + messages.error(request, busy_message(busy, request.user)) else: messages.success( request, _("Volume: %(vol)s is deleted.") % {"vol": volname} diff --git a/webvirtcloud/middleware.py b/webvirtcloud/middleware.py index 8cd97b03..cfba43a0 100644 --- a/webvirtcloud/middleware.py +++ b/webvirtcloud/middleware.py @@ -1,9 +1,11 @@ from django.contrib import messages from django.http import JsonResponse -from django.shortcuts import render +from django.shortcuts import redirect, render from django.utils.translation import gettext_lazy as _ import libvirt from libvirt import libvirtError +from operations import claims +from operations.utils import busy_message, summary from vrtManager.util import ConnectionFailed API_PREFIX = "/api/" @@ -55,9 +57,17 @@ class ExceptionMiddleware: if request.path_info.startswith(API_PREFIX): if isinstance(exception, libvirtError): return JsonResponse({"detail": error_text(request, exception)}, status=400) + if isinstance(exception, claims.Busy): + return JsonResponse({"detail": busy_message(exception, request.user), + "operation": summary(exception.blocker, request.user) if exception.blocker else None}, + status=409) if isinstance(exception, TimeoutError): return JsonResponse({"detail": "Operation is busy. Please retry."}, status=409) return None + if isinstance(exception, claims.Busy) and request.method not in ("GET", "HEAD"): + # back to the form with the reason; a GET is never redirected to itself + messages.error(request, busy_message(exception, request.user)) + return redirect(request.get_full_path()) if isinstance(exception, libvirtError): messages.error( request, diff --git a/webvirtcloud/tests/test_busy_refusal.py b/webvirtcloud/tests/test_busy_refusal.py new file mode 100644 index 00000000..2d1fe091 --- /dev/null +++ b/webvirtcloud/tests/test_busy_refusal.py @@ -0,0 +1,52 @@ +"""A request refused because its resource is held answers at once: the API +with 409 and the job holding it, a page with the reason on the form it came +from. Only the job's owner and superusers see who started it.""" + +import json +from unittest.mock import MagicMock + +from django.contrib.auth import get_user_model +from django.contrib.messages import get_messages +from django.contrib.messages.storage.fallback import FallbackStorage +from django.contrib.sessions.middleware import SessionMiddleware +from django.test import RequestFactory, TestCase + +from operations import claims +from operations.models import Operation +from operations.utils import busy_message +from webvirtcloud.middleware import ExceptionMiddleware + + +class BusyRefusalTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user("busy-owner", password="pw") + self.other = User.objects.create_user("busy-other", password="pw") + self.job = Operation.objects.create(kind="test.noop", user=self.owner, username="busy-owner", + submission_key="k", resource="instance:u1", resource_label="c/vm") + + def handle(self, path, busy): + request = RequestFactory().post(path) + SessionMiddleware(lambda r: None).process_request(request) + request._messages = FallbackStorage(request) + request.user = self.owner + response = ExceptionMiddleware(MagicMock()).process_exception(request, busy) + return response, [str(m) for m in get_messages(request)] + + def test_the_api_answers_409_with_the_job(self): + response, _ = self.handle("/api/v1/computes/1/instances/", claims.Busy("compute:1", self.job)) + self.assertEqual(response.status_code, 409) + body = json.loads(response.content) + self.assertIn("busy-owner", body["detail"]) + self.assertEqual(body["operation"]["id"], str(self.job.id)) + + def test_a_page_is_sent_back_to_its_form_with_the_reason(self): + response, messages = self.handle("/computes/1/create/", claims.Busy("compute:1")) + self.assertEqual(response.status_code, 302) + self.assertEqual(response.url, "/computes/1/create/") + self.assertTrue(any("short operation" in m for m in messages), messages) + + def test_only_the_owner_learns_who_started_the_job(self): + busy = claims.Busy("instance:u1", self.job) + self.assertIn("busy-owner", busy_message(busy, self.owner)) + self.assertNotIn("busy-owner", busy_message(busy, self.other))