feat(instances): move a VM's disk to another storage pool of its host

The Disks tab has a Move button for each disk: it opens a dialog to
choose another storage pool of the same host, and the move runs as a job.

- Running or paused VM: libvirt copies the disk with a block job
  (TRANSIENT_JOB, as the VM is persistent) into a file it creates in the
  pool, then pivots. The pivot names the copy in both the running and
  the next start's definition.
- Shut-off VM: the volume is copied into the pool and the copy named in
  the definition.

The source file goes then, unless a VM or volume of any compute uses it
(shared storage holds it for every host that mounts it), a compute
cannot be read, or the delete fails: the job lists such a file under
"Left behind", and the move stands.

A disk moves when:
- the VM is defined and running, paused or shut off, without snapshots,
  a saved state or a block job;
- the disk is a writable, unshared, unencrypted raw or qcow2 file of a
  storage pool, without a backing file, and the same file in both
  definitions;
- the pool keeps disk files, is another directory, and has no file of
  that name (libvirt would write into it).
Otherwise it is refused with the reason before anything changes.

The job claims the VM and the file's name in both pools. A cancel aborts
the block copy, and ends the job cancelled once the VM is back on its
file. A failed move deletes the copy it made, unless a definition names
it (a lost reply after the pivot); a file there before the copy began is
never deleted. Recovery after a stopped worker
ends a copy left mirroring, matched by its destination, and reads the
outcome from the definitions.

Superusers only, like the other disk edits.

instances/jobs.py

@@ -923,3 +923,120 @@ class MigrateInstance(Kind):
             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.")
