feat(instances): migration runs as a job

instances/api/serializers.py

@@ -96,6 +96,7 @@ class MigrateSerializer(serializers.ModelSerializer):
     autoconverge = serializers.BooleanField(default=True)
     compress = serializers.BooleanField(default=False)
     unsafe = serializers.BooleanField(default=False)
+    submission_key = serializers.CharField(write_only=True)
 
     class Meta:
         model = MigrateInstance
@@ -107,4 +108,5 @@ class MigrateSerializer(serializers.ModelSerializer):
             "autoconverge",
             "compress",
             "unsafe",
+            "submission_key",
         ]

instances/api/viewsets.py

@@ -5,7 +5,6 @@ from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django.urls import reverse
 from instances.models import Flavor, Instance
-from instances.utils import migrate_instance
 from instances.views import destroy as instance_destroy
 from instances.views import (
     force_off,
@@ -184,22 +183,17 @@ class MigrateViewSet(viewsets.ViewSet):
     serializer_class = MigrateSerializer
     queryset = ""
 
-    @extend_schema(responses=StatusSerializer)
+    @extend_schema(responses={202: OperationSerializer, 400: dict})
     def create(self, request):
         serializer = MigrateSerializer(data=request.data)
-        if serializer.is_valid():
-            instance = serializer.validated_data["instance"]
-            target_host = serializer.validated_data["target_compute"]
-            options = {
-                name: serializer.validated_data.get(name, False)
-                for name in ("live", "unsafe", "offline", "autoconverge", "compress")
-            }
-            # migrate_instance logs the outcome; it returns when it is done
-            mode = migrate_instance(target_host, instance, request.user, **options)
-
-            return Response({"status": f"instance is migrated ({mode})"})
-        else:
+        if not serializer.is_valid():
             return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+        data = serializer.validated_data
+        raw = {"instance": data["instance"].pk, "target": data["target_compute"].pk,
+               **{name: data.get(name, False) for name in ("live", "unsafe", "offline", "autoconverge", "compress")}}
+        op = jobs.submit("instance.migrate", request.user, raw, data["submission_key"])
+        return Response(OperationSerializer(op, context={"request": request}).data, status=status.HTTP_202_ACCEPTED,
+                        headers={"Location": reverse("operation-detail", args=[op.pk])})
 
 
 class FlavorViewSet(viewsets.ModelViewSet):

instances/jobs.py

@@ -77,9 +77,9 @@ def _recovery_proxy(op):
         raise  # unreachable: recovery waits and retries
 
 
-def _libvirt_busy(proxy):
+def _libvirt_busy(dom):
     """libvirt still runs a job on the VM that the worker no longer waits for."""
-    return proxy.instance.isActive() and proxy.instance.jobInfo()[0] != VIR_DOMAIN_JOB_NONE
+    return dom.isActive() and dom.jobInfo()[0] != VIR_DOMAIN_JOB_NONE
 
 
 def _log(op, state, message):
@@ -326,7 +326,7 @@ class CreateSnapshot(Kind):
         proxy = _recovery_proxy(op)
         if proxy is None:
             return Outcome(Operation.UNCERTAIN, error=GONE)
-        if _libvirt_busy(proxy):
+        if _libvirt_busy(proxy.instance):
             return Outcome(Operation.RUNNING, phase=gettext_noop("libvirt is still taking the snapshot"))
         if op.params["name"] in proxy.instance.snapshotListNames(0):
             return Outcome(Operation.SUCCEEDED)
@@ -378,7 +378,7 @@ class RevertSnapshot(Kind):
         proxy = _recovery_proxy(op)
         if proxy is None:
             return Outcome(Operation.UNCERTAIN, error=GONE)
-        if _libvirt_busy(proxy):
+        if _libvirt_busy(proxy.instance):
             return Outcome(Operation.RUNNING, phase=gettext_noop("libvirt is still reverting the VM"))
         if _current(proxy) == op.params["name"] != op.intent.get("was_current"):
             return Outcome(Operation.SUCCEEDED)
@@ -580,3 +580,85 @@ class CreateInstance(Kind):
         compute, _sep, name = op.resource_label.rpartition("/")
         message = "Instance is created"
         return compute, name, message if state == Operation.SUCCEEDED else f"{message} ({state})"
+
+
+MIGRATE_OPTIONS = ("live", "unsafe", "offline", "autoconverge", "compress")
+
+
+def _target(instance, target_pk):
+    """The destination; refused when the VM is already there or it cannot be reached."""
+    target = Compute.objects.get(pk=target_pk)
+    if target.pk == instance.compute_id:
+        raise util.OperationError(_("The instance is already on %(compute)s") % {"compute": target.name})
+    if not target.status:
+        raise util.OperationError(_("%(compute)s is not reachable") % {"compute": target.name})
+    return target
+
+
+def _domain(compute_pk, uuid):
+    """The VM on the compute, or None when it has none there."""
+    c = Compute.objects.get(pk=compute_pk)
+    try:
+        return wvmConnect(c.hostname, c.login, c.password, c.type).wvm.lookupByUUIDString(uuid)
+    except libvirtError as err:
+        if err.get_error_code() != VIR_ERR_NO_DOMAIN:
+            raise  # unreachable: recovery waits and retries
+        return None
+
+
+@register
+class MigrateInstance(Kind):
+    name = "instance.migrate"
+    label = gettext_noop("Migrate VM")
+
+    def normalize(self, raw):
+        return {"instance": int(raw["instance"]), "target": int(raw["target"]),
+                **{k: bool(raw.get(k)) for k in MIGRATE_OPTIONS}}
+
+    def validate(self, user, params):
+        if not user.is_superuser:
+            raise PermissionDenied
+        _target(_instance(params["instance"]), params["target"])
+
+    def plan(self, user, params):
+        instance = _instance(params["instance"])
+        keys = vm_claims(Compute.objects.get(pk=params["target"]), instance.name)
+        keys.update({claims.compute_key(instance.compute_id): Claim.SHARED,
+                     claims.reconcile_key(instance.compute_id): Claim.EXCLUSIVE,
+                     claims.instance_key(instance.uuid): Claim.EXCLUSIVE})
+        return Plan(claims=keys, resource=claims.instance_key(instance.uuid),
+                    label=f"{instance.compute.name}/{instance.name}",
+                    intent={"compute": instance.compute_id, "uuid": instance.uuid})
+
+    def precheck(self, op, ctx):
+        if op.user is None or not op.user.is_active or not op.user.is_superuser:
+            raise PermissionDenied
+        instance = _instance(op.params["instance"])
+        if instance.compute_id != op.intent["compute"]:
+            raise util.OperationError("The VM moved to another compute after the job was queued")
+        _target(instance, op.params["target"])
+
+    def run(self, op, ctx):
+        p = op.params
+        instance = _instance(p["instance"])
+        ctx.record_intent()
+        ctx.set_phase(gettext_noop("Migrating the VM"))
+        # migrate_instance logs the outcome
+        utils.migrate_instance(Compute.objects.get(pk=p["target"]), instance, op.user,
+                               **{k: p[k] for k in MIGRATE_OPTIONS})
+        return {}
+
+    def reconcile(self, op):
+        source = _domain(op.intent["compute"], op.intent["uuid"])
+        dest = _domain(op.params["target"], op.intent["uuid"])
+        if source is not None and _libvirt_busy(source):
+            return Outcome(Operation.RUNNING, phase=gettext_noop("libvirt is still migrating the VM"))
+        if source is None and dest is not None:
+            instance = Instance.objects.filter(pk=op.params["instance"], compute_id=op.intent["compute"]).first()
+            if instance is not None:  # the job stopped before it moved the record
+                utils.record_migration(instance, Compute.objects.get(pk=op.params["target"]))
+            return Outcome(Operation.SUCCEEDED)
+        if source is not None and dest is None:
+            return Outcome(Operation.FAILED, error="The VM stayed on the source host")
+        return Outcome(Operation.UNCERTAIN, error="The VM is on both computes or on neither; "
+                                                  "check them, then resolve this job.")

instances/templates/instances/settings_tab.html

@@ -580,7 +580,7 @@
         <div role="tabpanel" class="tab-pane tab-pane-bordered" id="migrate">
             <p>{% trans "For migration both host servers must have equal settings and OS type" %}</p>
             <form action="{% url 'instances:migrate' instance.id %}" class="ms-3 form" method="post" role="form" aria-label="Migrate instance form">
-                {% csrf_token %}
+                {% csrf_token %}{% submission_key %}
                 <div class="row">
                     <label class="col-sm-3 col-form-label">{% trans "Original host" %}</label>
                     <div class="col-sm-6">
@@ -642,7 +642,7 @@
                     </div>
                 </div>
                 {% if computes_count != 1 %}
-                    <button type="submit" class="btn btn-lg btn-success float-end" onclick="showPleaseWaitDialog()" name="migrate" >{% trans "Migrate" %}</button>
+                    <button type="submit" class="btn btn-lg btn-success float-end" name="migrate" >{% trans "Migrate" %}</button>
                 {% else %}
                     <button class="btn btn-lg btn-success float-end disabled" disabled>{% trans "Migrate" %}</button>
                 {% endif %}

instances/tests/test_error_handling.py

@@ -83,7 +83,7 @@ class ErrorHandlingTestCase(TestCase):
 
     def migrate(self, target_id, up=True):
         with patch("instances.models.wvmInstance"), patch.object(Compute, "status", up):
-            return self.client.post(self.url("migrate"), {"compute_id": target_id})
+            return self.client.post(self.url("migrate"), {"compute_id": target_id, "submission_key": "key-migrate01"})
 
     def test_migrate_to_a_down_host_is_reported_and_not_logged_as_done(self):
         target = Compute.objects.create(name="down", hostname="10.0.0.9", login="", password="", type=4)
@@ -122,7 +122,8 @@ class ApiErrorTestCase(TestCase):
         target = Compute.objects.create(name="api-down", hostname="10.0.0.9", login="", password="", type=4)
         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}
