From 826eaac813f59b8a607a46e7e5db3cd83e2722f2 Mon Sep 17 00:00:00 2001 From: catborise Date: Sat, 7 Nov 2026 22:40:54 +0300 Subject: [PATCH] fix(migrate): a copied disk's source file is deleted only when no compute may use it The source host alone was asked: a file on storage the computes share, used by a VM or volume of another compute, was deleted under it. The check the disk move makes, every compute and a host that cannot tell counting as using it, moves to instances.utils and serves both. The destination is not asked when its copy has the source's path: its VM uses that copy, another file. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Kk4h2TcUiaWmU5MUszYSea --- instances/jobs.py | 32 +----------- instances/tests/test_disk_move_job.py | 10 ++-- instances/tests/test_migration_log.py | 20 ++++++-- instances/utils.py | 51 +++++++++++++++----- vrtManager/instance.py | 5 +- vrtManager/tests/test_disk_copy_migration.py | 2 +- 6 files changed, 66 insertions(+), 54 deletions(-) diff --git a/instances/jobs.py b/instances/jobs.py index ea5f2184..5d5639ed 100644 --- a/instances/jobs.py +++ b/instances/jobs.py @@ -3,7 +3,6 @@ import math import os import re -import threading import time from django.db import transaction @@ -1149,35 +1148,6 @@ def _movable(user, params): return instance, _proxy(instance).disk_move_plan(params["dev"], params["pool"]) -USED_CHECK_WAIT = 60 # seconds a disk move waits for the computes to tell whether its old file is used - - -def _used_anywhere(path): - """Whether a VM or volume of any compute uses path (paths_in_use), as - shared storage holds it for every host that mounts it; a compute that - cannot be read might, and so might one that does not answer within - USED_CHECK_WAIT: libvirt has no call timeout, and a stuck host would - hold the job for as long as it is stuck.""" - computes = list(Compute.objects.all()) - answer = [] - - def check(): - for c in computes: - try: - if path in wvmConnect(c.hostname, c.login, c.password, c.type).paths_in_use(): - answer.append(True) - return - except libvirtError: - answer.append(True) - return - answer.append(False) - - thread = threading.Thread(target=check, daemon=True) - thread.start() - thread.join(USED_CHECK_WAIT) - return answer[0] if answer else True - - @register class MoveDisk(Kind): """A disk of a VM to another storage pool of its host (disk_move_plan).""" @@ -1226,7 +1196,7 @@ class MoveDisk(Kind): raise # the move stands: a source file that cannot go is only left behind try: - if not _used_anywhere(move["path"]): + if not utils.used_anywhere(move["path"]): move["volume"].delete(0) return {"left_behind": []} except libvirtError: diff --git a/instances/tests/test_disk_move_job.py b/instances/tests/test_disk_move_job.py index 7a421049..ced10689 100644 --- a/instances/tests/test_disk_move_job.py +++ b/instances/tests/test_disk_move_job.py @@ -58,7 +58,7 @@ class MoveDiskJobTestCase(TransactionTestCase): self.proxy.move_disk.assert_not_called() # nothing ran in the request ctx = self.start(op) get_kind(op.kind).precheck(op, ctx) - with patch("instances.jobs.wvmConnect") as hosts: + with patch("instances.utils.wvmConnect") as hosts: hosts.return_value.paths_in_use.return_value = set() result = get_kind(op.kind).run(op, ctx) self.proxy.move_disk.assert_called_once() @@ -79,7 +79,7 @@ class MoveDiskJobTestCase(TransactionTestCase): volume = self.proxy.disk_move_plan.return_value["volume"] volume.delete.reset_mock() volume.delete.side_effect = libvirtError("busy") if problem == "delete fails" else None - with patch("instances.jobs.wvmConnect") as hosts: + with patch("instances.utils.wvmConnect") as hosts: reads = [here, other] if not isinstance(problem, libvirtError) else [here, problem] hosts.return_value.paths_in_use.side_effect = reads result = get_kind(op.kind).run(op, ctx) @@ -135,13 +135,13 @@ class MoveDiskJobTestCase(TransactionTestCase): class UsedAnywhereTestCase(TransactionTestCase): def test_a_host_that_does_not_answer_counts_as_using_the_file(self): - from instances import jobs + from instances import utils Compute.objects.create(name="stuck", hostname="192.0.2.62", login="root", password="", type=1) stuck = threading.Event() self.addCleanup(stuck.set) - with patch("instances.jobs.wvmConnect") as hosts, patch.object(jobs, "USED_CHECK_WAIT", 0.2): + with patch("instances.utils.wvmConnect") as hosts, patch.object(utils, "USED_CHECK_WAIT", 0.2): hosts.return_value.paths_in_use.side_effect = lambda: stuck.wait(10) and set() started = time.monotonic() - self.assertTrue(jobs._used_anywhere(OLD)) + self.assertTrue(utils.used_anywhere(OLD)) self.assertLess(time.monotonic() - started, 2) diff --git a/instances/tests/test_migration_log.py b/instances/tests/test_migration_log.py index 9bc44a97..747ee423 100644 --- a/instances/tests/test_migration_log.py +++ b/instances/tests/test_migration_log.py @@ -34,18 +34,21 @@ class MigrationLogTests(TestCase): self.proxy.remove_nvram.side_effect = remove_nvram else: self.proxy.remove_nvram.return_value = remove_nvram - self.proxy.paths_in_use.return_value = set(in_use) if delete_fails: self.proxy.get_volume_by_path.return_value.delete.side_effect = delete_fails with patch.object(Compute, "status", True), \ patch("instances.utils.wvmInstances") as instances, \ patch("instances.utils.wvmInstance"), \ + patch("instances.utils.wvmConnect") as hosts, \ patch.object(Instance, "proxy", self.proxy), \ patch.object(Instance, "autostart", False): + hosts.return_value.paths_in_use.return_value = set(in_use) if isinstance(outcome, Exception): instances.return_value.moveto.side_effect = outcome else: - instances.return_value.moveto.return_value = (outcome, list(copied)) + # each copy at the same path on the destination, unless a pair names its own + pairs = [item if isinstance(item, tuple) else (item, item) for item in copied] + instances.return_value.moveto.return_value = (outcome, pairs) return migrate_instance(self.dest, self.vm, self.admin, **request) def test_the_nvram_file_left_on_the_source_is_removed(self): @@ -128,6 +131,17 @@ class MigrationLogTests(TestCase): self.vm.refresh_from_db() self.assertEqual(self.vm.compute, self.dest) + def test_the_destination_is_asked_only_when_its_copy_has_another_path(self): + """Its VM uses the copy: at the same path there, that is another file.""" + path = "/var/lib/libvirt/images/vm.qcow2" + with patch("instances.utils.used_anywhere", return_value=False) as used: + self.migrate("live, disks copied: vda", copied=[path], live=True, delete_source=True) + self.assertEqual(used.call_args.kwargs, {"skip": self.dest.pk}) + self.vm.compute = self.source + self.vm.save() + self.migrate("live, disks copied: vda", copied=[(path, "/srv/fast/vm.qcow2")], live=True, delete_source=True) + self.assertEqual(used.call_args.kwargs, {"skip": None}) + def test_a_source_copy_another_vm_or_volume_uses_stays(self): disks = ["/var/lib/libvirt/images/vm.qcow2", "/var/lib/libvirt/images/base.qcow2"] self.migrate("live, disks copied: vda, vdb", copied=disks, in_use={disks[1]}, live=True, @@ -135,5 +149,5 @@ class MigrationLogTests(TestCase): self.proxy.get_volume_by_path.assert_called_once_with(disks[0]) self.assertEqual(rows()[-2:], [ ("VM vm: delete the source copy /var/lib/libvirt/images/vm.qcow2", True), - ("VM vm: keep the source copy /var/lib/libvirt/images/base.qcow2: another VM or volume uses it", True), + ("VM vm: keep the source copy /var/lib/libvirt/images/base.qcow2: a VM or volume of a compute may use it", True), ]) diff --git a/instances/utils.py b/instances/utils.py index 07f7f4d2..3314cb25 100644 --- a/instances/utils.py +++ b/instances/utils.py @@ -2,6 +2,7 @@ import os import random import re import string +import threading from collections import namedtuple from accounts.models import UserInstance, UserAttributes @@ -70,6 +71,35 @@ def paths_used_on_other_computes(compute, own): return used & set(own) +USED_CHECK_WAIT = 60 # seconds a disk move or a migration waits for the computes to tell whether its old file is used + + +def used_anywhere(path, skip=None): + """Whether a VM or volume of any compute uses path (paths_in_use), as + shared storage holds it for every host that mounts it; a compute that + cannot be read might, and so might one that does not answer within + USED_CHECK_WAIT: libvirt has no call timeout, and a stuck host would + hold the job for as long as it is stuck. skip: a compute not asked.""" + computes = list(Compute.objects.exclude(pk=skip)) + answer = [] + + def check(): + for c in computes: + try: + if path in wvmConnect(c.hostname, c.login, c.password, c.type).paths_in_use(): + answer.append(True) + return + except libvirtError: + answer.append(True) + return + answer.append(False) + + thread = threading.Thread(target=check, daemon=True) + thread.start() + thread.join(USED_CHECK_WAIT) + return answer[0] if answer else True + + def disk_gib(disks): """The size of the disks in whole GiB each, as the quota counts them.""" return sum(int(disk["size"] or 0) >> 30 for disk in disks) @@ -225,8 +255,9 @@ def migrate_instance( 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 another VM or - volume uses them, and are named in the log otherwise. Returns the mode.""" + pools); 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) def log(action, error=None): @@ -285,16 +316,12 @@ def migrate_instance( if conn_new is not None: conn_new.close() - in_use = set() - if copied and delete_source: - try: - in_use = instance.proxy.paths_in_use() - except libvirtError as error: - log("check the source copies before deleting them", error) - delete_source = False - for path in copied: - if not delete_source or path in in_use: - log(f"keep the source copy {path}" + (": another VM or volume uses it" if path in in_use else "")) + for path, copy in copied: + # shared storage: a VM or volume of any compute may use it; the destination's + # VM uses the copy, another file when it has the same path there + used = delete_source and used_anywhere(path, skip=new_compute.pk if copy == path else None) + if not delete_source or used: + log(f"keep the source copy {path}" + (": a VM or volume of a compute may use it" if used else "")) continue try: instance.proxy.get_volume_by_path(path).delete(0) diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 27417452..f783dd78 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -129,7 +129,8 @@ class wvmInstances(wvmConnect): 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 the paths of the copied disks.""" + 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"] @@ -203,7 +204,7 @@ class wvmInstances(wvmConnect): self.move_definition(dom, conn) else: dom.migrate(self.wvm, flags, None, uri, 0) - return ", ".join(mode), [copy["path"] for copy in copies] + return ", ".join(mode), [(copy["path"], copy["dest"]) for copy in copies] def migrate_running(self, conn, state, flags, uri, copies): """Migrates the VM of conn as a running VM, which carries its TPM diff --git a/vrtManager/tests/test_disk_copy_migration.py b/vrtManager/tests/test_disk_copy_migration.py index 16d82eb6..dff4be43 100644 --- a/vrtManager/tests/test_disk_copy_migration.py +++ b/vrtManager/tests/test_disk_copy_migration.py @@ -135,7 +135,7 @@ class DiskCopyTestCase(unittest.TestCase): dest = destination(Pool("default", IMAGES), Pool("fast", FAST)) mode, copied = migrate(dest, source, {"vda": "fast", "vdb": "default"}) self.assertEqual(mode, "live, disks copied: vda to fast, vdb to default") - self.assertEqual(copied, [f"{IMAGES}/vm.qcow2", f"{IMAGES}/data.raw"]) + self.assertEqual(copied, [(f"{IMAGES}/vm.qcow2", f"{FAST}/vm.qcow2"), (f"{IMAGES}/data.raw", f"{IMAGES}/data.raw")]) params, flags = source.instance.migrate3.call_args.args[1:] self.assertEqual(params[libvirt.VIR_MIGRATE_PARAM_MIGRATE_DISKS], ["vda", "vdb"]) self.assertEqual(params[libvirt.VIR_MIGRATE_PARAM_URI], "tcp://192.0.2.9")