fix(storages): keep volumes a VM still needs

libvirt deletes any volume it is asked to. The storage page and the API
deleted a VM's disk, the base image of another volume or the file a
snapshot was taken on without a check, breaking the VM or its
snapshots.

wvmStorage.del_volume now refuses a volume a domain on the compute uses
as a disk or a backing file of one, a file of a domain's snapshots, or
the backing file of another volume. The check is the one destroy
already used, moved to wvmConnect and extended to snapshot files, so
destroy also keeps the files of another VM's snapshots. The storage
page shows the refusal; the API answers 400 with it, as for other
refusals. Deleting a disk from a VM refuses after the detach when the
VM's own snapshots need the volume, and says the volume was kept.

Network disks count by their source name, pool volume disks by their
volume's path. A volume whose metadata cannot be read stops the check
instead of counting as unused; only one removed meanwhile is skipped.
The check and the delete hold the compute lock, so no VM operation
attaches the volume in between; a busy compute is a message on the page
and 409 in the API.

instances/tests/test_destroy.py

@@ -3,10 +3,12 @@ from unittest.mock import patch
 
 from computes.models import Compute
 from django.contrib.auth import get_user_model
+from django.contrib.messages import get_messages
 from django.test import TestCase
 from django.urls import reverse
 from instances.models import Instance
 from libvirt import VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, VIR_DOMAIN_UNDEFINE_NVRAM, libvirtError
+from vrtManager.util import OperationError
 
 
 class DestroyOrderTestCase(TestCase):
@@ -66,6 +68,20 @@ class DestroyOrderTestCase(TestCase):
             self.destroy()
         self.assertEqual(calls, [False, False])
 
+    def test_a_volume_the_vms_snapshots_need_is_kept_after_the_detach(self):
+        disk = {"dev": "vdb", "path": "/pool/data.qcow2", "storage": "default", "image": "data.qcow2"}
+        with patch("instances.models.wvmInstance") as wvm, patch("instances.views.wvmStorage") as storage:
+            proxy = wvm.return_value
+            proxy.get_disk_devices.return_value = [disk]
+            proxy.paths_used_by_other_domains.return_value = set()
+            proxy.wait_disk_detached.return_value = True
+            storage.return_value.del_volume.side_effect = OperationError("Volume data.qcow2 is in use")
+            response = self.client.post(reverse("instances:delete_vol", args=[self.instance.id]), {"dev": "vdb"})
+        self.assertEqual(response.status_code, 302)
+        proxy.detach_disk.assert_called_once_with("vdb")
+        messages = [str(m) for m in get_messages(response.wsgi_request)]
+        self.assertTrue(any("data.qcow2 is in use" in m and "detached" in m for m in messages), messages)
+
     def test_confirmation_page_takes_no_lock(self):
         calls = []
 

instances/tests/test_live_dataloss.py

@@ -321,6 +321,33 @@ class LiveDataLossTestCase(TestCase):
         pool.refresh(0)
         self.assertEqual({name for name in pool.listVolumes() if name.startswith(P + "r14-rev")}, set())
 
+    # Delete a volume from the storage page
+
+    def test_the_storage_page_keeps_volumes_a_vm_needs(self):
+        disk = livetest.create_volume(self.conn, P + "vol-disk")
+        dom, inst = self.vm("vol-disk", [disk])
+        inst.proxy.create_external_snapshot("a")  # disk is now the file snapshot a was taken on
+        template = livetest.create_volume(self.conn, P + "vol-template")
+        pool = self.conn.storagePoolLookupByName(livetest.POOL)
+        pool.createXML(
+            f"<volume><name>{P}vol-clone.qcow2</name><capacity unit='MiB'>64</capacity>"
+            f"<target><format type='qcow2'/></target><backingStore><path>{template}</path>"
+            "<format type='qcow2'/></backingStore></volume>",
+            0,
+        )
+        unused = livetest.create_volume(self.conn, P + "vol-unused")
+        top = self.disk_sources(dom)["vda"]
+        url = reverse("storage", args=[self.compute.id, livetest.POOL])
+
+        for path in (disk, template, top):
+            with self.subTest(path):
+                response = self.client.post(url, {"del_volume": "1", "volname": os.path.basename(path)})
+                self.assertEqual(response.status_code, 302)
+                self.assertTrue(livetest.volume_exists(self.conn, path), "a volume a VM needs was deleted")
+                self.assertTrue(any("is in use" in str(m) for m in get_messages(response.wsgi_request)))
+        self.client.post(url, {"del_volume": "1", "volname": os.path.basename(unused)})
+        self.assertFalse(livetest.volume_exists(self.conn, unused), "an unused volume was kept")
+
     # Delete a volume only after it is detached
 
     def _delete_attached_volume(self, pause):

instances/views.py

@@ -1223,7 +1223,13 @@ def delete_vol(request, pk):
             % {"dev": dev, "vol": disk["image"]},
         )
         return _back(request, pk, "disks")
-    conn_delete.del_volume(disk["image"])
+    try:
+        conn_delete.del_volume(disk["image"])
+    except util.OperationError as err:  # a file of the VM's own snapshots
+        messages.warning(
+            request,
+            _("Disk %(dev)s was detached, but its volume was kept: %(err)s") % {"dev": dev, "err": err},
+        )
 
     addlogmsg(request.user.username, instance.compute.name, instance.name, msg)
     return _back(request, pk, "disks")