From aadf5270f2524f0642660b4dfae945184ea6814a Mon Sep 17 00:00:00 2001 From: catborise Date: Sat, 7 Nov 2026 22:47:08 +0300 Subject: [PATCH] fix(migrate): recovery removes the disk copies of a migration whose worker stopped A worker that stopped during the copy left the partial files on the destination: libvirt ends the migration and the VM stays, but the files took the space and refused the next try, as a copy never replaces a file. The job records the copies before libvirt makes them, and recovery deletes them while the destination has no VM of it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Kk4h2TcUiaWmU5MUszYSea --- instances/jobs.py | 23 ++++++++++++++- instances/tests/test_migrate_job.py | 30 +++++++++++++++++++- instances/utils.py | 4 ++- vrtManager/instance.py | 11 ++++--- vrtManager/tests/test_disk_copy_migration.py | 8 ++++++ 5 files changed, 69 insertions(+), 7 deletions(-) diff --git a/instances/jobs.py b/instances/jobs.py index db738bc7..16201f19 100644 --- a/instances/jobs.py +++ b/instances/jobs.py @@ -20,7 +20,7 @@ from lxml import etree from vrtManager import util from vrtManager.connection import wvmConnect from vrtManager.create import cpu_choices, wvmCreate -from vrtManager.instance import HOST_DEVICES, check_snapshot_name, wvmInstance +from vrtManager.instance import HOST_DEVICES, check_snapshot_name, wvmInstance, wvmInstances from vrtManager.nodedevs import check_hostdevs from accounts.models import UserInstance, UserSSHKey @@ -994,6 +994,25 @@ def _shut_off_again(op, dom): dom.destroy() +def _remove_copies(op): + """The disk copies a migration that stopped made on the destination, + which has no VM of it: they would take the space and refuse the next + try (a copy never replaces a file). One that cannot go is logged.""" + if not op.intent.get("copies"): + return + target = Compute.objects.get(pk=op.params["target"]) + dest = wvmInstances(target.hostname, target.login, target.password, target.type) + try: + copies = [{"pool": dest.wvm.storagePoolLookupByName(c["pool"]), "dest": c["dest"]} for c in op.intent["copies"]] + dest.remove_partial_copies(op.intent["uuid"], copies, util.OperationError("the migration stopped")) + except (libvirtError, util.OperationError) as err: + source, _sep, name = op.resource_label.rpartition("/") + write(op.username, source, name, f"VM {name}: remove the partial disk copies on {target.name}: " + f"{safe_text(err)}", False, op.client_ip, op.intent["uuid"]) + finally: + dest.close() + + def _moved(op, dest): """What migrate_instance does once libvirt moved the VM, for a job that stopped before it: autostart is per host, and the source keeps the @@ -1076,6 +1095,7 @@ class MigrateInstance(Kind): utils.migrate_instance(Compute.objects.get(pk=p["target"]), instance, op.user, disk_pools=p["disk_pools"], ip=op.client_ip, on_moved=lambda: ctx.record_intent(outcome_logged=True), + on_copy=lambda copies: ctx.record_intent(copies=copies), **{k: p[k] for k in MIGRATE_OPTIONS}) except Exception as err: # migrate_instance wrote the error's Logs row; recovery reads where the VM is, @@ -1128,6 +1148,7 @@ class MigrateInstance(Kind): return Outcome(Operation.SUCCEEDED, result=_moved(op, dest)) if source is not None and dest is None: _shut_off_again(op, source) + _remove_copies(op) if op.cancel_requested: return Outcome(Operation.CANCELLED) return Outcome(Operation.FAILED, error=say(_("The VM stayed on the source host"))) diff --git a/instances/tests/test_migrate_job.py b/instances/tests/test_migrate_job.py index 9e5d304f..0deb8081 100644 --- a/instances/tests/test_migrate_job.py +++ b/instances/tests/test_migrate_job.py @@ -1,5 +1,6 @@ """A migration runs as a job: the page queues it, the worker migrates.""" +from functools import partial from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -12,10 +13,11 @@ from computes.models import Compute from instances.jobs import MigrateInstance from instances.models import Instance from logs.models import Logs -from libvirt import VIR_DOMAIN_XML_INACTIVE, VIR_ERR_SYSTEM_ERROR, VIR_FROM_RPC, libvirtError +from libvirt import VIR_DOMAIN_XML_INACTIVE, VIR_ERR_NO_DOMAIN, VIR_ERR_SYSTEM_ERROR, VIR_FROM_RPC, libvirtError from operations.jobs import Context, get_kind from operations.models import Claim, Operation, acquire from operations.utils import public_error_text +from vrtManager.instance import wvmInstances from vrtManager.util import OperationError UUID = "31313131-4242-5353-6464-757575757575" @@ -85,6 +87,32 @@ class MigrateJobTestCase(TransactionTestCase): self.vm.refresh_from_db() self.assertEqual(self.vm.compute, self.dest) + def test_recovery_removes_the_copies_of_a_migration_that_stopped(self): + op = Operation(params={"instance": self.vm.pk, "target": self.dest.pk}, resource_label="mig-src/mig-vm", + intent={"compute": self.source.pk, "uuid": UUID, + "copies": [{"pool": "fast", "dest": "/srv/fast/mig-vm.qcow2"}]}) + dom = MagicMock(**{"isActive.return_value": True, "jobInfo.return_value": [0] * 12}) + with patch("instances.jobs.wvmInstances") as hosts: + dest = hosts.return_value + with patch("instances.jobs._domain", side_effect=[dom, None]): # source, destination + self.assertEqual(MigrateInstance().reconcile(op).state, Operation.FAILED) + uuid, copies, _error = dest.remove_partial_copies.call_args.args + self.assertEqual((uuid, copies), (UUID, [{"pool": dest.wvm.storagePoolLookupByName.return_value, + "dest": "/srv/fast/mig-vm.qcow2"}])) + dest.wvm.storagePoolLookupByName.assert_called_once_with("fast") + # a copy that cannot go is logged; the job ends as it would + pool = dest.wvm.storagePoolLookupByName.return_value + pool.storageVolLookupByName.return_value.delete.side_effect = libvirtError("busy") + dest.wvm.lookupByUUIDString.side_effect = libvirtError("Domain not found") + dest.wvm.lookupByUUIDString.side_effect.err = (VIR_ERR_NO_DOMAIN,) + dest._find_volume, dest.delete_nvram = wvmInstances._find_volume, lambda name: None + dest.remove_partial_copies = partial(wvmInstances.remove_partial_copies, dest) + with patch("instances.jobs._domain", side_effect=[dom, None]): + self.assertEqual(MigrateInstance().reconcile(op).state, Operation.FAILED) + self.assertEqual(Logs.objects.latest("id").message, "VM mig-vm: remove the partial disk copies on mig-dst: " + "the migration stopped; remove the partial disk copies " + "from the destination host: /srv/fast/mig-vm.qcow2") + def test_recovery_shuts_off_a_shut_off_vm_the_migration_left_started(self): self.vm.proxy.get_status.return_value = 5 op = self.queue() diff --git a/instances/utils.py b/instances/utils.py index 3314cb25..f904a7bd 100644 --- a/instances/utils.py +++ b/instances/utils.py @@ -250,12 +250,13 @@ def migrate_instance( disk_pools=None, ip=None, on_moved=None, + on_copy=None, ): """Migrates instance to new_compute and logs the outcome (the mode used, or the error), then each later step that failed in a row of its own; on_moved is called once the outcome's row is written. disk_pools: {target: pool} of the disks copied into pools of new_compute (moveto's - pools); their source files go with delete_source unless a VM or volume + pools), on_copy as moveto's; their source files go with delete_source unless a VM or volume of any compute may use them (used_anywhere), and are named in the log otherwise. Returns the mode.""" source, target = instance.compute, migration_target(new_compute) @@ -285,6 +286,7 @@ def migrate_instance( compress, uri=new_compute.migration_uri, pools=disk_pools, + on_copy=on_copy, ) except Exception as error: log(f"migrate to {target}", error) diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 79729784..aab376b9 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -116,7 +116,7 @@ def _rate(before, after, seconds, scale=1): class wvmInstances(wvmConnect): def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None, - pools=None): + pools=None, on_copy=None): """Migrates the VM of conn (a wvmInstance on the source) here. The mode follows the VM's state, read once: a running VM migrates @@ -128,9 +128,10 @@ class wvmInstances(wvmConnect): (tcp://address) on the destination; None lets libvirt use the destination's own hostname. pools: {target: pool here} of the disks libvirt copies into those pools during the migration (disk_copies); - the other disks must be on storage both hosts share. Returns the mode - and options used, for the log, and each copied disk's path and the - path of its copy here.""" + the other disks must be on storage both hosts share. on_copy is + called with the copies ({pool, dest}: a pool here and its path) before + libvirt makes them. Returns the mode and options used, for the log, + and each copied disk's path and the path of its copy here.""" state = conn.get_status() if state == 5: mode = ["offline"] @@ -193,6 +194,8 @@ class wvmInstances(wvmConnect): # the TPM state is a file of the source host: a live migration carries it tpm_state = etree.fromstring(definitions[0]).xpath("devices/tpm/backend[@type='emulator']") if copies or (state == 5 and tpm_state): + if copies and on_copy is not None: + on_copy([{"pool": copy["pool"].name(), "dest": copy["dest"]} for copy in copies]) note = self.migrate_running(conn, state, flags, uri, copies) if note: mode.append(note) diff --git a/vrtManager/tests/test_disk_copy_migration.py b/vrtManager/tests/test_disk_copy_migration.py index 12ded116..1615929d 100644 --- a/vrtManager/tests/test_disk_copy_migration.py +++ b/vrtManager/tests/test_disk_copy_migration.py @@ -269,6 +269,14 @@ class DiskCopyTestCase(unittest.TestCase): raise libvirt.libvirtError(message) source.instance.migrate3.side_effect = migrate3 + def test_the_copies_are_told_before_libvirt_makes_them(self): + """A job records them: recovery removes them after a worker that stopped.""" + source, pool = Source(1, domain(disk(f"{IMAGES}/vm.qcow2"))), Pool("fast", FAST) + told = [] + source.instance.migrate3.side_effect = lambda *args: self.assertEqual(told, [[{"pool": "fast", "dest": f"{FAST}/vm.qcow2"}]]) + migrate(destination(pool), source, {"vda": "fast"}, on_copy=told.append) + source.instance.migrate3.assert_called_once() + def test_a_failed_migration_removes_the_partial_copies(self): source, pool = Source(1, domain(disk(f"{IMAGES}/vm.qcow2"))), Pool("fast", FAST) self.fail_after_copying(source, pool, "operation aborted: canceled by client")