+
+
+DISK_DEV = re.compile(r"[a-z][a-z0-9]*")
+
+
+def _movable(user, params):
+    """The disk move's VM, for a superuser, and its plan (disk_move_plan)."""
+    if user is None or not user.is_active or not user.is_superuser:
+        raise PermissionDenied
+    instance = _instance(params["instance"])
+    return instance, _proxy(instance).disk_move_plan(params["dev"], params["pool"])
+
+
+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."""
+    for c in Compute.objects.all():
+        try:
+            if path in wvmConnect(c.hostname, c.login, c.password, c.type).paths_in_use():
+                return True
+        except libvirtError:
+            return True
+    return False
+
+
+@register
+class MoveDisk(Kind):
+    """A disk of a VM to another storage pool of its host (disk_move_plan)."""
+
+    name = "disk.move"
+    label = gettext_noop("Move disk")
+    cancellable = True  # a block copy is aborted; a shut-off VM's volume copy runs to its end
+    progress_bytes = True  # libvirt block job cur/end
+
+    def normalize(self, raw):
+        dev = str(raw.get("dev", ""))
+        if not DISK_DEV.fullmatch(dev):
+            raise util.OperationError("Choose a disk of the VM")
+        return {"instance": int(raw["instance"]), "dev": dev, "pool": str(raw.get("pool", ""))}
+
+    def validate(self, user, params):
+        _movable(user, params)
+
+    def plan(self, user, params):
+        instance, move = _movable(user, params)
+        plan = _vm_plan(params)
+        source_pool = move["volume"].storagePoolLookupByVolume().name()
+        name = os.path.basename(move["path"])
+        for pool in (source_pool, params["pool"]):
+            plan.claims[claims.volume_key(instance.compute_id, pool, name)] = Claim.EXCLUSIVE
+        return plan
+
+    def precheck(self, op, ctx):
+        instance, _move = _movable(op.user, op.params)
+        if instance.compute_id != op.intent["compute"]:
+            raise util.OperationError("The VM moved to another compute after the job was queued")
+
+    def run(self, op, ctx):
+        instance, move = _movable(op.user, op.params)
+        proxy = _proxy(instance)
+        ctx.record_intent(path=move["path"], dest=move["dest"])
+        ctx.set_phase(gettext_noop("Copying the disk"))
+        try:
+            proxy.move_disk(move)
+        except libvirtError:
+            op.refresh_from_db(fields=["cancel_requested"])
+            # cancelled only when the copy stopped and the VM kept its file
+            if op.cancel_requested and proxy.disk_named(move["dev"]) == {move["path"]} \
+                    and not (move["running"] and proxy.instance.blockJobInfo(move["dev"], 0)):
+                raise Cancelled from None
+            raise
+        # the move stands: a source file that cannot go is only left behind
+        try:
+            if not _used_anywhere(move["path"]):
+                move["volume"].delete(0)
+                return {"left_behind": []}
+        except libvirtError:
+            pass
+        return {"left_behind": [move["path"]]}
+
+    def progress(self, op):
+        info = _proxy(_instance(op.params["instance"])).instance.blockJobInfo(op.params["dev"], 0)
+        return (info["cur"], info["end"]) if info else None
+
+    def cancel(self, op):
+        dom = _proxy(_instance(op.params["instance"])).instance
+        # the worker sends a cancel once: one before the copy began waits for it
+        while dom.isActive() and not dom.blockJobInfo(op.params["dev"], 0):
+            if not Operation.objects.filter(pk=op.pk, state__in=Operation.ACTIVE).exists():
+                return
+            time.sleep(1)
+        if dom.isActive():
+            dom.blockJobAbort(op.params["dev"], 0)
+
+    def reconcile(self, op):
+        proxy = _recovery_proxy(op)
+        if proxy is None:
+            return Outcome(Operation.UNCERTAIN, error=GONE)
+        dev, path, dest = op.params["dev"], op.intent.get("path"), op.intent.get("dest")
+        dom = proxy.instance
+        # a copy no worker waits for mirrors forever: it ends here
+        if dest and dom.isActive() and dom.blockJobInfo(dev, 0):
+            mirror = etree.fromstring(dom.XMLDesc(0)).xpath(f"devices/disk[target/@dev='{dev}']/mirror/source/@file")
+            if mirror != [dest]:
+                return Outcome(Operation.UNCERTAIN, error=f"Another block job runs on {dev}; check the VM")
+            dom.blockJobAbort(dev, 0)
+        named = proxy.disk_named(dev)
+        if named == {dest}:
+            return Outcome(Operation.SUCCEEDED, result={"left_behind": [path]})
+        if named == {path}:
+            return Outcome(Operation.FAILED, error=f"The disk stayed in {path}; remove {dest} if it is there")
+        return Outcome(Operation.UNCERTAIN, error=f"The definitions of {dev} name {', '.join(sorted(map(str, named)))}; "
+                                                  "check the VM")
+
+    def log_entry(self, op, state):
+        return _log(op, state, f"move disk {op.params['dev']} to pool {op.params['pool']}")

instances/templates/instances/move_disk.html

@@ -0,0 +1,30 @@
+{% load i18n %}
+{% load bootstrap_icons %}
+{% load operations %}
+{# the Move button of a disk and its dialog: another storage pool of this host #}
+<button type="button" class="btn btn-sm btn-secondary" data-bs-toggle="modal" data-bs-target="#movedisk{{ id }}" title="{% trans "Move to another pool" %}">
+    {% bs_icon 'arrow-left-right' %}
+</button>
+<div class="modal fade" id="movedisk{{ id }}" tabindex="-1" aria-labelledby="movedisk{{ id }}-label" aria-hidden="true">
+    <div class="modal-dialog">
+        <form class="modal-content text-wrap" action="{% url 'instances:move_disk' instance.id %}" method="post">
+            {% csrf_token %}{% submission_key %}
+            <input type="hidden" name="dev" value="{{ disk.dev }}">
+            <div class="modal-header">
+                <h5 class="modal-title" id="movedisk{{ id }}-label">{% blocktrans with dev=disk.dev %}Move {{ dev }}{% endblocktrans %}</h5>
+                <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="{% trans "Close" %}"></button>
+            </div>
+            <div class="modal-body">
+                <label class="form-label" for="movedisk{{ id }}-pool">{% trans "Storage pool" %}</label>
+                <select class="form-select" name="pool" id="movedisk{{ id }}-pool" required>
+                    {% for storage in storages_host %}{% if storage != disk.storage %}<option value="{{ storage }}">{{ storage }}</option>{% endif %}{% endfor %}
+                </select>
+                <div class="form-text text-break">{% blocktrans with path=disk.path %}The disk is copied into the pool, a running VM's without stopping it, and the VM uses the copy; {{ path }} is deleted then, unless a VM or volume of any host uses it.{% endblocktrans %}</div>
+            </div>
+            <div class="modal-footer">
+                <button type="button" class="btn btn-secondary" data-bs-dismiss="modal">{% trans "Cancel" %}</button>
+                <button type="submit" class="btn btn-primary">{% trans "Move" %}</button>
+            </div>
+        </form>
+    </div>
+</div>

instances/templates/instances/settings_tab.html

@@ -312,6 +312,7 @@
                                         <input type="hidden" name="name" value="{{ disk.image }}">
                                         {% include 'instances/edit_instance_volume.html' with id=forloop.counter0 %}
                                     </form>
+                                    {% if request.user.is_superuser %}{% include 'instances/move_disk.html' with id=forloop.counter0 %}{% endif %}
                                     <form class="d-inline" action="{% url 'instances:detach_vol' instance.id %}" method="post">
                                         {% csrf_token %}
                                         <input type="hidden" name="path" value="{{ disk.path }}">

instances/tests/test_disk_move_job.py

@@ -0,0 +1,130 @@
+"""Moving a disk to another pool of its host runs as a job: the page queues
+it, the worker moves it, recovery ends a copy no worker waits for and reads
+the outcome from the VM's definitions."""
+
+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 libvirt import libvirtError
+
+from appsettings.settings import get_settings
+from computes.models import Compute
+from instances.jobs import MoveDisk
+from instances.models import Instance
+from operations.jobs import Cancelled, Context, get_kind
+from operations.models import Claim, Operation, acquire
+
+UUID = "41414141-4242-5353-6464-757575757575"
+OLD, NEW = "/var/lib/libvirt/images/vm.qcow2", "/srv/fast/vm.qcow2"
+
+
+class MoveDiskJobTestCase(TransactionTestCase):
+    def setUp(self):
+        get_settings()
+        self.compute = Compute.objects.create(name="mv-host", hostname="192.0.2.60", login="root", password="", type=1)
+        self.vm = Instance.objects.create(compute=self.compute, name="mv-vm", uuid=UUID)
+        self.admin = get_user_model().objects.create_superuser("mv_admin", "a@example.com", "pw")
+        self.client.force_login(self.admin)
+        patch.object(Compute, "status", True).start()
+        self.proxy = patch("instances.jobs.wvmInstance").start().return_value
+        self.proxy.disk_move_plan.return_value = {
+            "dev": "vda", "path": OLD, "dest": NEW, "running": True,
+            "volume": MagicMock(**{"storagePoolLookupByVolume.return_value.name.return_value": "default"})}
+        patch("instances.views.get_instance", return_value=self.vm).start()
+        self.addCleanup(patch.stopall)
+
+    def queue(self, **form):
+        self.client.post(reverse("instances:move_disk", args=[self.vm.pk]),
+                         {"dev": "vda", "pool": "fast", "submission_key": "key-move0001", **form})
+        return Operation.objects.filter(kind="disk.move").first()
+
+    def start(self, op):
+        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_page_queues_it_and_the_worker_moves_it(self):
+        op = self.queue()
+        self.assertTrue({(f"instance:{UUID}", "exclusive"), (f"volume:{self.compute.pk}:default/vm.qcow2", "exclusive"),
+                         (f"volume:{self.compute.pk}:fast/vm.qcow2", "exclusive")}
+                        <= set(Claim.objects.values_list("resource", "mode")))
+        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:
+            hosts.return_value.paths_in_use.return_value = set()
+            result = get_kind(op.kind).run(op, ctx)
+        self.proxy.move_disk.assert_called_once()
+        self.proxy.disk_move_plan.return_value["volume"].delete.assert_called_once_with(0)  # the source went
+        op.refresh_from_db()
+        self.assertEqual((op.intent["path"], op.intent["dest"], result), (OLD, NEW, {"left_behind": []}))
+
+    def test_a_source_file_any_host_may_use_is_left_behind(self):
+        """Shared storage: a VM or volume of another host may use it."""
+        Compute.objects.create(name="mv-other", hostname="192.0.2.61", login="root", password="", type=1)
+        cases = ((set(), {OLD}, None), (set(), set(), libvirtError("unreachable")),
+                 (set(), set(), "delete fails"))
+        for n, (here, other, problem) in enumerate(cases):
+            with self.subTest(case=n):
+                Operation.objects.all().delete()
+                op = self.queue(submission_key=f"key-left{n:04d}")
+                ctx = self.start(op)
+                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:
+                    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)
+                self.assertEqual(result, {"left_behind": [OLD]})
+                self.assertEqual(volume.delete.called, problem == "delete fails")
+
+    def test_only_a_superuser_with_a_disk_name_queues_it(self):
+        self.assertIsNone(self.queue(dev="vda']"))
+        self.client.force_login(get_user_model().objects.create_user("mv_user", password="pw"))
+        self.assertIsNone(self.queue())
+
+    def test_a_cancel_ends_cancelled_only_when_the_vm_kept_its_file(self):
+        op = self.queue()
+        ctx = self.start(op)
+        self.proxy.move_disk.side_effect = libvirtError("The copy of vda ended before it was complete")
+        self.proxy.instance.blockJobInfo.return_value = {}
+        self.proxy.disk_named.return_value = {OLD}
+        Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+        with self.assertRaises(Cancelled):
+            get_kind(op.kind).run(op, ctx)
+        # the pivot may have happened: the error stands, not a cancel
+        self.proxy.disk_named.return_value = {NEW}
+        with self.assertRaisesRegex(libvirtError, "ended before it was complete"):
+            get_kind(op.kind).run(op, ctx)
+
+    def reconcile(self, named, job=None, mirror=NEW):
+        op = Operation(params={"instance": self.vm.pk, "dev": "vda", "pool": "fast"},
+                       intent={"compute": self.compute.pk, "uuid": UUID, "path": OLD, "dest": NEW})
+        dom = self.proxy.instance
+        dom.isActive.return_value = True
+        dom.blockJobInfo.return_value = job or {}
+        dom.XMLDesc.return_value = (f"<domain><devices><disk><target dev='vda'/><mirror><source file='{mirror}'/>"
+                                    "</mirror></disk></devices></domain>")
+        self.proxy.disk_named.return_value = named
+        return MoveDisk().reconcile(op)
+
+    def test_recovery_reads_the_definitions(self):
+        self.assertEqual(self.reconcile({NEW}).state, Operation.SUCCEEDED)
+        failed = self.reconcile({OLD})
+        self.assertEqual((failed.state, failed.error), (Operation.FAILED, f"The disk stayed in {OLD}; remove {NEW} "
+                                                                           "if it is there"))
+        self.assertEqual(self.reconcile({OLD, NEW}).state, Operation.UNCERTAIN)
+        self.proxy.instance.blockJobAbort.assert_not_called()
+
+    def test_recovery_ends_the_copy_no_worker_waits_for(self):
+        self.assertEqual(self.reconcile({OLD}, job={"cur": 2, "end": 2}).state, Operation.FAILED)
+        self.proxy.instance.blockJobAbort.assert_called_once_with("vda", 0)
+        # a block job of something else is not ended
+        self.proxy.instance.blockJobAbort.reset_mock()
+        self.assertEqual(self.reconcile({OLD}, job={"cur": 1, "end": 2}, mirror="/other").state, Operation.UNCERTAIN)
+        self.proxy.instance.blockJobAbort.assert_not_called()