+                reverse("instance-migrate-list"), {"instance": self.instance.id, "target_compute": target.id,
+                                                "submission_key": "key-migrate02"}
             )
         self.assertEqual(response.status_code, 400)
         self.assertIn("not reachable", response.json()["detail"])

instances/tests/test_migrate_form.py

@@ -10,6 +10,7 @@ from django.contrib.auth import get_user_model
 from django.test import TestCase
 from django.urls import reverse
 from instances.models import Instance
+from operations.models import Operation
 from rest_framework.test import APIClient
 
 DOMAIN = """<domain type='test'><name>migrate-form-vm</name><memory unit='MiB'>128</memory><vcpu>1</vcpu>
@@ -55,10 +56,13 @@ class MigrateFormTests(TestCase):
         api = APIClient()
         api.force_authenticate(self.admin)
         request = {"instance": self.vm.id, "target_compute": self.other.id, "live": True, "offline": False}
-        with patch("instances.api.viewsets.migrate_instance", return_value="live") as migrate:
-            for encoding in ("json", "multipart"):
-                api.post("/api/v1/migrate/", request, format=encoding)
-            api.post("/api/v1/migrate/", {**request, "autoconverge": False}, format="json")
-            api.post("/api/v1/migrate/", {**request, "unsafe": True}, format="json")
-        self.assertEqual([c.kwargs["autoconverge"] for c in migrate.call_args_list], [True, True, False, True])
-        self.assertEqual([c.kwargs["unsafe"] for c in migrate.call_args_list], [False, False, False, True])
+        with patch("instances.api.viewsets.jobs.submit") as submit:
+            submit.return_value = Operation.objects.create(kind="instance.migrate", user=self.admin,
+                                                           username=self.admin.username, submission_key="key-api00000")
+            for n, encoding in enumerate(("json", "multipart"), 1):
+                api.post("/api/v1/migrate/", {**request, "submission_key": f"key-api0000{n}"}, format=encoding)
+            api.post("/api/v1/migrate/", {**request, "autoconverge": False, "submission_key": "key-api00003"},
+                     format="json")
+            api.post("/api/v1/migrate/", {**request, "unsafe": True, "submission_key": "key-api00004"}, format="json")
+        self.assertEqual([c.args[2]["autoconverge"] for c in submit.call_args_list], [True, True, False, True])
+        self.assertEqual([c.args[2]["unsafe"] for c in submit.call_args_list], [False, False, False, True])

instances/tests/test_migrate_job.py

@@ -0,0 +1,76 @@
+"""A migration runs as a job: the page queues it, the worker migrates."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+from django.urls import reverse
+
+from appsettings.settings import get_settings
+from computes.models import Compute
+from instances.jobs import MigrateInstance
+from instances.models import Instance
+from operations.jobs import Context, get_kind
+from operations.models import Claim, Operation, acquire
+
+UUID = "31313131-4242-5353-6464-757575757575"
+
+
+class MigrateJobTestCase(TransactionTestCase):
+    def setUp(self):
+        get_settings()
+        self.source = Compute.objects.create(name="mig-src", hostname="192.0.2.50", login="root", password="", type=1)
+        self.dest = Compute.objects.create(name="mig-dst", hostname="192.0.2.51", login="root", password="", type=1)
+        self.vm = Instance.objects.create(compute=self.source, name="mig-vm", uuid=UUID)
+        self.client.force_login(get_user_model().objects.create_superuser("mig_admin", "a@example.com", "pw"))
+        proxy = MagicMock()
+        proxy.get_autostart.return_value = False
+        proxy.get_nvram.return_value = None
+        patch.object(Compute, "status", True).start()
+        patch.object(Instance, "proxy", proxy).start()
+        self.dest_conn = patch("instances.utils.wvmInstances").start().return_value
+        self.dest_conn.moveto.return_value = "live"
+        self.addCleanup(patch.stopall)
+
+    def queue(self, **form):
+        self.client.post(reverse("instances:migrate", args=[self.vm.pk]),
+                         {"compute_id": self.dest.pk, "submission_key": "key-mig00001", **form})
+        return Operation.objects.get()
+
+    def start(self, op):
+        """The job as the worker takes it."""
+        acquire("wvc-test:1:a", "wvc-test")
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker="wvc-test:1:a")
+        op.refresh_from_db()
+        return Context(op, SimpleNamespace(identity="wvc-test:1:a"))
+
+    def test_the_worker_migrates_and_moves_the_record(self):
+        op = self.queue(live_migrate="true")
+        self.assertEqual(set(Claim.objects.values_list("resource", "mode")), {
+            (f"instance:{UUID}", "exclusive"), (f"name:{self.dest.pk}:mig-vm", "exclusive"),
+            (f"compute:{self.source.pk}", "shared"), (f"reconcile:{self.source.pk}", "exclusive"),
+            (f"compute:{self.dest.pk}", "shared"), (f"reconcile:{self.dest.pk}", "exclusive")})
+        self.dest_conn.moveto.assert_not_called()  # nothing ran in the request
+        ctx = self.start(op)
+        get_kind(op.kind).precheck(op, ctx)
+        get_kind(op.kind).run(op, ctx)
+        self.assertTrue(self.dest_conn.moveto.call_args.args[2])  # live
+        self.vm.refresh_from_db()
+        self.assertEqual(self.vm.compute, self.dest)
+
+    def test_recovery_reads_where_the_vm_is(self):
+        op = Operation(params={"instance": self.vm.pk, "target": self.dest.pk},
+                       intent={"compute": self.source.pk, "uuid": UUID})
+        dom = MagicMock()
+        dom.isActive.return_value = False
+        with patch("instances.jobs._domain", side_effect=[dom, None]):  # source, destination
+            self.assertEqual(MigrateInstance().reconcile(op).state, Operation.FAILED)
+        dom.isActive.return_value = True
+        dom.jobInfo.return_value = [2] + [0] * 11  # VIR_DOMAIN_JOB_UNBOUNDED
+        with patch("instances.jobs._domain", side_effect=[dom, dom]):
+            self.assertEqual(MigrateInstance().reconcile(op).state, Operation.RUNNING)
+        with patch("instances.jobs._domain", side_effect=[None, dom]):
+            self.assertEqual(MigrateInstance().reconcile(op).state, Operation.SUCCEEDED)
+        self.vm.refresh_from_db()
+        self.assertEqual(self.vm.compute, self.dest)

