j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
Code Issues Projects Releases Wiki Activity

refactor(computes): the lock helpers take claims instead of file locks

Browse source
This commit is contained in:
catborise 2026-10-13 17:45:42 +03:00
parent fca4a5d91c
commit 68812ddbdf
18 changed files with 545 additions and 318 deletions
Download patch file Download diff file Expand all files Collapse all files

6
computes/tests/test_api_security.py
View file

@ -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()

40
computes/tests/test_operation_locks.py
View file

@ -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):

143
computes/tests/test_views.py
View file

@ -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"])

157
computes/utils.py
View file

@ -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

5
instances/api/viewsets.py
View file

@ -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:

2
instances/tests/test_clone_permission.py
View file

@ -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

4
instances/tests/test_destroy.py
View file

@ -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

25
instances/tests/test_moved_while_waiting.py
View file

@ -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

192
instances/tests/test_security_mutations.py
View file

@ -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))

49
instances/views.py
View file

@ -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]))

53
operations/claims.py
View file

@ -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)

12
operations/jobs.py Normal file
View file

@ -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

68
operations/tests/test_hold.py Normal file
View file

@ -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

32
operations/utils.py
View file

@ -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

5
storages/tests/test_views.py
View file

@ -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):

6
storages/views.py
View file

@ -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}

12
webvirtcloud/middleware.py
View file

@ -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,

52
webvirtcloud/tests/test_busy_refusal.py Normal file
View file

@ -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))