instances/tests/test_live_jobs.py

@@ -197,6 +197,39 @@ class LiveJobsTestCase(TransactionTestCase):
         self.assertEqual((livetest.volume_exists(self.conn, path), livetest.volume_exists(dest_conn, path)),
                          (True, True))
 
+    def test_a_disk_moves_to_another_pool_and_back(self):
+        """There while the VM runs (block copy and pivot), back shut off
+        (volume copy); each time the source file goes."""
+        other = self.conn.storagePoolDefineXML(
+            f"<pool type='dir'><name>{P}move</name><target><path>{livetest.POOL_PATH}-move</path></target></pool>")
+        other.create(libvirt.VIR_STORAGE_POOL_CREATE_WITH_BUILD)
+
+        def remove_other():
+            other.refresh(0)
+            for vol in other.listAllVolumes():
+                vol.delete(0)
+            other.destroy()
+            other.delete(0)
+            other.undefine()
+        self.addCleanup(remove_other)
+        path = livetest.create_volume(self.conn, P + "move")
+        dom = livetest.define_vm(self.conn, P + "move", [path])
+        dom.create()
+        instance = Instance.objects.create(compute=self.compute, name=dom.name(), uuid=dom.UUIDString())
+        moved = f"{livetest.POOL_PATH}-move/{P}move.qcow2"
+
+        def source(flags):
+            return dom.XMLDesc(flags).split("<source file='")[1].split("'")[0]
+
+        self.run_job(jobs.submit("disk.move", self.admin, {"instance": instance.pk, "dev": "vda", "pool": P + "move"},
+                                 "key-move0001"))
+        self.assertEqual((source(0), source(libvirt.VIR_DOMAIN_XML_INACTIVE)), (moved, moved))
+        self.assertFalse(livetest.volume_exists(self.conn, path))  # no host uses the source file: it went
+        dom.destroy()
+        self.run_job(jobs.submit("disk.move", self.admin, {"instance": instance.pk, "dev": "vda", "pool": livetest.POOL},
+                                 "key-move0002"))
+        self.assertEqual(source(libvirt.VIR_DOMAIN_XML_INACTIVE), path)
+
     def test_a_shut_off_uefi_vm_migrates_with_its_nvram_there_and_back(self):
         dest, _dest_conn = self.second_host()
         hosts = {self.compute.pk: wvmConnect(*self.args), dest.pk: wvmConnect(dest.hostname, dest.login, "", 2)}

