From 31cbe60aa4003ef33257aacf1c7698cea329e6bb Mon Sep 17 00:00:00 2001 From: catborise Date: Thu, 15 Oct 2026 20:20:16 +0300 Subject: [PATCH] feat(instances): migration runs as a job --- instances/api/serializers.py | 2 + instances/api/viewsets.py | 22 +-- instances/jobs.py | 90 +++++++++- .../templates/instances/settings_tab.html | 4 +- instances/tests/test_error_handling.py | 5 +- instances/tests/test_migrate_form.py | 18 +- instances/tests/test_migrate_job.py | 76 ++++++++ instances/tests/test_migration_bookkeeping.py | 21 +-- instances/tests/test_migration_log.py | 21 +-- instances/tests/test_security_mutations.py | 106 +---------- instances/utils.py | 170 ++++++++---------- instances/views.py | 30 ++-- 12 files changed, 283 insertions(+), 282 deletions(-) create mode 100644 instances/tests/test_migrate_job.py diff --git a/instances/api/serializers.py b/instances/api/serializers.py index 56b9185a..ce4d2451 100644 --- a/instances/api/serializers.py +++ b/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", ] diff --git a/instances/api/viewsets.py b/instances/api/viewsets.py index 7dd0f01b..6300c304 100644 --- a/instances/api/viewsets.py +++ b/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): diff --git a/instances/jobs.py b/instances/jobs.py index cf94ffb8..e5c16417 100644 --- a/instances/jobs.py +++ b/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.") diff --git a/instances/templates/instances/settings_tab.html b/instances/templates/instances/settings_tab.html index aebce899..2c527829 100644 --- a/instances/templates/instances/settings_tab.html +++ b/instances/templates/instances/settings_tab.html @@ -580,7 +580,7 @@

{% trans "For migration both host servers must have equal settings and OS type" %}

- {% csrf_token %} + {% csrf_token %}{% submission_key %}
@@ -642,7 +642,7 @@
{% if computes_count != 1 %} - + {% else %} {% endif %} diff --git a/instances/tests/test_error_handling.py b/instances/tests/test_error_handling.py index 3a167a65..b65b854a 100644 --- a/instances/tests/test_error_handling.py +++ b/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"]) diff --git a/instances/tests/test_migrate_form.py b/instances/tests/test_migrate_form.py index 8ddbb98f..a188395f 100644 --- a/instances/tests/test_migrate_form.py +++ b/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 = """migrate-form-vm1281 @@ -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]) diff --git a/instances/tests/test_migrate_job.py b/instances/tests/test_migrate_job.py new file mode 100644 index 00000000..09c2f2ba --- /dev/null +++ b/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) diff --git a/instances/tests/test_migration_bookkeeping.py b/instances/tests/test_migration_bookkeeping.py index 734f44fa..99f0deb5 100644 --- a/instances/tests/test_migration_bookkeeping.py +++ b/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) diff --git a/instances/tests/test_migration_log.py b/instances/tests/test_migration_log.py index 4b8722ab..5674c497 100644 --- a/instances/tests/test_migration_log.py +++ b/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)"}) diff --git a/instances/tests/test_security_mutations.py b/instances/tests/test_security_mutations.py index f3977234..9c38d947 100644 --- a/instances/tests/test_security_mutations.py +++ b/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)) diff --git a/instances/utils.py b/instances/utils.py index 880d3e7e..3278bbd5 100644 --- a/instances/utils.py +++ b/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 diff --git a/instances/views.py b/instances/views.py index 3f24e88c..6c9bbe18 100755 --- a/instances/views.py +++ b/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