diff --git a/computes/tests.py b/computes/tests.py index 0750134d..9a7b82bd 100644 --- a/computes/tests.py +++ b/computes/tests.py @@ -1,7 +1,7 @@ import os from django.core.exceptions import ObjectDoesNotExist from django.shortcuts import reverse -from django.test import TestCase +from django.test import TestCase, TransactionTestCase from vrtManager.connection import CONN_SOCKET, connection_manager from .models import Compute @@ -185,3 +185,133 @@ class ComputesTestCase(TestCase): ) ) self.assertEqual(response.status_code, 200) + + +class ComputeConcurrencyTestCase(TransactionTestCase): + def test_concurrent_refresh_instance_database(self): + import threading + from django.db import connection + from unittest.mock import MagicMock + from computes.utils import refresh_instance_database + from instances.models import Instance + + compute = Compute.objects.create( + name="concurrent-test", + hostname="127.0.0.1", + login="test", + password="pwd", + type=CONN_SOCKET, + ) + + mock_dom1 = MagicMock() + mock_dom1.UUIDString.return_value = "11111111-2222-3333-4444-555555555555" + mock_dom1.name.return_value = "vm-concurrent-1" + + mock_dom2 = MagicMock() + mock_dom2.UUIDString.return_value = "22222222-3333-4444-5555-666666666666" + mock_dom2.name.return_value = "vm-concurrent-2" + + mock_proxy = MagicMock() + mock_proxy.wvm.listAllDomains.return_value = [mock_dom1, mock_dom2] + compute.proxy = mock_proxy + + exceptions = [] + + def worker(): + try: + refresh_instance_database(compute) + except Exception as e: + exceptions.append(e) + finally: + connection.close() + + connection.close() + threads = [threading.Thread(target=worker) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + + 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 + from computes.utils import refresh_instance_database + from instances.models import Instance + + compute = Compute.objects.create( + name="lock-fail-test", + hostname="127.0.0.1", + login="test", + password="pwd", + type=CONN_SOCKET, + ) + Instance.objects.create(compute=compute, name="existing-vm", uuid="existing-uuid-1234") + + # Libvirt returns empty list - if it did not fail closed, existing-vm would be deleted! + mock_proxy = MagicMock() + mock_proxy.wvm.listAllDomains.return_value = [] + compute.proxy = mock_proxy + + # 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) + + # 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) + + # 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) + + # 4. When flock times out + curr_time = [0.0] + def advance_time(): + curr_time[0] += 20.0 + return curr_time[0] + + with patch("fcntl.flock", side_effect=BlockingIOError("Resource temporarily unavailable")): + with patch("time.time", side_effect=advance_time): + refresh_instance_database(compute) + self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) + + 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", + 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() + + 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", + login="test", + password="pwd", + type=CONN_SOCKET, + ) + executed = [] + with libvirt_compute_lock(compute, timeout=1.0): + executed.append("outer") + with libvirt_compute_lock(compute, timeout=1.0): + executed.append("inner") + with libvirt_compute_lock(compute, timeout=1.0): + executed.append("deep") + self.assertEqual(executed, ["outer", "inner", "deep"]) diff --git a/computes/utils.py b/computes/utils.py index a7075606..1fa40628 100644 --- a/computes/utils.py +++ b/computes/utils.py @@ -1,9 +1,156 @@ +from contextlib import contextmanager +import fcntl import logging +import os +import stat +import tempfile +import threading +import time from django.db import transaction +from computes.models import Compute from instances.models import Instance 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()}") + 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_compute_lock(compute, timeout=15.0): + """ + Context manager providing multi-thread and multi-process synchronization + for libvirt compute operations and instance database synchronization. + Supports re-entrant locking within the same thread. + Fails closed on lock directory or permission errors. + """ + compute_pk = getattr(compute, "pk", getattr(compute, "id", None)) + if not compute_pk: + yield + return + + held_locks = _get_thread_held_locks() + if compute_pk in held_locks: + held_locks[compute_pk] += 1 + try: + yield + finally: + held_locks[compute_pk] -= 1 + if held_locks[compute_pk] <= 0: + held_locks.pop(compute_pk, None) + return + + thread_lock = _get_compute_thread_lock(compute_pk) + start_time = time.time() + acquired_thread = thread_lock.acquire(timeout=timeout) + if not acquired_thread: + logger.error("Timeout waiting for in-process thread lock on compute %s; aborting to fail closed", compute_pk) + raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}") + + try: + lock_dir = _get_lock_directory() + if not lock_dir: + logger.error("Failed to acquire application lock directory for compute %s; aborting to fail closed", compute_pk) + raise OSError(f"Could not acquire secure lock directory for compute {compute_pk}") + + lock_file_path = os.path.join(lock_dir, f"sync_compute_{compute_pk}.lock") + if os.path.islink(lock_file_path): + logger.error("Lock file %s is a symlink! Aborting to fail closed", lock_file_path) + raise PermissionError(f"Compute lock file {lock_file_path} is a symlink") + + flags = os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0) + try: + lock_fd = os.open(lock_file_path, flags, 0o600) + except OSError as e: + logger.error("Could not open synchronization lock file for compute %s: %s; aborting to fail closed", compute_pk, e) + raise + + lock_acquired = False + try: + while time.time() - start_time < timeout: + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_acquired = True + break + except (BlockingIOError, OSError): + time.sleep(0.05) + + if not lock_acquired: + logger.error("Timeout waiting for cross-process synchronization lock on compute %s; aborting to fail closed", compute_pk) + raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}") + + held_locks[compute_pk] = 1 + try: + yield + finally: + held_locks.pop(compute_pk, None) + finally: + if lock_acquired: + try: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + except OSError: + pass + try: + os.close(lock_fd) + except OSError: + pass + finally: + thread_lock.release() + def refresh_instance_database(compute): """ @@ -16,49 +163,66 @@ def refresh_instance_database(compute): - If libvirt is unreachable (connection/network error), the database is left untouched to avoid false deletions. """ - try: - if hasattr(compute, "status") and compute.status is False: - return - domains = compute.proxy.wvm.listAllDomains() - except Exception as e: - logger.warning( - "Failed to retrieve domains from compute %s (%s): %s", - getattr(compute, "name", compute), - getattr(compute, "hostname", ""), - e, - ) + compute_pk = getattr(compute, "pk", getattr(compute, "id", None)) + if not compute_pk: return - # Map libvirt UUIDs to current domain names - host_domains = {} - for d in domains: - try: - host_domains[d.UUIDString()] = d.name() - except Exception as e: - logger.warning( - "Error reading domain metadata from compute %s: %s", - getattr(compute, "name", compute), - e, - ) - # If we fail to read a domain's metadata, abort sync to prevent accidental deletion - return + try: + with libvirt_compute_lock(compute, timeout=15.0): + with transaction.atomic(): + comp_locked = Compute.objects.select_for_update().filter(pk=compute_pk).first() + if not comp_locked: + return - db_instances = Instance.objects.filter(compute=compute) - db_uuids = set(db_instances.values_list("uuid", flat=True)) + if hasattr(compute, "status") and compute.status is False: + return - with transaction.atomic(): - # 1. Update name for any VM whose name changed out-of-band in virsh/virt-manager - for inst in db_instances: - if inst.uuid in host_domains: - new_name = host_domains[inst.uuid] - if inst.name != new_name: - inst.name = new_name - inst.save(update_fields=["name"]) + try: + domains = compute.proxy.wvm.listAllDomains() + except Exception as e: + logger.warning( + "Failed to retrieve domains from compute %s (%s): %s", + getattr(compute, "name", compute), + getattr(compute, "hostname", ""), + e, + ) + return - # 2. Delete instances that were genuinely undefined/deleted in virsh/virt-manager - db_instances.exclude(uuid__in=host_domains.keys()).delete() + # Map libvirt UUIDs to current domain names + host_domains = {} + for d in domains: + try: + host_domains[d.UUIDString()] = d.name() + except Exception as e: + logger.warning( + "Error reading domain metadata from compute %s: %s", + getattr(compute, "name", compute), + e, + ) + return - # 3. Create new instances that were created out-of-band in virsh/virt-manager - for uuid, name in host_domains.items(): - if uuid not in db_uuids: - Instance.objects.create(compute=compute, name=name, uuid=uuid) + db_instances = list(Instance.objects.filter(compute=compute).select_for_update()) + db_uuids = set(inst.uuid for inst in db_instances) + + # 1. Update name for any VM whose name changed out-of-band in virsh/virt-manager + for inst in db_instances: + if inst.uuid in host_domains: + new_name = host_domains[inst.uuid] + if inst.name != new_name: + inst.name = new_name + inst.save(update_fields=["name"]) + + # 2. Delete instances that were genuinely undefined/deleted in virsh/virt-manager + Instance.objects.filter(compute=compute).exclude(uuid__in=host_domains.keys()).delete() + + # 3. Create or update new instances that were created out-of-band in virsh/virt-manager + for uuid, name in host_domains.items(): + if uuid not in db_uuids: + Instance.objects.get_or_create( + compute=compute, + uuid=uuid, + defaults={"name": name}, + ) + except (OSError, PermissionError, TimeoutError) as e: + logger.error("Could not synchronize instances for compute %s: %s; failing closed", compute_pk, e) + return diff --git a/instances/api/viewsets.py b/instances/api/viewsets.py index cde1b9d2..569ceb67 100644 --- a/instances/api/viewsets.py +++ b/instances/api/viewsets.py @@ -285,26 +285,28 @@ class CreateInstanceViewSet(viewsets.ViewSet): machine = "q35" firmware["secure"] = "yes" - ret = conn.create_instance( - name=serializer.validated_data["name"], - memory=serializer.validated_data["memory"], - vcpu=serializer.validated_data["vcpu"], - vcpu_mode=serializer.validated_data["vcpu_mode"], - uuid=util.randomUUID(), - arch=arch, - machine=machine, - firmware=firmware, - volumes=volume_list, - networks=serializer.validated_data["networks"], - nwfilter=serializer.validated_data["nwfilter"], - graphics=serializer.validated_data["graphics"], - virtio=serializer.validated_data["virtio"], - listener_addr=serializer.validated_data["listener_addr"], - video=serializer.validated_data["video"], - console_pass=serializer.validated_data["console_pass"], - mac=serializer.validated_data["mac"], - qemu_ga=serializer.validated_data["qemu_ga"], - ) + with utils.libvirt_compute_lock(compute): + ret = conn.create_instance( + name=serializer.validated_data["name"], + memory=serializer.validated_data["memory"], + vcpu=serializer.validated_data["vcpu"], + vcpu_mode=serializer.validated_data["vcpu_mode"], + uuid=util.randomUUID(), + arch=arch, + machine=machine, + firmware=firmware, + volumes=volume_list, + networks=serializer.validated_data["networks"], + nwfilter=serializer.validated_data["nwfilter"], + graphics=serializer.validated_data["graphics"], + virtio=serializer.validated_data["virtio"], + listener_addr=serializer.validated_data["listener_addr"], + video=serializer.validated_data["video"], + console_pass=serializer.validated_data["console_pass"], + mac=serializer.validated_data["mac"], + qemu_ga=serializer.validated_data["qemu_ga"], + ) + utils.refresh_instance_database(compute) msg = f"Instance {serializer.validated_data['name']} is created" return Response({"status": msg}) else: diff --git a/instances/migrations/0012_instance_unique_compute_uuid.py b/instances/migrations/0012_instance_unique_compute_uuid.py new file mode 100644 index 00000000..9385d72b --- /dev/null +++ b/instances/migrations/0012_instance_unique_compute_uuid.py @@ -0,0 +1,94 @@ +# Generated by Django 4.2.30 on 2026-09-27 10:57 + +from django.db import migrations, models + + +def deduplicate_instances(apps, schema_editor): + Instance = apps.get_model('instances', 'Instance') + UserInstance = apps.get_model('accounts', 'UserInstance') + from django.db.models import Count, Q + import uuid as uuid_mod + + # Assign distinct UUIDs to legacy/corrupt records with null, empty, or whitespace UUIDs + # so they are not treated as duplicates or collapsed together + empty_uuid_instances = Instance.objects.filter( + Q(uuid__isnull=True) | Q(uuid="") | Q(uuid__regex=r"^\s*$") + ) + for inst in empty_uuid_instances: + inst.uuid = str(uuid_mod.uuid4()) + inst.save(update_fields=['uuid']) + + duplicates = ( + Instance.objects.exclude(uuid__isnull=True).exclude(uuid="") + .values('compute_id', 'uuid') + .annotate(count=Count('id')) + .filter(count__gt=1) + ) + for item in duplicates: + instances = list(Instance.objects.filter( + compute_id=item['compute_id'], + uuid=item['uuid'] + ).order_by('-id')) + primary = instances[0] + # Preserve is_template and non-default drbd across duplicate rows + merged_is_template = any(getattr(inst, 'is_template', False) for inst in instances) + non_default_drbd = next((inst.drbd for inst in instances if inst.drbd and inst.drbd != 'None'), None) + updated = False + if primary.is_template != merged_is_template: + primary.is_template = merged_is_template + updated = True + if non_default_drbd and primary.drbd != non_default_drbd: + primary.drbd = non_default_drbd + updated = True + if updated: + primary.save() + + for dup in instances[1:]: + # Reassign or merge UserInstance relationships + for ui in UserInstance.objects.filter(instance=dup): + existing = UserInstance.objects.filter(instance=primary, user=ui.user).first() + if not existing: + ui.instance = primary + ui.save() + else: + # Merge permissions with logical OR so no stronger permission is lost + updated = False + if ui.is_change and not existing.is_change: + existing.is_change = True + updated = True + if ui.is_delete and not existing.is_delete: + existing.is_delete = True + updated = True + if ui.is_vnc and not existing.is_vnc: + existing.is_vnc = True + updated = True + if updated: + existing.save() + ui.delete() + dup.delete() + + +class Migration(migrations.Migration): + + dependencies = [ + ('instances', '0011_createinstance_migrateinstance'), + ('accounts', '0001_initial'), + ] + + operations = [ + migrations.RunPython(deduplicate_instances, migrations.RunPython.noop), + migrations.AlterField( + model_name='flavor', + name='id', + field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), + ), + migrations.AlterField( + model_name='instance', + name='id', + field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), + ), + migrations.AddConstraint( + model_name='instance', + constraint=models.UniqueConstraint(fields=('compute', 'uuid'), name='unique_compute_uuid'), + ), + ] diff --git a/instances/models.py b/instances/models.py index 0a200314..8d34f88f 100644 --- a/instances/models.py +++ b/instances/models.py @@ -32,6 +32,11 @@ class Instance(models.Model): objects = InstanceManager() + class Meta: + constraints = [ + models.UniqueConstraint(fields=["compute", "uuid"], name="unique_compute_uuid") + ] + def __str__(self): return f"{self.compute}/{self.name}" diff --git a/instances/utils.py b/instances/utils.py index 0307f3f7..6af6131a 100644 --- a/instances/utils.py +++ b/instances/utils.py @@ -4,7 +4,9 @@ import string from accounts.models import UserInstance, UserAttributes from appsettings.settings import app_settings +from computes.utils import libvirt_compute_lock from django.conf import settings +from django.db import transaction from django.utils.translation import gettext_lazy as _ from vrtManager.connection import connection_manager from vrtManager.instance import wvmInstance, wvmInstances @@ -124,45 +126,86 @@ def migrate_instance( return if new_compute == instance.compute: return - try: - conn_migrate = wvmInstances( - new_compute.hostname, - new_compute.login, - new_compute.password, - new_compute.type, - ) + c1, c2 = ( + (instance.compute, new_compute) + if instance.compute.id < new_compute.id + else (new_compute, instance.compute) + ) + with libvirt_compute_lock(c1): + with libvirt_compute_lock(c2): + conn_migrate = None + try: + conn_migrate = wvmInstances( + new_compute.hostname, + new_compute.login, + new_compute.password, + new_compute.type, + ) - autostart = instance.autostart - conn_migrate.moveto( - instance.proxy, - instance.name, - live, - unsafe, - xml_del, - offline, - autoconverge, - compress, - postcopy, - ) - finally: - conn_migrate.close() + autostart = instance.autostart + conn_migrate.moveto( + instance.proxy, + instance.name, + live, + unsafe, + xml_del, + offline, + autoconverge, + compress, + postcopy, + ) + finally: + if conn_migrate is not None: + conn_migrate.close() - try: - conn_new = wvmInstance( - new_compute.hostname, - new_compute.login, - new_compute.password, - new_compute.type, - instance.name, - ) + conn_new = None + try: + conn_new = wvmInstance( + new_compute.hostname, + new_compute.login, + new_compute.password, + new_compute.type, + instance.name, + ) - if autostart: - conn_new.set_autostart(1) - finally: - conn_new.close() + if autostart: + conn_new.set_autostart(1) + finally: + if conn_new is not None: + conn_new.close() - instance.compute = new_compute - instance.save() + with transaction.atomic(): + target_inst = Instance.objects.filter( + compute=new_compute, uuid=instance.uuid + ).first() + if target_inst and target_inst.id != instance.id: + for ui in UserInstance.objects.filter(instance=instance): + existing_ui = UserInstance.objects.filter( + instance=target_inst, user=ui.user + ).first() + if not existing_ui: + ui.instance = target_inst + ui.save() + else: + updated = False + if ui.is_change and not existing_ui.is_change: + existing_ui.is_change = True + updated = True + if ui.is_delete and not existing_ui.is_delete: + existing_ui.is_delete = True + updated = True + if ui.is_vnc and not existing_ui.is_vnc: + existing_ui.is_vnc = True + updated = True + if updated: + existing_ui.save() + ui.delete() + instance.delete() + instance.id = target_inst.id + instance.compute = new_compute + else: + instance.compute = new_compute + instance.save() def refr(compute):