fix(instances): a finished migration is recorded, and a stale request does not migrate

What migrate_instance did around the migration itself:

- It checked the destination with a 1 s TCP probe of the libvirt port,
  which passes while libvirt is down or refuses the login. It now uses
  the compute's connection status, as the rest of the app does.
- Two requests for the same VM waited for the compute locks one after
  the other; the second then migrated with the VM's old location. The
  VM's record is read again under the locks, and the request is
  refused when the VM is no longer on the source.
- Restoring autostart on the destination came after the migration but
  before the database move; when it failed, the VM ran on the
  destination while the app recorded it on the source. Autostart is
  now read from libvirt and a failure to restore it is only logged.
- A database failure after a successful migration was logged as
  nothing; the log now says the VM migrated and recording it failed.
- When the destination's list had already made a record for the VM,
  the merge kept the owners but dropped the template mark.

computes/test_migration_address.py

@@ -10,6 +10,7 @@ from django.urls import reverse
 
 from computes.models import Compute
 from computes.validators import validate_migration_address
+from instances.models import Instance
 from instances.utils import migrate_instance
 
 
@@ -46,13 +47,11 @@ class MigrationAddressTests(TestCase):
         source = Compute.objects.create(name="ma-src", hostname="10.0.0.8", login="root", password="", type=2)
         dest = Compute.objects.create(name="ma-dst", hostname="kvm9", login="root", password="", type=2,
                                       migration_address="198.51.100.9")
-        vm = MagicMock(compute=source, autostart=False)
-        vm.name = "vm"
-        with patch("instances.utils.connection_manager.host_is_up", return_value=True), \
+        vm = Instance.objects.create(compute=source, name="vm", uuid="u-ma")
+        user = get_user_model().objects.create_superuser("ma-mig", "mm@example.com", "pw")
+        with patch.object(Compute, "status", True), \
              patch("instances.utils.wvmInstances") as instances, \
              patch("instances.utils.wvmInstance"), \
-             patch("instances.utils.Instance") as model:
-            model.objects.filter.return_value.first.return_value = None
-            user = get_user_model().objects.create_superuser("ma-mig", "mm@example.com", "pw")
+             patch.object(Instance, "proxy", MagicMock()):
             migrate_instance(dest, vm, user, live=True)
         self.assertEqual(instances.return_value.moveto.call_args.kwargs["uri"], "tcp://198.51.100.9")

instances/test_error_handling.py

@@ -78,9 +78,7 @@ class ErrorHandlingTestCase(TestCase):
         self.assertEqual(response.url, "http://testserver/instances/?x=1#boot_opt")
 
     def migrate(self, target_id, up=True):
-        with patch("instances.models.wvmInstance"), patch(
-            "instances.utils.connection_manager.host_is_up", return_value=up
-        ):
+        with patch("instances.models.wvmInstance"), patch.object(Compute, "status", up):
             return self.client.post(self.url("migrate"), {"compute_id": target_id})
 
     def test_migrate_to_a_down_host_is_reported_and_not_logged_as_done(self):
@@ -118,9 +116,7 @@ class ApiErrorTestCase(TestCase):
 
     def test_migrate_to_a_down_host_is_not_started(self):
         target = Compute.objects.create(name="api-down", hostname="10.0.0.9", login="", password="", type=4)
-        with patch("instances.models.wvmInstance"), patch(
-            "instances.utils.connection_manager.host_is_up", return_value=False
-        ):
+        with patch("instances.models.wvmInstance"), patch.object(Compute, "status", False):
             response = self.client.post(
                 reverse("instance-migrate-list"), {"instance": self.instance.id, "target_compute": target.id}
             )

instances/test_migration_bookkeeping.py

