j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00
Code Issues Projects Releases Wiki Activity

fix(migrate): recovery removes the disk copies of a migration whose worker stopped

Browse source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Kk4h2TcUiaWmU5MUszYSea
This commit is contained in:
catborise 2026-11-07 22:47:08 +03:00
parent d2a637cf33
commit aadf5270f2
5 changed files with 69 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

23
instances/jobs.py
View file

@ -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")))

30
instances/tests/test_migrate_job.py
View file

@ -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()

4
instances/utils.py
View file

@ -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)

11
vrtManager/instance.py
View file

@ -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)

8
vrtManager/tests/test_disk_copy_migration.py
View file

@ -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")