instances/tests/test_migration_bookkeeping.py

@@ -1,6 +1,5 @@
-"""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."""
+"""What migrate_instance does around the migration itself: it records a
+finished migration even when autostart fails."""
 
 import re
 from pathlib import Path
@@ -40,22 +39,6 @@ class MigrationBookkeepingTests(TestCase):
             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)

instances/tests/test_migration_log.py

@@ -1,15 +1,13 @@
-"""A migration is logged where it runs (migrate_instance), for the VM page
-and the API alike: the mode it used on success, the error on failure."""
+"""A migration is logged where it runs (migrate_instance, in the job): the
+mode it used on success, the error on failure."""
 
 from unittest.mock import MagicMock, patch
 
 from computes.models import Compute
 from django.contrib.auth import get_user_model
 from django.test import TestCase
-from django.urls import reverse
 from libvirt import libvirtError
 from logs.models import Logs
-from rest_framework.test import APIClient
 
 from instances.models import Instance
 from instances.utils import migrate_instance
@@ -99,18 +97,3 @@ class MigrationLogTests(TestCase):
             Logs.objects.get().message,
             "Instance migration to 198.51.100.9 failed: A shut-off VM cannot be migrated live",
         )
-
-    def test_the_page_and_the_api_log_once_through_migrate_instance(self):
-        self.client.force_login(self.admin)
-        api = APIClient()
-        api.force_authenticate(self.admin)
-        with patch("instances.views.utils.migrate_instance", return_value="live") as page_migrate:
-            self.client.post(reverse("instances:migrate", args=[self.vm.id]),
-                             {"compute_id": self.dest.id, "live_migrate": "true"})
-        with patch("instances.api.viewsets.migrate_instance", return_value="live") as api_migrate:
-            response = api.post("/api/v1/migrate/", {"instance": self.vm.id, "target_compute": self.dest.id,
-                                                     "live": True, "offline": False}, format="json")
-        self.assertEqual(page_migrate.call_count, 1)
-        self.assertEqual(api_migrate.call_count, 1)
-        self.assertFalse(Logs.objects.exists())  # migrate_instance logs; it is patched here
-        self.assertEqual(response.json(), {"status": "instance is migrated (live)"})