@@ -0,0 +1,87 @@
+"""What migrate_instance does around the migration itself: it checks the
+destination with the app's connection, re-reads the VM under the compute
+locks, and records a finished migration even when autostart fails."""
+
+from unittest.mock import MagicMock, patch
+
+from accounts.models import UserInstance
+from computes.models import Compute
+from django.contrib.auth import get_user_model
+from django.db import DatabaseError
+from django.test import TestCase
+from logs.models import Logs
+
+from instances.models import Instance
+from instances.utils import migrate_instance
+from vrtManager import util
+
+
+class MigrationBookkeepingTests(TestCase):
+    def setUp(self):
+        self.admin = get_user_model().objects.create_superuser("mb-admin", "mb@example.com", "pw")
+        self.source = Compute.objects.create(name="mb-src", hostname="198.51.100.8", login="root", password="", type=2)
+        self.dest = Compute.objects.create(name="mb-dst", hostname="198.51.100.9", login="root", password="", type=2)
+        self.vm = Instance.objects.create(compute=self.source, name="vm", uuid="u-mb", is_template=True)
+
+    def migrate(self, vm=None, up=True, autostart=False, autostart_error=None):
+        proxy = MagicMock()
+        proxy.get_autostart.return_value = autostart
+        with patch.object(Compute, "status", up), \
+             patch("instances.utils.wvmInstances") as instances, \
+             patch("instances.utils.wvmInstance") as instance, \
+             patch.object(Instance, "proxy", proxy):
+            instances.return_value.moveto.return_value = "live"
+            if autostart_error:
+                instance.return_value.set_autostart.side_effect = autostart_error
+            self.moveto = instances.return_value.moveto
+            self.set_autostart = instance.return_value.set_autostart
+            return migrate_instance(self.dest, vm or self.vm, self.admin, live=True)
+
+    def test_a_destination_the_app_cannot_connect_to_is_refused(self):
+        with self.assertRaisesRegex(util.OperationError, "mb-dst is not reachable"):
+            self.migrate(up=False)
+        self.moveto.assert_not_called()
+
+    def test_a_vm_moved_meanwhile_is_not_migrated_again(self):
+        stale = Instance.objects.get(id=self.vm.id)
+        Instance.objects.filter(id=self.vm.id).update(compute=self.dest)  # another request moved it
+        with self.assertRaisesRegex(util.OperationError, "no longer on mb-src"):
+            self.migrate(vm=stale)
+        self.moveto.assert_not_called()
+        Instance.objects.filter(id=self.vm.id).delete()  # merged into a record on the destination
+        with self.assertRaisesRegex(util.OperationError, "no longer on mb-src"):
+            self.migrate(vm=stale)
+        self.moveto.assert_not_called()
+
+    def test_autostart_is_read_from_libvirt_and_restored_on_the_destination(self):
+        self.vm.__dict__["autostart"] = False  # an earlier, cached read
+        self.migrate(autostart=True)
+        self.set_autostart.assert_called_once_with(1)
+
+    def test_a_failed_autostart_does_not_stop_the_records(self):
+        self.migrate(autostart=True, autostart_error=util.OperationError("no autostart"))
+        self.assertEqual(Instance.objects.get(id=self.vm.id).compute, self.dest)
+        self.assertEqual(
+            Logs.objects.get().message,
+            "Instance is migrated(live) to 198.51.100.9; setting its autostart there failed: no autostart",
+        )
+
+    def test_a_failed_record_is_logged_as_such(self):
+        with patch.object(Instance, "save", side_effect=DatabaseError("locked")), \
+             self.assertRaises(DatabaseError):
+            self.migrate()
+        self.assertEqual(
+            Logs.objects.get().message,
+            "Instance is migrated(live) to 198.51.100.9, but recording it failed: locked",
+        )
+
+    def test_a_record_the_destination_already_has_keeps_template_and_owners(self):
+        owner = get_user_model().objects.create_user("mb-owner")
+        UserInstance.objects.create(user=owner, instance=self.vm, is_change=True)
+        found = Instance.objects.create(compute=self.dest, name="vm", uuid="u-mb")  # listed there meanwhile
+        old_id = self.vm.id
+        self.migrate()
+        self.assertFalse(Instance.objects.filter(id=old_id).exists())
+        found.refresh_from_db()
+        self.assertTrue(found.is_template)
+        self.assertEqual(UserInstance.objects.get(user=owner).instance_id, found.id)

