feat(instances): a running VM can be cloned

libvirt's backup copies the disks of a running or paused VM from one
instant into new volumes while it keeps running; the clone is defined
shut off as before. The VM is held for the copy, the copy can be
cancelled, and recovery reads the job's records and libvirt.

instances/jobs.py

@@ -8,7 +8,7 @@ from django.core.exceptions import PermissionDenied
 from django.utils.translation import gettext_noop
 from django.utils.translation import gettext_noop as _
 from libvirt import (VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE, VIR_DOMAIN_JOB_NONE, VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL,
-                     VIR_ERR_NO_DOMAIN, libvirtError)
+                     VIR_ERR_NO_DOMAIN, VIR_ERR_NO_STORAGE_VOL, libvirtError)
 from vrtManager import util
 from vrtManager.connection import wvmConnect
 from vrtManager.create import wvmCreate
@@ -391,11 +391,39 @@ class RevertSnapshot(Kind):
         return _log(op, state, f"Revert {kind}: {op.params['name']}")
 
 
+def _delete_created(proxy, op):
+    """The volumes this job made; nothing else."""
+    for entry in op.intent.get("created", []):
+        try:
+            proxy.get_volume_by_path(entry["path"]).delete(0)
+        except libvirtError as err:
+            if err.get_error_code() != VIR_ERR_NO_STORAGE_VOL:
+                raise
+
+
+CLONE_STATE_REFUSED = "The VM must be running, paused or shut off to be cloned"
+THAW_FAILED = "The guest's file systems could not be thawed ({error}); thaw them with: virsh domfsthaw {vm}"
+
+
+def _check_clone_state(proxy, data):
+    """A shut-off VM is copied as it is; a running or paused one only when a
+    live clone can work; other states are refused."""
+    status = proxy.get_status()
+    if status in (1, 3):
+        proxy.live_clone_targets(data, **_snapshot_free_space())
+    elif status != 5:
+        raise util.OperationError(CLONE_STATE_REFUSED)
+
+
 @register
 class CloneInstance(Kind):
+    """A shut-off VM is copied volume by volume; a running or paused one with
+    libvirt's push backup, which copies its disks from one instant."""
+
     name = "instance.clone"
     label = gettext_noop("Clone VM")
-    time_limit = 12 * 3600  # a copy cannot be stopped: over time is only noted
+    cancellable = True  # libvirt aborts a live copy; a shut-off copy runs to its end
+    progress_bytes = True  # libvirt backup job disk_processed/disk_total
 
     def normalize(self, raw):
         fields = {k: str(v) for k, v in raw.items() if k in CLONE_FIELDS or CLONE_POST_KEY_RE.fullmatch(k)}
@@ -403,11 +431,14 @@ class CloneInstance(Kind):
 
     def plan(self, user, params):
         instance = _clonable(user, params["instance"])
-        if instance.proxy.get_status() != 5:
-            raise util.OperationError("Shut the VM down before cloning it")
         data = clone_data(user, instance, params["fields"])
         keys = vm_claims(instance.compute, data["name"])