instances/tests/test_security_mutations.py

@@ -3,8 +3,7 @@ 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.db import connection
-from django.test import TestCase, TransactionTestCase
+from django.test import TestCase
 from django.urls import reverse
 
 from accounts.models import UserInstance
@@ -336,106 +335,3 @@ class InstanceSecurityMutationsTestCase(TestCase):
             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))

instances/utils.py

@@ -5,13 +5,11 @@ 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 libvirt import libvirtError
 from logs.views import addlogmsg
-from vrtManager import util
 from vrtManager.instance import wvmInstance, wvmInstances
 
 from .models import Instance
@@ -167,101 +165,89 @@ def migrate_instance(
 ):
     """Migrates instance to new_compute and logs the outcome (the mode
     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 new_compute.status:
-        raise util.OperationError(_("%(compute)s is not reachable") % {"compute": new_compute.name})
     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(
-                    new_compute.hostname,
-                    new_compute.login,
-                    new_compute.password,
-                    new_compute.type,
-                )
+    conn_migrate = None
+    try:
+        conn_migrate = wvmInstances(
+            new_compute.hostname,
+            new_compute.login,
+            new_compute.password,
+            new_compute.type,
+        )
 
-                autostart = instance.proxy.get_autostart()
-                nvram = instance.proxy.get_nvram()
-                mode = conn_migrate.moveto(
-                    instance.proxy,
-                    instance.name,
-                    live,
-                    unsafe,
-                    offline,
-                    autoconverge,
-                    compress,
-                    uri=new_compute.migration_uri,
-                    timeout=MIGRATION_TIME_LIMIT,
-                )
-            except Exception as error:
-                msg = _("Instance migration to %(target)s failed: %(error)s") % {
-                    "target": migration_target(new_compute),
-                    "error": error,
-                }
-                addlogmsg(user.username, source.hostname, instance.name, msg)
-                raise
-            finally:
-                if conn_migrate is not None:
-                    conn_migrate.close()
+        autostart = instance.proxy.get_autostart()
+        nvram = instance.proxy.get_nvram()
+        mode = conn_migrate.moveto(
+            instance.proxy,
+            instance.name,
+            live,
+            unsafe,
+            offline,
+            autoconverge,
+            compress,
+            uri=new_compute.migration_uri,
+            timeout=MIGRATION_TIME_LIMIT,
+        )
+    except Exception as error:
+        msg = _("Instance migration to %(target)s failed: %(error)s") % {
+            "target": migration_target(new_compute),
+            "error": error,
+        }
+        addlogmsg(user.username, source.hostname, instance.name, msg)
+        raise
+    finally:
+        if conn_migrate is not None:
+            conn_migrate.close()
 
-            # 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),
+    # 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)
+        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()
+
+    # libvirt keeps the NVRAM file on the source; a VM created there
+    # later with the same name would start with these variables.
+    if nvram:
+        conn_new = None
+        try:
+            conn_new = wvmInstances(
+                new_compute.hostname, new_compute.login, new_compute.password, new_compute.type
+            )
+            instance.proxy.remove_nvram(nvram, conn_new)
+        except Exception as error:  # the migration stands and is recorded
+            msg = _("%(msg)s; removing its NVRAM file %(path)s on the source failed: %(error)s") % {
+                "msg": msg,
+                "path": nvram,
+                "error": error,
             }
-            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)
-                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()
+        finally:
+            if conn_new is not None:
+                conn_new.close()
 
-            # libvirt keeps the NVRAM file on the source; a VM created there
-            # later with the same name would start with these variables.
-            if nvram:
-                conn_new = None
-                try:
-                    conn_new = wvmInstances(
-                        new_compute.hostname, new_compute.login, new_compute.password, new_compute.type
-                    )
-                    instance.proxy.remove_nvram(nvram, conn_new)
-                except Exception as error:  # the migration stands and is recorded
-                    msg = _("%(msg)s; removing its NVRAM file %(path)s on the source failed: %(error)s") % {
-                        "msg": msg,
-                        "path": nvram,
-                        "error": error,
-                    }
-                finally:
-                    if conn_new is not None:
-                        conn_new.close()
-
-            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
+    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
 
     addlogmsg(user.username, source.hostname, instance.name, msg)
     return mode

instances/views.py

@@ -723,30 +723,24 @@ def destroy(request, pk):
 @require_POST
 @superuser_only
 def migrate(request, pk):
-    instance = get_instance(request.user, pk)
-
+    get_instance(request.user, pk)
     compute_id = request.POST.get("compute_id", "")
-    live = request.POST.get("live_migrate", False)
-    unsafe = request.POST.get("unsafe_migrate", False)
-    offline = request.POST.get("offline_migrate", False)
-    autoconverge = request.POST.get("autoconverge", False)
-    compress = request.POST.get("compress", False)
-
     if not compute_id.isdigit():
         raise Http404
     target_host = get_object_or_404(Compute, pk=compute_id)
-
+    raw = {"instance": pk, "target": target_host.pk, "live": request.POST.get("live_migrate"),
+           "unsafe": request.POST.get("unsafe_migrate"), "offline": request.POST.get("offline_migrate"),
+           "autoconverge": request.POST.get("autoconverge"), "compress": request.POST.get("compress")}
     try:
-        # migrate_instance logs the outcome
-        utils.migrate_instance(target_host, instance, request.user, live, unsafe, offline, autoconverge, compress)
+        op = jobs.submit("instance.migrate", request.user, raw, request.POST.get("submission_key", ""))
     except claims.Busy as busy:
-        return _busy(request, instance.id, busy)
-    except (libvirtError, OSError) as err:
-        messages.error(request, err)
-
-    return get_safe_redirect(
-        request, default=reverse("instances:instance", args=[instance.id])
-    )
+        return _busy(request, pk, busy)
+    except (jobs.SubmissionConflict, libvirtError) as err:
+        messages.error(request, error_text(request, err))
+    else:
+        messages.info(request, _("The VM is being migrated; follow it in the job panel."))
+        jobs.remember_job(request, op)
+    return get_safe_redirect(request, default=reverse("instances:instance", args=[pk]))
 
 
 GSTFSD_PORT = 16510