instances/test_migration_log.py

@@ -23,7 +23,7 @@ class MigrationLogTests(TestCase):
         self.vm = Instance.objects.create(compute=self.source, name="vm", uuid="u-ml")
 
     def migrate(self, outcome, **request):
-        with patch("instances.utils.connection_manager.host_is_up", return_value=True), \
+        with patch.object(Compute, "status", True), \
              patch("instances.utils.wvmInstances") as instances, \
              patch("instances.utils.wvmInstance"), \
              patch.object(Instance, "proxy", MagicMock()), \
@@ -45,6 +45,7 @@ class MigrationLogTests(TestCase):
         self.migrate("live", live=True)
         self.assertEqual(Logs.objects.get().message, "Instance is migrated(live) to 198.51.100.9 via 203.0.113.9")
         self.vm.compute = self.source
+        self.vm.save()
         self.dest.migration_address = "198.51.100.9"  # the host name itself
         self.migrate("live", live=True)
         self.assertEqual(Logs.objects.latest("id").message, "Instance is migrated(live) to 198.51.100.9")

instances/test_quota.py

@@ -85,10 +85,8 @@ class QuotaTests(TestCase):
     def test_a_user_over_quota_can_still_shrink(self):
         for n in range(5):
             self.own(self.up, n)  # 10 vCPUs used of 8
-        # (the TCP probe of the previous code would skip these VMs and hide the case)
-        with patch("instances.utils.connection_manager.host_is_up", return_value=True):
-            self.assertEqual(self.check(0, -1, 0, 0), ("", 0))
-            self.assertEqual(self.check(0, 0, -512, 0), ("", 0))
+        self.assertEqual(self.check(0, -1, 0, 0), ("", 0))
+        self.assertEqual(self.check(0, 0, -512, 0), ("", 0))
 
     def test_unlimited_users_are_not_checked(self):
         UserAttributes.objects.filter(user=self.user).update(max_instances=-1, max_cpus=-1, max_memory=0, max_disk_size=-1)

instances/test_security_mutations.py

@@ -236,6 +236,7 @@ class InstanceSecurityMutationsTestCase(TestCase):
         c2.password = ""
 
         inst1 = MagicMock()
+        inst1.id = "inst1"
         inst1.compute = c1
         inst1.name = "inst1"
         inst1.autostart = False
@@ -243,6 +244,7 @@ class InstanceSecurityMutationsTestCase(TestCase):
         inst1.uuid = "u1"
 
         inst2 = MagicMock()
+        inst2.id = "inst2"
         inst2.compute = c2
         inst2.name = "inst2"
         inst2.autostart = False
@@ -269,15 +271,20 @@ class InstanceSecurityMutationsTestCase(TestCase):
         def slow_moveto(*args, **kwargs):
             time.sleep(0.05)
 
-        with patch("instances.utils.connection_manager.host_is_up", return_value=True), \
-             patch("instances.utils.wvmInstances") as mock_instances_cls, \
+        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
-            mock_instance_model.objects.filter.return_value.first.return_value = None
+            # 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)

instances/utils.py

@@ -12,7 +12,6 @@ from django.utils.translation import gettext_lazy as _
 from libvirt import libvirtError
 from logs.views import addlogmsg
 from vrtManager import util
-from vrtManager.connection import connection_manager
 from vrtManager.instance import wvmInstance, wvmInstances
 
 from .models import Instance