-        keys[claims.instance_key(instance.uuid)] = Claim.SHARED
+        # the VM's definition is read again when the clone is defined
+        keys[claims.instance_key(instance.uuid)] = Claim.EXCLUSIVE
+        for disk in instance.config_disks:
+            if name := data.get(f"disk-{disk['dev']}"):
+                keys[claims.volume_key(instance.compute_id, disk["storage"], name)] = Claim.EXCLUSIVE
+        _check_clone_state(instance.proxy, data)
         return Plan(claims=keys, resource=claims.instance_key(instance.uuid),
                     label=f"{instance.compute.name}/{instance.name}",
                     intent={"compute": instance.compute_id, "uuid": util.randomUUID(), "name": data["name"],
@@ -417,18 +448,162 @@ class CloneInstance(Kind):
         instance = _clonable(op.user, op.params["instance"])
         if instance.compute_id != op.intent["compute"]:
             raise util.OperationError("The VM moved to another compute after the job was queued")
+        _check_clone_state(instance.proxy, op.intent["clone"])
 
     def run(self, op, ctx):
         instance = _instance(op.params["instance"])
-        ctx.record_intent()
+        proxy = instance.proxy
+        clone, uuid = op.intent["clone"], op.intent["uuid"]
+        status = proxy.get_status()
+        if status == 5:
+            ctx.record_intent(live=False)
+            ctx.set_phase(gettext_noop("Copying disks"))
+            proxy.clone_instance(clone, uuid)
+            record_vm(op, owner=True)
+            return {}
+        if status not in (1, 3):
+            raise util.OperationError(CLONE_STATE_REFUSED)
+        targets = proxy.live_clone_targets(clone, **_snapshot_free_space())
+        ctx.record_intent(live=True, targets=targets)
         ctx.set_phase(gettext_noop("Copying disks"))
-        instance.proxy.clone_instance(op.intent["clone"], op.intent["uuid"])
-        record_vm(op, owner=True)
-        return {}
+
+        copies = []  # the targets at the paths libvirt gave their volumes
+        for t in targets:
+            ctx.record_intent(planned=[*op.intent.get("planned", []), {"dev": t["dev"], "path": t["path"]}])
+            path = proxy.create_clone_target(t, clone["disk_owner_uid"], clone["disk_owner_gid"])
+            ctx.record_intent(created=[*op.intent.get("created", []), {"dev": t["dev"], "path": path}])
+            copies.append({**t, "path": path})
+
+        # a completed result libvirt still holds is of an earlier job; after
+        # copy_started, one can only be of this job's backup
+        proxy.discard_backup_result()
+        ctx.record_intent(copy_started=True)
+        op.refresh_from_db(fields=["cancel_requested"])
+        if op.cancel_requested:
+            _delete_created(proxy, op)
+            raise Cancelled
+
+        freeze = status == 1 and proxy.is_agent_ready()
+        warning = ""
+        try:
+            try:
+                if freeze:
+                    ctx.record_intent(frozen=True)
+                    proxy.fs_freeze()
+                proxy.begin_clone_backup(copies)
+            finally:
+                if freeze:
+                    try:
+                        proxy.fs_thaw()
+                    except libvirtError as err:  # frozen stays recorded: recovery thaws again
+                        warning = THAW_FAILED.format(error=err, vm=instance.name)
+                    else:
+                        ctx.record_intent(frozen=False)
+        except libvirtError as err:
+            # no backup of ours runs: the copy failed (if this read fails, recovery decides)
+            if proxy.backup_targets() != {c["path"] for c in copies}:
+                ctx.record_intent(copy_result=[False, str(err)])
+            raise
+        op.refresh_from_db(fields=["cancel_requested"])
+        if op.cancel_requested:  # the cancel hook may have run before the backup began
+            try:
+                self.cancel(op)
+            except libvirtError:
+                pass  # the hook aborted it, or it ended: its result tells
+
+        while proxy.backup_progress() is not None:
+            time.sleep(2)
+        result = proxy.backup_result()
+        ctx.record_intent(copy_result=list(result) if result else None)
+        if result is None:
+            raise util.OperationError("libvirt has no result of the copy")  # recovery decides
+        success, errmsg = result
+        if success:
+            proxy.define_clone(clone, uuid, {c["dev"]: c["path"] for c in copies})
+            record_vm(op, owner=True)
+            return {"warning": warning} if warning else {}
+        _delete_created(proxy, op)
+        op.refresh_from_db(fields=["cancel_requested"])
+        if op.cancel_requested:
+            raise Cancelled
+        raise util.OperationError(" ".join(filter(None, [errmsg or "The copy failed", warning])))
+
+    def progress(self, op):
+        return _proxy(_instance(op.params["instance"])).backup_progress() if op.intent.get("live") else None
+
+    def cancel(self, op):
+        # a shut-off copy, or a live one not started yet (run checks the request first)
+        if not op.intent.get("live"):
+            return
+        proxy = _proxy(_instance(op.params["instance"]))
+        created = {c["path"] for c in op.intent.get("created", [])}
+        if created and proxy.backup_targets() == created:
+            proxy.abort_backup()
 
     def reconcile(self, op):
-        disks = [v for k, v in op.intent["clone"].items() if k.startswith("disk-") and v]
-        return vm_outcome(op, owner=True, volumes=disks)
+        if not op.intent.get("live"):
+            disks = [v for k, v in op.intent["clone"].items() if k.startswith("disk-") and v]
+            return vm_outcome(op, owner=True, volumes=disks)
+        proxy = _recovery_proxy(op)
+        if proxy is None:
+            created = ", ".join(c["path"] for c in op.intent.get("created", []))
+            return Outcome(Operation.UNCERTAIN, error=f"{GONE}; check these volumes, then resolve this job: {created}")
+        warning = ""
+        if op.intent.get("frozen") and proxy.get_status() == 1:
+            try:
+                proxy.fs_thaw()
+            except libvirtError as err:
+                warning = " " + THAW_FAILED.format(error=err, vm=op.resource_label.rpartition("/")[2])
+        outcome = self._live_outcome(proxy, op)
+        outcome.error += warning
+        return outcome
+
+    def _live_outcome(self, proxy, op):
+        """The spec's recovery table: a defined clone first, then a volume
+        still being created, since it comes before the copy."""
+        if _domain(op.intent["compute"], op.intent["uuid"]) is not None:
+            record_vm(op, owner=True)
+            return Outcome(Operation.SUCCEEDED)
+        created = op.intent.get("created", [])
+        paths = [c["path"] for c in created]
+        ended = Operation.CANCELLED if op.cancel_requested else Operation.FAILED
+        done = {c["dev"] for c in created}
+        planned = {p["dev"] for p in op.intent.get("planned", [])}
+        for t in op.intent.get("targets", []):
+            if t["dev"] not in planned or t["dev"] in done:
+                continue
+            try:
+                proxy.wvm.storagePoolLookupByName(t["pool"]).storageVolLookupByName(t["name"])
+            except libvirtError as err:
+                if err.get_error_code() != VIR_ERR_NO_STORAGE_VOL:
+                    raise
+            else:
+                paths.append(t["path"])
+        check = f"check these volumes, then resolve this job: {', '.join(paths)}"
+        if len(paths) > len(created):
+            return Outcome(Operation.UNCERTAIN, error=f"A volume of the clone was being created; {check}")
+        if not op.intent.get("copy_started"):
+            _delete_created(proxy, op)
+            return Outcome(ended, error="The job stopped before the copy started")
+        # recorded only once this job's backup ended or never started
+        result = op.intent.get("copy_result")
+        if result is None:
+            targets = proxy.backup_targets()
+            if targets is not None:
+                if targets == set(paths):
+                    return Outcome(Operation.RUNNING, phase=gettext_noop("libvirt is still copying the disks"),
+                                   progress=proxy.backup_progress())
+                return Outcome(Operation.UNCERTAIN, error=f"Another backup runs on the VM; {check}")
+            result = proxy.backup_result()
+        if result is None:
+            return Outcome(Operation.UNCERTAIN, error=f"libvirt has no result of the copy; {check}")
+        success, errmsg = result
+        if not success:
+            _delete_created(proxy, op)
+            return Outcome(ended, error=errmsg or "The copy failed")
+        proxy.define_clone(op.intent["clone"], op.intent["uuid"], {c["dev"]: c["path"] for c in created})
+        record_vm(op, owner=True)
+        return Outcome(Operation.SUCCEEDED)
 
     def log_entry(self, op, state):
         compute, _sep, source = op.resource_label.rpartition("/")

instances/templates/instances/settings_tab.html

@@ -805,7 +805,7 @@
                         <textarea name="clone-description" class="form-control"></textarea>
                     </div>
                 </div>
-                {% if instance.status == 5 %}
+                {% if instance.status == 1 or instance.status == 3 or instance.status == 5 %}
                     <button type="submit" class="btn btn-lg btn-success float-end" name="clone">{% trans "Clone" %}</button>
                 {% else %}
                     <button class="btn btn-lg btn-success float-end disabled" disabled name="clone">{% trans "Clone" %}</button>

instances/tests/test_clone_job.py

@@ -2,6 +2,7 @@
 
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
+from uuid import uuid4
 
 import libvirt
 from django.contrib.auth import get_user_model
@@ -14,7 +15,7 @@ from appsettings.settings import get_settings
 from computes.models import Compute
 from instances.jobs import CloneInstance
 from instances.models import Instance
-from operations.jobs import Context
+from operations.jobs import Cancelled, Context
 from vrtManager.util import OperationError
 from operations.models import Claim, Operation, acquire
 
@@ -22,7 +23,7 @@ from operations.models import Claim, Operation, acquire
 def proxy():
     p = MagicMock()
     p.get_status.return_value = 5
-    p.get_disk_devices.return_value = [{"dev": "vda", "image": "src.qcow2", "size": 1 << 30}]
+    p.get_disk_devices.return_value = [{"dev": "vda", "image": "src.qcow2", "size": 1 << 30, "storage": "default"}]
     p.get_net_devices.return_value = [{"mac": "52:54:00:aa:bb:cc"}]
     p.get_vcpu.return_value = 1
     p.get_memory.return_value = 512
@@ -56,7 +57,8 @@ class CloneJobTestCase(TransactionTestCase):
         c = self.compute.pk
         self.assertEqual(set(Claim.objects.values_list("resource", "mode")), {
             (f"compute:{c}", "shared"), (f"reconcile:{c}", "exclusive"),
-            (f"instance:{self.source.uuid}", "shared"), (f"name:{c}:copy", "exclusive")})
+            (f"instance:{self.source.uuid}", "exclusive"), (f"name:{c}:copy", "exclusive"),
+            (f"volume:{c}:default/copy.qcow2", "exclusive")})
         self.proxy.clone_instance.assert_not_called()
 
         acquire("wvc-test:1:a", "wvc-test")