instances/urls.py

@@ -59,6 +59,7 @@ urlpatterns = [
     path("<int:pk>/delete_network/", views.delete_network, name="delete_network"),
     path("<int:pk>/host_devices/", views.host_devices, name="host_devices"),
     path("<int:pk>/migrate_plan/", views.migrate_plan, name="migrate_plan"),
+    path("<int:pk>/move_disk/", views.move_disk, name="move_disk"),
     path("<int:pk>/add_usb_hostdev/", views.add_usb_hostdev, name="add_usb_hostdev"),
     path("<int:pk>/delete_usb_hostdev/", views.delete_usb_hostdev, name="delete_usb_hostdev"),
     path("<int:pk>/add_pci_hostdev/", views.add_pci_hostdev, name="add_pci_hostdev"),

instances/views.py

@@ -866,6 +866,22 @@ def migrate(request, pk):
     return get_safe_redirect(request, default=reverse("instances:instance", args=[pk]))
 
 
+@require_POST
+@superuser_only
+def move_disk(request, pk):
+    get_instance(request.user, pk)
+    raw = {"instance": pk, "dev": request.POST.get("dev", ""), "pool": request.POST.get("pool", "")}
+    try:
+        op = jobs.submit("disk.move", request.user, raw, request.POST.get("submission_key", ""))
+    except claims.Busy as busy:
+        return _busy(request, pk, busy)
+    except (jobs.SubmissionConflict, libvirtError) as err:
+        messages.error(request, error_text(request, err))
+    else:
+        jobs.remember_job(request, op)
+    return _back(request, pk, "disks")
+
+
 @superuser_only
 def migrate_plan(request, pk):
     """The disks of the Migrate tab for the chosen destination, loaded when

static/icon_cache/arrow-left-right_None_None_None.svg

@@ -0,0 +1,8 @@
+<?xml version="1.0" ?>
+<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" fill="currentColor" class="bi bi-arrow-left-right" viewBox="0 0 16 16">
+	
+  
+	<path fill-rule="evenodd" d="M1 11.5a.5.5 0 0 0 .5.5h11.793l-3.147 3.146a.5.5 0 0 0 .708.708l4-4a.5.5 0 0 0 0-.708l-4-4a.5.5 0 0 0-.708.708L13.293 11H1.5a.5.5 0 0 0-.5.5m14-7a.5.5 0 0 1-.5.5H2.707l3.147 3.146a.5.5 0 1 1-.708.708l-4-4a.5.5 0 0 1 0-.708l4-4a.5.5 0 1 1 .708.708L2.707 4H14.5a.5.5 0 0 1 .5.5"/>
+	
+
+</svg>

vrtManager/connection.py

@@ -921,6 +921,16 @@ class wvmConnect(object):
                 return pool
         return None
 
+    def file_pools(self):
+        """The active pools of this host that keep disks as files, a copy's
+        destination: {name: (pool, directory)}."""
+        pools = {}
+        for pool in self.wvm.listAllStoragePools(libvirt.VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
+            tree = etree.fromstring(pool.XMLDesc(0))
+            if tree.get("type") in ("dir", "fs", "netfs") and tree.findtext("target/path"):
+                pools[pool.name()] = (pool, tree.findtext("target/path").rstrip("/"))
+        return pools
+
     @contextlib.contextmanager
     def _nvram_pool(self):
         """An active, refreshed pool on NVRAM_DIR: libvirt reads and deletes

vrtManager/instance.py

@@ -15,6 +15,7 @@ try:
         VIR_DOMAIN_AFFECT_CONFIG,
         VIR_DOMAIN_AFFECT_LIVE,
         VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP,
+        VIR_DOMAIN_PAUSED,
         VIR_DOMAIN_RUNNING,
         VIR_DOMAIN_SHUTOFF,
         VIR_DOMAIN_START_PAUSED,
@@ -26,6 +27,8 @@ try:
         VIR_ERR_NO_STORAGE_POOL,
         VIR_ERR_NO_STORAGE_VOL,
         VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE,
+        VIR_DOMAIN_BLOCK_COPY_TRANSIENT_JOB,
+        VIR_DOMAIN_BLOCK_JOB_ABORT_PIVOT,
         VIR_DOMAIN_BLOCK_RESIZE_BYTES,
         VIR_STORAGE_VOL_FILE,
         VIR_MIGRATE_AUTO_CONVERGE,
@@ -288,16 +291,6 @@ class wvmInstances(wvmConnect):
             return "a disk with a backing file is not copied"
         return None
 
-    def file_pools(self):
-        """The active pools of this host that keep disks as files, a copy's
-        destination: {name: (pool, directory)}."""
-        pools = {}
-        for pool in self.wvm.listAllStoragePools(VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
-            tree = ElementTree.fromstring(pool.XMLDesc(0))
-            if tree.get("type") in ("dir", "fs", "netfs") and tree.findtext("target/path"):
-                pools[pool.name()] = (pool, tree.findtext("target/path").rstrip("/"))
-        return pools
-
     def disk_copies(self, conn, definitions, pools):
         """The copies of the disks pools names a pool here for: {target, path,
         dest, pool}, from the first of definitions (XML, the one the migration
@@ -2107,6 +2100,125 @@ class wvmInstance(wvmConnect):
                 jobs.append({"dev": dev, "cur": info["cur"], "end": info["end"]})
         return jobs
 
+    def disk_move_plan(self, dev, pool_name):
+        """Where the disk dev of the VM goes in the pool pool_name of this
+        host: {dev, path, dest, format, pool, volume, running}. Refused with
+        the reason unless the VM is persistent, running, paused or shut off,
+        without snapshots, a saved state or a block job; the disk a raw or
+        qcow2 file of a storage pool, without a backing file, not read-only,
+        shareable or encrypted, the same file in both definitions; the pool
+        one of disk files, another directory, without a file of that name."""
+        dom = self.instance
+        state = dom.info()[0]
+        if not dom.isPersistent() or state not in (VIR_DOMAIN_RUNNING, VIR_DOMAIN_PAUSED, VIR_DOMAIN_SHUTOFF):
+            raise util.OperationError("A disk moves when its VM is defined and running, paused or shut off")
+        if dom.snapshotNum(0) or dom.hasManagedSaveImage(0) or self.get_snapshot_jobs():
+            raise util.OperationError("A disk of a VM with snapshots, a saved state or a running block job does not move")
+        # the running definition (the persistent one of a shut-off VM) first
+        disks = [ElementTree.fromstring(xml).find(f"devices/disk/target[@dev='{dev}']/..")
+                 for xml in (self._XMLDesc(0), self._XMLDesc(PERSISTENT_XML))]
+        sources = [disk.find("source") if disk is not None else None for disk in disks]
+        disk, source = disks[0], sources[0]
+        if disk is None or disk.get("type") != "file" or disk.get("device") != "disk" or None in sources \
+                or len({s.get("file") for s in sources}) != 1:
+            raise util.OperationError(f"{dev} is not a disk file of the VM, the same in its running and next start's "
+                                      "definition")
+        driver = disk.find("driver")
+        fmt = driver.get("type") if driver is not None else None
+        if fmt not in ("raw", "qcow2") or disk.find("readonly") is not None or disk.find("shareable") is not None \
+                or disk.find("encryption") is not None or source.find("encryption") is not None:
+            raise util.OperationError(f"{dev} does not move: only writable raw and qcow2 disks do")
+        path = source.get("file")
+        try:
+            volume = self.wvm.storageVolLookupByPath(path)
+        except libvirtError as err:
+            if err.get_error_code() != VIR_ERR_NO_STORAGE_VOL:
+                raise
+            raise util.OperationError(f"{dev} does not move: no storage pool has {path}") from err
+        if ElementTree.fromstring(volume.XMLDesc(0)).findtext("backingStore/path"):
+            raise util.OperationError(f"{dev} does not move: it has a backing file")
+        pools = self.file_pools()
+        if pool_name not in pools:
+            raise util.OperationError(f"{pool_name} is not an active pool of disk files of this host")
+        pool, directory = pools[pool_name]
+        if directory == os.path.dirname(path):
+            raise util.OperationError(f"{dev} is in pool {pool_name} already")
+        pool.refresh(0)
+        name = os.path.basename(path)
+        if self._find_volume(pool, name) is not None:
+            raise util.OperationError(f"Pool {pool_name} already has a file {name}")
+        return {"dev": dev, "path": path, "dest": f"{directory}/{name}", "format": fmt, "pool": pool,
+                "volume": volume, "running": state != VIR_DOMAIN_SHUTOFF}
+
+    def move_disk(self, move):
+        """Moves the disk as disk_move_plan planned it. A running or paused
+        VM's is copied by a block job (TRANSIENT_JOB: the VM is persistent)
+        into a file libvirt creates; at the pivot libvirt names the copy in
+        both definitions. A shut-off VM's volume is copied into the pool and
+        the copy named in the definition. The source file stays (see
+        MoveDisk). A failed move deletes the copy it made, unless a
+        definition names it; a file there before the copy began is never
+        ours."""
+        dom, dev, dest = self.instance, move["dev"], move["dest"]
+        made = False
+        try:
+            if move["running"]:
+                xml = (f"<disk type='file'><driver type='{move['format']}'/>"
+                       f"<source file='{util.xml_escape(dest)}'/></disk>")
+                dom.blockCopy(dev, xml, None, VIR_DOMAIN_BLOCK_COPY_TRANSIENT_JOB)
+                made = True
+                while True:  # mirrored once the copy caught up: cur == end
+                    info = dom.blockJobInfo(dev, 0)
+                    if not info:
+                        raise util.OperationError(f"The copy of {dev} ended before it was complete")
+                    if info["end"] and info["cur"] == info["end"]:
+                        break
+                    time.sleep(1)
+                dom.blockJobAbort(dev, VIR_DOMAIN_BLOCK_JOB_ABORT_PIVOT)
+                move["pool"].refresh(0)  # libvirt made the file past the pool: it lists it now
+            else:
+                move["pool"].createXMLFrom(
+                    f"<volume><name>{util.xml_escape(os.path.basename(dest))}</name>"
+                    f"<capacity>{move['volume'].info()[1]}</capacity>"
+                    f"<target><format type='{move['format']}'/></target></volume>", move["volume"], 0)
+                made = True
+                tree = ElementTree.fromstring(self._XMLDesc(PERSISTENT_XML))
+                tree.find(f"devices/disk/target[@dev='{dev}']/../source").set("file", dest)
+                self._defineXML(ElementTree.tostring(tree, encoding="unicode"))
+        except libvirtError as error:
+            if made:
+                self.remove_disk_copy(move, error)
+            raise
+
+    def disk_named(self, dev):
+        """The files the running (if active) and next start's definitions name for dev."""
+        flags = [0, PERSISTENT_XML] if self.instance.isActive() else [PERSISTENT_XML]
+        files = set()
+        for flag in flags:
+            disk = ElementTree.fromstring(self.instance.XMLDesc(flag)).find(f"devices/disk/target[@dev='{dev}']/..")
+            source = disk.find("source") if disk is not None else None
+            files.add(source.get("file") if source is not None else None)
+        return files
+
+    def remove_disk_copy(self, move, error):
+        """Deletes the copy of a failed move once nothing uses it: no block
+        job runs on the disk and no definition names it. Otherwise it stays,
+        and the error says so."""
+        dev, dest = move["dev"], move["dest"]
+        try:
+            if move["running"] and self.instance.blockJobInfo(dev, 0):
+                self.instance.blockJobAbort(dev, 0)
+            named = dest in self.disk_named(dev)
+            if not named:
+                move["pool"].refresh(0)
+                copy = self._find_volume(move["pool"], os.path.basename(dest))
+                if copy is not None:
+                    copy.delete(0)
+        except libvirtError as cleanup_error:
+            raise util.OperationError(f"{error}; remove {dest} if it is there: {cleanup_error}") from error
+        if named:
+            raise util.OperationError(f"{error}; the VM may use {dest} now: check its disk {dev}") from error
+
     def _snapshot(self, name, external):
         """The named snapshot; a snapshot of the other kind is refused."""
         flag = VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL if external else VIR_DOMAIN_SNAPSHOT_LIST_INTERNAL

vrtManager/tests/test_disk_move.py

@@ -0,0 +1,156 @@
+"""A disk moves to another storage pool of its host: a running or paused
+VM's by a block copy and a pivot, a shut-off VM's by a volume copy and a
+redefinition; the source file stays. Disks that cannot move are refused
+before anything changes, and a failed move deletes the copy unless a
+definition names it."""
+
+import unittest
+from unittest import mock
+
+import libvirt
+from django.conf import settings
+
+if not settings.configured:
+    settings.configure(MAC_OUI="52:54:10")
+
+from vrtManager import util
+from vrtManager.instance import wvmInstance
+
+IMAGES = "/var/lib/libvirt/images"
+FAST = "/srv/fast"
+
+
+def error(code, message="error"):
+    err = libvirt.libvirtError(message)
+    err.err = (code, 0, message, 2, None, None, None, 0, 0)
+    return err
+
+
+def domain(path=f"{IMAGES}/vm.qcow2", fmt="qcow2", extra=""):
+    return (f"<domain><devices><disk type='file' device='disk'><driver name='qemu' type='{fmt}'/>"
+            f"<source file='{path}'/><target dev='vda'/>{extra}</disk></devices></domain>")
+
+
+class Pool:
+    def __init__(self, files=()):
+        self.files = set(files)
+        self.deleted = []
+
+    refreshes = 0
+
+    def refresh(self, flags=0):
+        self.refreshes += 1
+
+    def storageVolLookupByName(self, name):
+        if name not in self.files:
+            raise error(libvirt.VIR_ERR_NO_STORAGE_VOL)
+        volume = mock.Mock()
+        volume.delete.side_effect = lambda flags: self.deleted.append(name)
+        return volume
+
+
+def vm(state=1, xml=None, config=None, backing=None, fast=None, snapshots=0):
+    """A wvmInstance of a VM in state whose vda is in IMAGES; pool fast is on FAST."""
+    proxy = wvmInstance.__new__(wvmInstance)
+    proxy.instance = dom = mock.Mock()
+    dom.info.return_value = [state, 0, 0, 1, 0]
+    dom.isPersistent.return_value = True
+    dom.isActive.return_value = state != 5
+    dom.snapshotNum.return_value = snapshots
+    dom.hasManagedSaveImage.return_value = 0
+    dom.blockJobInfo.return_value = {}
+    xml = xml or domain()
+    dom.XMLDesc.side_effect = lambda flags: (config or xml) if flags & libvirt.VIR_DOMAIN_XML_INACTIVE else xml
+    proxy.wvm = mock.Mock()
+    volume = proxy.wvm.storageVolLookupByPath.return_value
+    volume.info.return_value = [0, 1 << 30, 1 << 20]
+    volume.XMLDesc.return_value = f"<volume><backingStore><path>{backing}</path></backingStore></volume>" \
+        if backing else "<volume/>"
+    proxy.fast = fast or Pool()
+    proxy.file_pools = lambda: {"fast": (proxy.fast, FAST), "default": (Pool(), IMAGES)}
+    proxy.get_snapshot_jobs = lambda: []
+    return proxy
+
+
+class DiskMovePlanTestCase(unittest.TestCase):
+    def test_a_disk_file_moves_into_another_pool(self):
+        move = vm().disk_move_plan("vda", "fast")
+        self.assertEqual((move["path"], move["dest"], move["format"], move["running"]),
+                         (f"{IMAGES}/vm.qcow2", f"{FAST}/vm.qcow2", "qcow2", True))
+
+    def test_refusals(self):
+        cases = (
+            (vm(state=4), "fast", "defined and running, paused or shut off"),
+            (vm(snapshots=1), "fast", "snapshots"),
+            (vm(xml=domain(extra="<readonly/>")), "fast", "only writable raw and qcow2"),
+            (vm(xml=domain(fmt="vmdk")), "fast", "only writable raw and qcow2"),
+            (vm(config=domain(path=f"{IMAGES}/next.qcow2")), "fast", "the same in its running and next start"),
+            (vm(backing=f"{IMAGES}/base.qcow2"), "fast", "backing file"),
+            (vm(), "rbd", "not an active pool of disk files"),
+            (vm(), "default", "in pool default already"),
+            (vm(fast=Pool(files={"vm.qcow2"})), "fast", "Pool fast already has a file vm.qcow2"),
+        )
+        for proxy, pool, reason in cases:
+            with self.subTest(reason=reason):
+                with self.assertRaisesRegex(util.OperationError, reason):
+                    proxy.disk_move_plan("vda", pool)
+
+
+class MoveDiskTestCase(unittest.TestCase):
+    def test_a_running_vm_is_copied_and_pivoted(self):
+        proxy = vm()
+        proxy.instance.blockJobInfo.side_effect = [{"cur": 1, "end": 2}, {"cur": 2, "end": 2}]
+        with mock.patch("vrtManager.instance.time.sleep"):
+            proxy.move_disk(proxy.disk_move_plan("vda", "fast"))
+        dev, xml, params, flags = proxy.instance.blockCopy.call_args.args
+        self.assertEqual((dev, flags), ("vda", libvirt.VIR_DOMAIN_BLOCK_COPY_TRANSIENT_JOB))
+        self.assertIn(f"<source file='{FAST}/vm.qcow2'/>", xml)
+        proxy.instance.blockJobAbort.assert_called_once_with("vda", libvirt.VIR_DOMAIN_BLOCK_JOB_ABORT_PIVOT)
+        self.assertEqual(proxy.fast.refreshes, 2)  # the plan's, and one so that the pool lists the copy
+
+    def test_a_shut_off_vm_gets_a_copy_and_a_new_definition(self):
+        proxy = vm(state=5)
+        proxy.fast.createXMLFrom = mock.Mock()
+        proxy.move_disk(proxy.disk_move_plan("vda", "fast"))
+        volume_xml, source, flags = proxy.fast.createXMLFrom.call_args.args
+        self.assertIn("<name>vm.qcow2</name><capacity>1073741824</capacity>", volume_xml)
+        self.assertIn(f'<source file="{FAST}/vm.qcow2" />', proxy.wvm.defineXML.call_args.args[0])
+        proxy.instance.blockCopy.assert_not_called()
+
+    def test_a_copy_that_ends_early_is_deleted(self):
+        proxy = vm()
+        proxy.instance.blockJobInfo.return_value = {}  # aborted by a cancel
+        move = proxy.disk_move_plan("vda", "fast")
+        proxy.fast.files.add("vm.qcow2")  # libvirt made it
+        with self.assertRaisesRegex(util.OperationError, "ended before it was complete"):
+            proxy.move_disk(move)
+        self.assertEqual(proxy.fast.deleted, ["vm.qcow2"])
+
+    def test_a_copy_a_definition_names_is_kept(self):
+        """The pivot may have happened before its reply was lost."""
+        proxy = vm()
+        proxy.instance.blockJobInfo.side_effect = [{"cur": 2, "end": 2}, {}]
+        proxy.instance.blockJobAbort.side_effect = libvirt.libvirtError("Cannot recv data")
+        move = proxy.disk_move_plan("vda", "fast")
+        proxy.instance.XMLDesc.side_effect = lambda flags: domain(path=f"{FAST}/vm.qcow2")
+        proxy.fast.files.add("vm.qcow2")
+        with self.assertRaisesRegex(util.OperationError, f"the VM may use {FAST}/vm.qcow2 now"):
+            proxy.move_disk(move)
+        self.assertEqual(proxy.fast.deleted, [])
+
+    def test_a_file_made_by_someone_else_is_never_deleted(self):
+        """It appeared after the plan, and the copy failed on it."""
+        proxy = vm(state=5)
+        move = proxy.disk_move_plan("vda", "fast")
+        proxy.fast.files.add("vm.qcow2")
+        proxy.fast.createXMLFrom = mock.Mock(side_effect=error(libvirt.VIR_ERR_STORAGE_VOL_EXIST))
+        with self.assertRaises(libvirt.libvirtError):
+            proxy.move_disk(move)
+        self.assertEqual(proxy.fast.deleted, [])
+        running = vm()
+        move = running.disk_move_plan("vda", "fast")
+        running.fast.files.add("vm.qcow2")
+        running.instance.blockCopy.side_effect = error(libvirt.VIR_ERR_OPERATION_FAILED)
+        with self.assertRaises(libvirt.libvirtError):
+            running.move_disk(move)
+        self.assertEqual(running.fast.deleted, [])