@@ -165,16 +164,17 @@ def migrate_instance(
     used, or the error). Returns the mode."""
     if new_compute == instance.compute:
         raise util.OperationError(_("The instance is already on %(compute)s") % {"compute": new_compute.name})
-    if not connection_manager.host_is_up(new_compute.type, new_compute.hostname):
+    if not new_compute.status:
         raise util.OperationError(_("%(compute)s is not reachable") % {"compute": new_compute.name})
-    source_host = instance.compute.hostname
-    c1, c2 = (
-        (instance.compute, new_compute)
-        if instance.compute.id < new_compute.id
-        else (new_compute, instance.compute)
-    )
+    source = instance.compute
+    c1, c2 = (source, new_compute) if source.id < new_compute.id else (new_compute, source)
     with libvirt_compute_lock(c1):
         with libvirt_compute_lock(c2):
+            # another request may have moved it while this one waited (its
+            # record may then be gone, merged into one on the destination)
+            current = Instance.objects.filter(id=instance.id).first()
+            if current is None or current.compute_id != source.id:
+                raise util.OperationError(_("The instance is no longer on %(compute)s") % {"compute": source.name})
             conn_migrate = None
             try:
                 conn_migrate = wvmInstances(
@@ -184,7 +184,7 @@ def migrate_instance(
                     new_compute.type,
                 )
 
-                autostart = instance.autostart
+                autostart = instance.proxy.get_autostart()
                 mode = conn_migrate.moveto(
                     instance.proxy,
                     instance.name,
@@ -200,67 +200,83 @@ def migrate_instance(
                     "target": migration_target(new_compute),
                     "error": error,
                 }
-                addlogmsg(user.username, source_host, instance.name, msg)
+                addlogmsg(user.username, source.hostname, instance.name, msg)
                 raise
             finally:
                 if conn_migrate is not None:
                     conn_migrate.close()
 
-            conn_new = None
-            try:
-                conn_new = wvmInstance(
-                    new_compute.hostname,
-                    new_compute.login,
-                    new_compute.password,
-                    new_compute.type,
-                    instance.name,
-                    uuid=instance.uuid,
-                )
-
-                if autostart:
+            # The VM runs on new_compute now: what follows must not stop it
+            # from being recorded there.
+            msg = _("Instance is migrated(%(mode)s) to %(target)s") % {
+                "mode": mode,
+                "target": migration_target(new_compute),
+            }
+            if autostart:
+                conn_new = None
+                try:
+                    conn_new = wvmInstance(
+                        new_compute.hostname,
+                        new_compute.login,
+                        new_compute.password,
+                        new_compute.type,
+                        instance.name,
+                        uuid=instance.uuid,
+                    )
                     conn_new.set_autostart(1)
-            finally:
-                if conn_new is not None:
-                    conn_new.close()
+                except (libvirtError, OSError) as error:
+                    msg = _("%(msg)s; setting its autostart there failed: %(error)s") % {"msg": msg, "error": error}
+                finally:
+                    if conn_new is not None:
+                        conn_new.close()
 
-            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()
+            try:
+                record_migration(instance, new_compute)
+            except Exception as error:
+                failed = _("%(msg)s, but recording it failed: %(error)s") % {"msg": msg, "error": error}
+                addlogmsg(user.username, source.hostname, instance.name, failed)
+                raise
 
-    msg = _("Instance is migrated(%(mode)s) to %(target)s") % {"mode": mode, "target": migration_target(new_compute)}
-    addlogmsg(user.username, source_host, instance.name, msg)
+    addlogmsg(user.username, source.hostname, instance.name, msg)
     return mode
 
 
+def record_migration(instance, new_compute):
+    """Moves instance to new_compute in the database. A record the list of
+    new_compute made for it meanwhile takes its owners and template mark."""
+    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()
+            if instance.is_template and not target_inst.is_template:
+                target_inst.is_template = True
+                target_inst.save(update_fields=["is_template"])
+            instance.delete()
+            instance.id = target_inst.id
+            instance.compute = new_compute
+        else:
+            instance.compute = new_compute
+            instance.save()
+
+
 def migration_target(compute):
     """The destination for the log: its host name, and the migration address
     when one is set and differs."""