@@ -110,3 +112,427 @@ class CloneJobTestCase(TransactionTestCase):
         conn.return_value.wvm.lookupByUUIDString.side_effect = missing
         conn.return_value.wvm.listAllStoragePools.return_value = [pool(["other.qcow2"])]
         self.assertEqual(CloneInstance().reconcile(op).state, Operation.FAILED)
+
+
+def libvirt_error(code):
+    err = libvirt.libvirtError("libvirt error")
+    err.err = (code, libvirt.VIR_FROM_QEMU, "libvirt error")
+    return err
+
+
+def target(dev):
+    return {"dev": dev, "pool": "default", "name": f"copy-{dev}.qcow2",
+            "path": f"/pool/copy-{dev}.qcow2", "capacity": 1 << 30, "format": "qcow2"}
+
+
+class LiveCloneJobTestCase(TransactionTestCase):
+    """A running or paused VM is cloned with libvirt's push backup."""
+
+    def setUp(self):
+        get_settings()
+        self.admin = get_user_model().objects.create_superuser("live_admin", "a@example.com", "pw")
+        self.compute = Compute.objects.create(name="live-c", hostname="192.0.2.32", login="root", password="", type=1)
+        self.source = Instance.objects.create(compute=self.compute, name="src", uuid="23232323-3434-5656-7878-909090909090")
+        self.proxy = proxy()
+        for target_name in ("instances.models.wvmInstance", "instances.jobs.wvmInstance"):
+            patcher = patch(target_name, return_value=self.proxy)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        patcher = patch("instances.jobs.time.sleep")
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.events = []
+        p = self.proxy
+        p.get_status.return_value = 1
+        p.is_agent_ready.return_value = True
+        p.live_clone_targets.return_value = [target("vda"), target("vdb")]
+        p.create_clone_target.side_effect = self.created
+        for name in ("fs_freeze", "fs_thaw", "begin_clone_backup", "define_clone", "abort_backup",
+                     "discard_backup_result"):
+            self.track(name)
+        p.backup_progress.side_effect = [(1, 2), None]
+        p.backup_result.return_value = (True, "")
+        p.backup_targets.return_value = None
+
+    def track(self, name, error=None):
+        def call(*args):
+            self.events.append(name)
+            if error is not None:
+                raise error
+        getattr(self.proxy, name).side_effect = call
+
+    def created(self, t, uid, gid):
+        self.events.append(("create", t["dev"]))
+        return f"/real/{t['name']}"  # libvirt's path, not the planned one
+
+    def op(self, **intent):
+        return Operation.objects.create(
+            kind="instance.clone", user=self.admin, username=self.admin.username, submission_key=f"key-{uuid4().hex[:12]}",
+            resource=f"instance:{self.source.uuid}", resource_label="live-c/src", params={"instance": self.source.pk},
+            intent={"compute": self.compute.pk, "uuid": "cdcdcdcd-0000-1111-2222-333333333333", "name": "copy",
+                    "clone": {"name": "copy", "disk-vda": "copy-vda.qcow2", "disk-vdb": "copy-vdb.qcow2",
+                              "disk_owner_uid": 107, "disk_owner_gid": 107}, **intent})
+
+    def run_job(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()
+        test = self
+
+        class Recording(Context):
+            def record_intent(self, **values):
+                test.events.append(("intent", *sorted(values)))
+                super().record_intent(**values)
+
+        return CloneInstance().run(op, Recording(op, SimpleNamespace(identity="wvc-test:1:a")))
+
+    def deleted(self):
+        return [c.args[0] for c in self.proxy.get_volume_by_path.call_args_list]
+
+    # path choice and claims
+
+    def test_a_shut_off_vm_is_copied_as_before(self):
+        self.proxy.get_status.return_value = 5
+        op = self.op()
+        CloneInstance().precheck(op, None)
+        self.run_job(op)
+        self.proxy.clone_instance.assert_called_once_with(op.intent["clone"], op.intent["uuid"])
+        self.proxy.live_clone_targets.assert_not_called()
+        self.assertFalse(Operation.objects.get(pk=op.pk).intent["live"])
+
+    def test_a_running_or_paused_vm_takes_the_live_path(self):
+        for status in (1, 3):
+            with self.subTest(status=status):
+                self.proxy.reset_mock()
+                self.proxy.get_status.return_value = status
+                self.proxy.backup_progress.side_effect = [None]
+                op = self.op()
+                CloneInstance().precheck(op, None)
+                self.assertEqual(self.proxy.live_clone_targets.call_count, 1)
+                self.run_job(op)
+                self.proxy.clone_instance.assert_not_called()
+                self.proxy.begin_clone_backup.assert_called_once()
+                intent = Operation.objects.get(pk=op.pk).intent
+                self.assertTrue(intent["live"])
+                op.delete()
+
+    def test_another_state_is_refused(self):
+        self.proxy.get_status.return_value = 4  # shutting down
+        with self.assertRaisesMessage(OperationError, "The VM must be running, paused or shut off to be cloned"):
+            CloneInstance().precheck(self.op(), None)
+        with self.assertRaisesMessage(OperationError, "The VM must be running, paused or shut off to be cloned"):
+            CloneInstance().plan(self.admin, {"instance": self.source.pk, "fields": {
+                "name": "copy", "disk-vda": "copy.qcow2", "clone-net-mac-0": "52:54:00:00:00:01"}})
+
+    def test_plan_holds_the_source_and_each_target_volume(self):
+        self.proxy.get_disk_devices.return_value = [
+            {"dev": "vda", "image": "src.qcow2", "size": 1 << 30, "storage": "default"},
+            {"dev": "vdb", "image": "data.raw", "size": 1 << 30, "storage": "fast"}]
+        plan = CloneInstance().plan(self.admin, {"instance": self.source.pk, "fields": {
+            "name": "copy", "disk-vda": "copy.qcow2", "disk-vdb": "copy.raw", "clone-net-mac-0": "52:54:00:00:00:01"}})
+        c = self.compute.pk
+        self.assertEqual(plan.claims[f"instance:{self.source.uuid}"], Claim.EXCLUSIVE)
+        self.assertEqual(plan.claims[f"volume:{c}:default/copy.qcow2"], Claim.EXCLUSIVE)
+        self.assertEqual(plan.claims[f"volume:{c}:fast/copy.raw"], Claim.EXCLUSIVE)
+        self.proxy.live_clone_targets.assert_called_once()  # a running VM is refused early
+
+    # run
+
+    def test_the_live_run_records_each_step_before_it(self):
+        op = self.op()
+        self.assertEqual(self.run_job(op), {})
+        self.assertEqual(self.events, [
+            ("intent", "live", "targets"),
+            ("intent", "planned"), ("create", "vda"), ("intent", "created"),
+            ("intent", "planned"), ("create", "vdb"), ("intent", "created"),
+            "discard_backup_result", ("intent", "copy_started"),
+            ("intent", "frozen"), "fs_freeze", "begin_clone_backup", "fs_thaw", ("intent", "frozen"),
+            ("intent", "copy_result"), "define_clone"])
+        intent = Operation.objects.get(pk=op.pk).intent
+        self.assertEqual(intent["planned"], [{"dev": "vda", "path": "/pool/copy-vda.qcow2"},
+                                             {"dev": "vdb", "path": "/pool/copy-vdb.qcow2"}])
+        self.assertEqual(intent["created"], [{"dev": "vda", "path": "/real/copy-vda.qcow2"},
+                                             {"dev": "vdb", "path": "/real/copy-vdb.qcow2"}])
+        self.assertEqual((intent["frozen"], intent["copy_result"]), (False, [True, ""]))
+        backup = self.proxy.begin_clone_backup.call_args.args[0]
+        self.assertEqual([t["path"] for t in backup], ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+        self.proxy.define_clone.assert_called_once_with(
+            op.intent["clone"], op.intent["uuid"], {"vda": "/real/copy-vda.qcow2", "vdb": "/real/copy-vdb.qcow2"})
+        clone = Instance.objects.get(uuid=op.intent["uuid"])
+        self.assertTrue(UserInstance.objects.filter(instance=clone, user=self.admin).exists())
+        self.assertEqual(self.deleted(), [])
+
+    def test_a_paused_vm_or_one_without_a_ready_agent_is_not_frozen(self):
+        for status, agent in ((3, True), (1, False)):
+            with self.subTest(status=status, agent=agent):
+                self.events.clear()
+                self.proxy.get_status.return_value = status
+                self.proxy.is_agent_ready.return_value = agent
+                self.proxy.backup_progress.side_effect = [None]
+                op = self.op()
+                self.run_job(op)
+                self.assertNotIn("fs_freeze", self.events)
+                self.assertNotIn("fs_thaw", self.events)
+                self.assertIn("begin_clone_backup", self.events)
+                op.delete()
+
+    def recover(self, op, defined=False):
+        op.refresh_from_db()
+        with patch("instances.jobs.wvmConnect") as conn:
+            if not defined:
+                conn.return_value.wvm.lookupByUUIDString.side_effect = libvirt_error(libvirt.VIR_ERR_NO_DOMAIN)
+            return CloneInstance().reconcile(op)
+
+    def test_a_backup_that_did_not_start_failed(self):
+        self.track("begin_clone_backup", libvirt_error(libvirt.VIR_ERR_OPERATION_INVALID))
+        op = self.op()
+        with self.assertRaises(libvirt.libvirtError):
+            self.run_job(op)
+        self.assertEqual(self.events[-4:], ["begin_clone_backup", "fs_thaw", ("intent", "frozen"),
+                                            ("intent", "copy_result")])
+        intent = Operation.objects.get(pk=op.pk).intent
+        self.assertFalse(intent["frozen"])
+        self.assertEqual(intent["copy_result"], [False, "libvirt error"])
+        # an earlier job's completed result was dropped before copy_started
+        self.assertLess(self.events.index("discard_backup_result"), self.events.index(("intent", "copy_started")))
+        self.assertEqual(self.deleted(), [])  # recovery deletes them
+        outcome = self.recover(op)
+        self.assertEqual((outcome.state, outcome.error), (Operation.FAILED, "libvirt error"))
+        self.assertNotIn("define_clone", self.events)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+
+    def test_the_guest_is_thawed_when_the_freeze_fails(self):
+        self.track("fs_freeze", libvirt_error(libvirt.VIR_ERR_AGENT_UNRESPONSIVE))
+        op = self.op()
+        with self.assertRaises(libvirt.libvirtError):
+            self.run_job(op)
+        self.assertEqual(self.events[-5:], [("intent", "frozen"), "fs_freeze", "fs_thaw", ("intent", "frozen"),
+                                            ("intent", "copy_result")])
+        self.proxy.begin_clone_backup.assert_not_called()
+        self.assertEqual(self.recover(op).state, Operation.FAILED)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+
+    def test_a_failed_thaw_leaves_the_thaw_to_recovery(self):
+        self.track("fs_freeze", libvirt_error(libvirt.VIR_ERR_AGENT_UNRESPONSIVE))
+        self.track("fs_thaw", libvirt_error(libvirt.VIR_ERR_AGENT_UNRESPONSIVE))
+        op = self.op()
+        with self.assertRaises(libvirt.libvirtError):
+            self.run_job(op)
+        self.assertTrue(Operation.objects.get(pk=op.pk).intent["frozen"])
+        self.events.clear()
+        outcome = self.recover(op)
+        self.assertEqual(self.events[0], "fs_thaw")
+        self.assertEqual(outcome.state, Operation.FAILED)
+        self.assertIn("virsh domfsthaw src", outcome.error)
+
+    def test_a_lost_connection_while_the_backup_starts_is_recoverys(self):
+        self.track("begin_clone_backup", libvirt_error(libvirt.VIR_ERR_INTERNAL_ERROR))
+        self.proxy.backup_targets.side_effect = libvirt_error(libvirt.VIR_ERR_INTERNAL_ERROR)
+        op = self.op()
+        with self.assertRaises(libvirt.libvirtError):
+            self.run_job(op)
+        self.assertNotIn("copy_result", Operation.objects.get(pk=op.pk).intent)
+        self.proxy.backup_targets.side_effect = None
+        self.proxy.backup_result.return_value = None  # dropped before copy_started
+        self.assertEqual(self.recover(op).state, Operation.UNCERTAIN)
+        self.assertEqual(self.deleted(), [])
+
+    def test_a_cancel_while_the_backup_starts_aborts_it(self):
+        op = self.op()
+
+        def begin(*args):  # the cancel hook found no backup yet
+            self.events.append("begin_clone_backup")
+            Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+            self.proxy.backup_targets.return_value = {"/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"}
+        self.proxy.begin_clone_backup.side_effect = begin
+        self.track("abort_backup", libvirt_error(libvirt.VIR_ERR_OPERATION_INVALID))  # the hook was first
+        self.proxy.backup_result.return_value = (False, "Operation canceled")
+        with self.assertRaises(Cancelled):
+            self.run_job(op)
+        self.assertLess(self.events.index("begin_clone_backup"), self.events.index("abort_backup"))
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+
+    def test_a_thaw_error_is_a_warning_of_the_result(self):
+        self.track("fs_thaw", libvirt_error(libvirt.VIR_ERR_AGENT_UNRESPONSIVE))
+        result = self.run_job(self.op())
+        self.assertIn("could not be thawed", result["warning"])
+        self.assertIn("thaw them with: virsh domfsthaw src", result["warning"])
+        self.proxy.define_clone.assert_called_once()
+
+    def test_a_failed_copy_deletes_only_what_the_job_created(self):
+        self.proxy.backup_result.return_value = (False, "No space left on device")
+        op = self.op()
+        with self.assertRaisesMessage(OperationError, "No space left on device"):
+            self.run_job(op)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+        self.proxy.define_clone.assert_not_called()
+
+    def test_a_cancelled_copy_ends_cancelled(self):
+        self.proxy.backup_result.return_value = (False, "Operation canceled")
+        op = self.op()
+
+        def progress():  # the cancel comes while the copy runs
+            Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+            return None
+        self.proxy.backup_progress.side_effect = progress
+        with self.assertRaises(Cancelled):
+            self.run_job(op)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+
+    def test_a_cancel_before_the_copy_starts_ends_cancelled_without_it(self):
+        op = self.op()
+        Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+        with self.assertRaises(Cancelled):
+            self.run_job(op)
+        self.proxy.begin_clone_backup.assert_not_called()
+        self.proxy.fs_freeze.assert_not_called()
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2", "/real/copy-vdb.qcow2"])
+
+    def test_a_copy_that_succeeded_wins_over_a_late_cancel(self):
+        op = self.op()
+
+        def progress():
+            Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+            return None
+        self.proxy.backup_progress.side_effect = progress
+        self.run_job(op)
+        self.proxy.define_clone.assert_called_once()
+        self.assertEqual(self.deleted(), [])
+
+    def test_no_result_goes_to_recovery(self):
+        self.proxy.backup_result.return_value = None
+        op = self.op()
+        with self.assertRaises(OperationError):
+            self.run_job(op)
+        self.assertIsNone(Operation.objects.get(pk=op.pk).intent["copy_result"])
+        self.assertEqual(self.deleted(), [])
+
+    # cancel and progress
+
+    CREATED = [{"dev": "vda", "path": "/real/copy-vda.qcow2"}]
+
+    def test_cancel_aborts_only_this_jobs_backup(self):
+        op = self.op(live=True, created=self.CREATED)
+        for targets, aborted in ((None, False), ({"/other/x.qcow2"}, False), ({"/real/copy-vda.qcow2"}, True)):
+            with self.subTest(targets=targets):
+                self.events.clear()
+                self.proxy.backup_targets.return_value = targets
+                CloneInstance().cancel(op)
+                self.assertEqual("abort_backup" in self.events, aborted)
+
+    def test_a_shut_off_clone_ignores_a_cancel(self):
+        self.proxy.get_status.return_value = 5
+        op = self.op(live=False)
+        CloneInstance().cancel(op)
+        self.proxy.backup_targets.assert_not_called()
+        Operation.objects.filter(pk=op.pk).update(cancel_requested=True)
+        op.refresh_from_db()
+        self.run_job(op)
+        self.proxy.clone_instance.assert_called_once()
+
+    def test_progress_of_a_live_clone_is_the_backups_bytes(self):
+        self.proxy.backup_progress.side_effect = None
+        self.proxy.backup_progress.return_value = (3, 8)
+        self.assertEqual(CloneInstance().progress(self.op(live=True)), (3, 8))
+        self.assertIsNone(CloneInstance().progress(self.op(live=False)))
+        self.assertTrue(CloneInstance.progress_bytes)
+        self.assertTrue(CloneInstance.cancellable)
+        self.assertIsNone(CloneInstance.time_limit)  # the worker never cancels a long copy
+
+    # recovery
+
+    def recovered(self, cancel=False, defined=False, **intent):
+        base = {"live": True, "targets": [target("vda")],
+                "planned": [{"dev": "vda", "path": "/pool/copy-vda.qcow2"}], "created": self.CREATED,
+                "copy_started": True}
+        op = self.op(**{**base, **intent})
+        op.cancel_requested = cancel
+        op.save()
+        with patch("instances.jobs.wvmConnect") as conn:
+            if not defined:
+                conn.return_value.wvm.lookupByUUIDString.side_effect = libvirt_error(libvirt.VIR_ERR_NO_DOMAIN)
+            return CloneInstance().reconcile(op), op
+
+    def test_recovery_before_the_copy_deletes_what_was_created(self):
+        outcome, _op = self.recovered(copy_started=None)
+        self.assertEqual(outcome.state, Operation.FAILED)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2"])
+        self.proxy.reset_mock()
+        outcome, _op = self.recovered(copy_started=None, cancel=True)
+        self.assertEqual(outcome.state, Operation.CANCELLED)
+
+    def test_recovery_of_a_volume_being_created_is_uncertain(self):
+        outcome, _op = self.recovered(copy_started=None, created=[],
+                                      planned=[{"dev": "vda", "path": "/pool/copy-vda.qcow2"}])
+        self.assertEqual(outcome.state, Operation.UNCERTAIN)
+        self.assertIn("/pool/copy-vda.qcow2", outcome.error)
+        self.proxy.wvm.storagePoolLookupByName.assert_called_with("default")
+        self.proxy.wvm.storagePoolLookupByName.return_value.storageVolLookupByName.assert_called_with("copy-vda.qcow2")
+        self.proxy.get_volume_by_path.return_value.delete.assert_not_called()
+        self.proxy.wvm.storagePoolLookupByName.return_value.storageVolLookupByName.side_effect = libvirt_error(
+            libvirt.VIR_ERR_NO_STORAGE_VOL)
+        outcome, _op = self.recovered(copy_started=None, created=[])
+        self.assertEqual(outcome.state, Operation.FAILED)  # it was never made
+
+    def test_recovery_watches_this_jobs_backup(self):
+        self.proxy.backup_targets.return_value = {"/real/copy-vda.qcow2"}
+        self.proxy.backup_progress.side_effect = None
+        self.proxy.backup_progress.return_value = (1, 4)
+        outcome, _op = self.recovered()
+        self.assertEqual((outcome.state, outcome.progress), (Operation.RUNNING, (1, 4)))
+        self.proxy.backup_targets.return_value = {"/other/x.qcow2"}
+        outcome, _op = self.recovered()
+        self.assertEqual(outcome.state, Operation.UNCERTAIN)
+        self.assertEqual(self.deleted(), [])
+
+    def test_recovery_defines_the_clone_of_a_successful_copy(self):
+        for intent in ({"copy_result": [True, ""]}, {}):
+            with self.subTest(intent=intent):
+                self.events.clear()
+                outcome, op = self.recovered(**intent)
+                self.assertEqual(outcome.state, Operation.SUCCEEDED)
+                self.proxy.define_clone.assert_called_with(op.intent["clone"], op.intent["uuid"],
+                                                           {"vda": "/real/copy-vda.qcow2"})
+                self.assertTrue(Instance.objects.filter(uuid=op.intent["uuid"]).exists())
+                Instance.objects.filter(uuid=op.intent["uuid"]).delete()
+                op.delete()
+
+    def test_recovery_of_a_defined_clone_succeeds(self):
+        self.proxy.backup_result.return_value = None  # libvirtd restarted after the clone was defined
+        self.proxy.backup_targets.return_value = {"/other/x.qcow2"}  # checked after it
+        outcome, op = self.recovered(defined=True)
+        self.assertEqual(outcome.state, Operation.SUCCEEDED)
+        self.assertNotIn("define_clone", self.events)
+        self.assertTrue(Instance.objects.filter(uuid=op.intent["uuid"]).exists())
+
+    def test_recovery_of_a_failed_copy_deletes_what_was_created(self):
+        outcome, _op = self.recovered(copy_result=[False, "No space left on device"])
+        self.assertEqual(outcome.state, Operation.FAILED)
+        self.assertIn("No space left on device", outcome.error)
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2"])
+        outcome, _op = self.recovered(copy_result=[False, "Operation canceled"], cancel=True)
+        self.assertEqual(outcome.state, Operation.CANCELLED)
+
+    def test_a_recorded_failure_wins_over_another_tools_backup(self):
+        self.proxy.backup_targets.return_value = {"/other/x.qcow2"}
+        outcome, _op = self.recovered(copy_result=[False, "another backup runs"])
+        self.assertEqual((outcome.state, outcome.error), (Operation.FAILED, "another backup runs"))
+        self.assertEqual(self.deleted(), ["/real/copy-vda.qcow2"])
+        outcome, _op = self.recovered(copy_result=[False, "another backup runs"], cancel=True)
+        self.assertEqual(outcome.state, Operation.CANCELLED)
+
+    def test_recovery_without_a_result_is_uncertain(self):
+        self.proxy.backup_result.return_value = None
+        outcome, _op = self.recovered()
+        self.assertEqual(outcome.state, Operation.UNCERTAIN)
+        self.assertIn("/real/copy-vda.qcow2", outcome.error)
+        self.assertEqual(self.deleted(), [])
+
+    def test_recovery_thaws_a_guest_left_frozen_first(self):
+        self.proxy.backup_targets.side_effect = lambda: self.events.append("backup_targets")
+        self.recovered(frozen=True)
+        self.assertEqual(self.events[:2], ["fs_thaw", "backup_targets"])
+        self.events.clear()
+        self.proxy.get_status.return_value = 5
+        self.recovered(frozen=True)
+        self.assertNotIn("fs_thaw", self.events)

instances/tests/test_page_controls.py

@@ -178,6 +178,17 @@ class PageControlsTestCase(PageControls, TestCase):
                 )
                 self.assertEqual(bool(buttons), enabled)
 
+    def test_a_running_paused_or_shut_off_vm_can_be_cloned(self):
+        changer = self.user("vm_cloner", perms=("clone_instances",), owner={"is_change": True})
+        for status, enabled in ((1, True), (3, True), (5, True), (4, False)):
+            with self.subTest(status=status):
+                self.controls(changer, status=status)
+                buttons = lxml_html.fromstring(self.last_page).xpath(
+                    "//form[@action=$a]//button[@type='submit']",
+                    a=reverse("instances:clone", args=[self.instance.id]),
+                )
+                self.assertEqual(bool(buttons), enabled)
+
     def test_owner_with_change_and_no_extra_permissions(self):
         changer = self.user("ctl_plain_changer", owner={"is_change": True})
         self.assertEqual(self.controls(changer), CONSOLE | POWER | CHANGE)

operations/static/operations/job_panel.js

@@ -70,8 +70,9 @@ function jobPanel(element, urls, csrfToken, initialIds) {
 
     function outcomeHtml(job) {
         var left = (job.result && job.result.left_behind) || [];
-        var warn = left.length ? '<div class="small mt-2">Files not removed:</div><ul class="list-unstyled small mb-0">'
-            + left.map(function (path) { return '<li class="text-break"><code>' + esc(path) + "</code></li>"; }).join("") + "</ul>" : "";
+        var note = job.result && job.result.warning ? '<div class="small mt-2">' + esc(job.result.warning) + "</div>" : "";
+        var warn = note + (left.length ? '<div class="small mt-2">Files not removed:</div><ul class="list-unstyled small mb-0">'
+            + left.map(function (path) { return '<li class="text-break"><code>' + esc(path) + "</code></li>"; }).join("") + "</ul>" : "");
         var buttons = '<button type="button" class="btn btn-sm btn-primary" data-refresh="1">Refresh</button>'
             + '<button type="button" class="btn btn-sm btn-outline-secondary" data-dismiss="' + esc(job.id)
             + '">Dismiss</button>';

operations/templates/operations/job.html

@@ -56,6 +56,9 @@
                     </ul>
                 </dd>
             {% endif %}
+            {% if job.result.warning %}
+                <dt class="col-sm-3">{% trans "Warning" %}</dt><dd class="col-sm-9 text-break">{{ job.result.warning }}</dd>
+            {% endif %}
             {% if job.result.links %}
                 <dt class="col-sm-3">{% trans "Results" %}</dt>
                 <dd class="col-sm-9">

operations/tests/js/job_panel.test.js

@@ -92,8 +92,9 @@ assert.strictEqual(window.backgroundJobRunning, true, "still true while the outc
 answer("/j/a", { status: 0 }, false); // a network error: asked again on the next poll
 advance(3000);
 answer("/r", empty);
-answer("/j/a", { ...running, state: "succeeded", result: { left_behind: ["/pool/a.one"] } });
+answer("/j/a", { ...running, state: "succeeded", result: { left_behind: ["/pool/a.one"], warning: "not <thawed>" } });
 assert.ok(el.innerHTML.includes('<li class="text-break"><code>/pool/a.one</code></li>'), "shows the warning from the result");
+assert.ok(el.innerHTML.includes('<div class="small mt-2">not &lt;thawed&gt;</div>'), "shows the result's warning escaped");
 assert.ok(el.innerHTML.includes('class="alert alert-success') && el.innerHTML.includes('<span class="badge bg-success">succeeded</span>')
     && el.innerHTML.includes('<div class="ms-auto d-flex gap-2"><button type="button" class="btn btn-sm btn-primary" data-refresh'),
     "the outcome has the state colour and its buttons on the right");

operations/tests/test_views.py

@@ -73,6 +73,13 @@ class JobsPageTestCase(TransactionTestCase):
         self.assertNotContains(response, "<script>x</script>")
         self.assertContains(response, "Waiting for recovery")
 
+    def test_the_detail_shows_the_results_warning_escaped(self):
+        Operation.objects.filter(pk=self.op.pk).update(result={"warning": "not <b>thawed</b>"},
+                                                       state=Operation.SUCCEEDED)
+        self.client.force_login(self.user)
+        response = self.client.get(reverse("operations:job", args=[self.op.pk]))
+        self.assertContains(response, "not &lt;b&gt;thawed&lt;/b&gt;")
+
     def test_pagination_keeps_the_filters(self):
         for i in range(51):
             Operation.objects.create(kind="test.noop", user=self.user, username="page_user",

vrtManager/instance.py

@@ -2088,7 +2088,7 @@ class wvmInstance(wvmConnect):
                 raise util.OperationError(f"Clone failed ({err}); remove these copies by hand: {', '.join(leftovers)}") from err
             raise
 
-    def live_clone_targets(self, clone_data, free_percent, free_min_gib):
+    def live_clone_targets(self, clone_data, free_percent=10, free_min_gib=1):
         """Refuses a live clone that cannot work, before anything is
         allocated; returns the new volume of each copied disk, in disk order:
         in the source volume's pool, at its virtual capacity and format. Each
@@ -2193,6 +2193,11 @@ class wvmInstance(wvmConnect):
             return None
         return bool(stats["success"]), stats.get("errmsg") or ""
 
+    def discard_backup_result(self):
+        """Drops the result libvirt keeps of the VM's last completed job, so
+        a later backup_result() can only be of a backup started after this."""
+        self.instance.jobStats(VIR_DOMAIN_JOB_STATS_COMPLETED)
+
     def abort_backup(self):
         self.instance.abortJob()
 

vrtManager/tests/test_live_clone.py

@@ -192,6 +192,11 @@ class BackupTestCase(unittest.TestCase):
         }
         self.assertIsNone(p.backup_result())
 
+    def test_discard_reads_the_result_without_keeping_it(self):
+        p = proxy()
+        p.discard_backup_result()
+        p.instance.jobStats.assert_called_once_with(libvirt.VIR_DOMAIN_JOB_STATS_COMPLETED)
+
     def test_freeze_thaw_abort(self):
         p = proxy()
         p.fs_freeze()