feat(storages): volume clone runs as a job

instances/templates/instance.html

@@ -2,6 +2,7 @@
 {% load static %}
 {% load vm_states %}
 {% load i18n %}
+{% load operations %}
 {% load bootstrap_icons %}
 
 {% block title %}{% trans "Instance" %} - {{ instance.name }}{% endblock %}
@@ -682,14 +683,7 @@
         return true;
     });
 </script>
-<script src="{% static 'operations/job_panel.js' %}"></script>
-<script>
-    jobPanel(document.getElementById("job-panel"), {
-        resource: "{% url 'operation-resource' %}?key=instance%3A{{ instance.uuid|urlencode }}",
-        job: function (id) { return "{% url 'operation-list' %}" + id + "/"; },
-        cancel: function (id) { return "{% url 'operation-list' %}" + id + "/cancel/"; },
-    }, "{{ csrf_token }}", {{ remembered_jobs|safe }});
-</script>
+{% job_panel_script job_resource %}
 <script src="{% static 'js/instance_tabs.js' %}"></script>
 <script>
     // Open the tab whose control targets "#<pane id>"

instances/views.py

@@ -205,7 +205,7 @@ def _instance_page(request, pk, instance, compute):
         instance.drbd = drbd_status(request, pk)
         instance.save()
 
-    remembered_jobs = json.dumps(jobs.take_remembered(request, claims.instance_key(instance.uuid)))
+    job_resource = claims.instance_key(instance.uuid)
     return render(request, "instance.html", locals(),)
 
 

operations/claims.py

@@ -40,6 +40,14 @@ def instance_key(uuid):
     return f"instance:{uuid}"
 
 
+def pool_key(compute_id, pool):
+    return f"pool:{compute_id}:{pool}"
+
+
+def volume_key(compute_id, pool, name):
+    return f"volume:{compute_id}:{pool}/{name}"
+
+
 def quota_key(user_id):
     return f"quota:user:{user_id}"
 

operations/templates/operations/job_panel_script.html

@@ -0,0 +1,9 @@
+{% load static %}
+<script src="{% static 'operations/job_panel.js' %}"></script>
+<script>
+    jobPanel(document.getElementById("job-panel"), {
+        resource: "{% url 'operation-resource' %}?key={{ resource|urlencode:'' }}",
+        job: function (id) { return "{% url 'operation-list' %}" + id + "/"; },
+        cancel: function (id) { return "{% url 'operation-list' %}" + id + "/cancel/"; },
+    }, "{{ csrf_token }}", {{ remembered|safe }});
+</script>

operations/templatetags/operations.py

@@ -1,3 +1,4 @@
+import json
 import uuid
 
 from django import template
@@ -5,6 +6,7 @@ from django.utils.html import format_html, format_html_join
 from django.utils.safestring import mark_safe
 from django.utils.translation import pgettext_lazy
 
+from operations.jobs import take_remembered
 from operations.models import Operation
 
 register = template.Library()
@@ -53,3 +55,10 @@ def state_label(state):
 def wrap_at_slash(label):
     """A "<compute>/<vm>" label that wraps after its slash only, never inside a name."""
     return format_html_join(mark_safe("/<wbr>"), '<span class="text-nowrap">{}</span>', ((part,) for part in label.split("/")))
+
+
+@register.inclusion_tag("operations/job_panel_script.html", takes_context=True)
+def job_panel_script(context, resource):
+    """The job panel's script for the page of resource."""
+    return {"resource": resource, "csrf_token": context.get("csrf_token"),
+            "remembered": json.dumps(take_remembered(context["request"], resource))}

storages/jobs.py

@@ -0,0 +1,63 @@
+"""Jobs of the storages app (operations framework)."""
+
+from django.core.exceptions import PermissionDenied
+from django.utils.translation import gettext_noop
+from vrtManager import util
+from vrtManager.storage import wvmStorage
+
+from computes.models import Compute
+from operations import claims
+from operations.jobs import Kind, Outcome, Plan, register
+from operations.models import Claim, Operation
+
+
+def _storage(params):
+    c = Compute.objects.get(pk=params["compute"])
+    return c, wvmStorage(c.hostname, c.login, c.password, c.type, params["pool"])
+
+
+@register
+class CloneVolume(Kind):
+    name = "volume.clone"
+    label = gettext_noop("Clone volume")
+    time_limit = 12 * 3600  # a copy cannot be stopped: over time is only noted
+
+    def normalize(self, raw):
+        fmt = str(raw.get("format") or "") if raw.get("convert") else ""
+        return {"compute": int(raw["compute"]), "pool": str(raw["pool"]), "image": str(raw["image"]),
+                "name": str(raw["name"]), "format": fmt, "prealloc": fmt == "qcow2" and bool(raw.get("meta_prealloc"))}
+
+    def validate(self, user, params):
+        if not user.is_superuser:
+            raise PermissionDenied
+
+    def plan(self, user, params):
+        compute, stg = _storage(params)
+        volumes = stg.get_volumes()
+        if params["image"] not in volumes:
+            raise util.OperationError(f"The pool has no volume named {params['image']}")
+        target = stg.clone_name(params["name"], params["format"] or stg.get_volume_format_type(params["image"]))
+        if target in volumes:
+            raise util.OperationError("Name of volume already in use")
+        pool = claims.pool_key(compute.pk, params["pool"])
+        return Plan(claims={claims.compute_key(compute.pk): Claim.SHARED, pool: Claim.SHARED,
+                            claims.volume_key(compute.pk, params["pool"], target): Claim.EXCLUSIVE},
+                    resource=pool, label=f"{compute.name}/{params['pool']}", intent={"target": target})
+
+    def run(self, op, ctx):
+        _compute, stg = _storage(op.params)
+        ctx.record_intent()
+        ctx.set_phase(gettext_noop("Copying the volume"))
+        created = stg.clone_volume(op.params["image"], op.params["name"], op.params["format"] or None,
+                                   op.params["prealloc"])
+        ctx.record_intent(created=created)
+        return {}
+
+    def reconcile(self, op):
+        if op.intent.get("created"):
+            return Outcome(Operation.SUCCEEDED)
+        _compute, stg = _storage(op.params)  # unreachable: recovery waits and retries
+        if op.intent["target"] in stg.get_volumes():
+            return Outcome(Operation.UNCERTAIN, error="The copy exists, but the job did not see it finish; "
+                                                      "check it, then resolve this job.")
+        return Outcome(Operation.FAILED, error="The copy was not made")

storages/templates/storage.html

@@ -1,5 +1,6 @@
 {% extends "base.html" %}
 {% load i18n %}
+{% load operations %}
 {% load static %}
 {% load bootstrap_icons %}
 {% block title %}{{ compute }} - {% trans "Storage" %}: {{ pool }}{% endblock %}
@@ -13,6 +14,7 @@
 
 {% block content %}
     {% include 'pleasewaitdialog.html' %}
+    <div id="job-panel" aria-live="polite"></div>
     <div class="row">
         <div class="col-lg-12">
             
@@ -119,7 +121,7 @@
                                         <h5 class="modal-title">{% trans "Clone image" %} <span class="text-danger">{{ volume.name }}</span></h5>
                                         <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
                                     </div>
-                                    <form method="post" role="form" aria-label="Clone image file form">{% csrf_token %}
+                                    <form method="post" role="form" aria-label="Clone image file form">{% csrf_token %}{% submission_key %}
                                         <div class="modal-body">
                                             <div class="row">
                                                 <label class="col-sm-3 col-form-label">{% trans "Name" %}</label>
@@ -153,7 +155,7 @@
                                         </div>
                                         <div class="modal-footer">
                                             <button type="button" class="btn btn-secondary" data-bs-dismiss="modal">{% trans "Close" %}</button>
-                                            <button type="submit" class="btn btn-primary" name="cln_volume" onclick="showPleaseWaitDialog()">{% trans "Clone" %}</button>
+                                            <button type="submit" class="btn btn-primary" name="cln_volume">{% trans "Clone" %}</button>
                                         </div>
                                     </form>
                                 </div> <!-- /.modal-content -->
@@ -193,6 +195,8 @@
 
 {% endblock %}
 {% block script %}
+    <script src="{% static 'js/poll.js' %}"></script>
+    {% job_panel_script job_resource %}
     <script src="{% static 'js/sortable.min.js' %}"></script>
     <script src="{% static 'js/filter-table.js' %}"></script>
     <script>

storages/tests/test_volume_clone_job.py

@@ -0,0 +1,59 @@
+"""Volume clone runs as a job: the page queues it, the worker copies."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+from django.urls import reverse
+
+from computes.models import Compute
+from operations.jobs import Context
+from operations.models import Claim, Operation, acquire
+from storages.jobs import CloneVolume
+
+
+class VolumeCloneJobTestCase(TransactionTestCase):
+    def setUp(self):
+        self.compute = Compute.objects.create(name="vol-c", hostname="192.0.2.20", login="root", password="", type=1)
+        admin = get_user_model().objects.create_superuser("vol_admin", "a@example.com", "pw")
+        self.client.force_login(admin)
+        patcher = patch("storages.jobs.wvmStorage")
+        self.stg = patcher.start().return_value
+        self.addCleanup(patcher.stop)
+        self.stg.get_volumes.return_value = ["base.qcow2"]
+        self.stg.get_volume_format_type.return_value = "qcow2"
+        self.stg.clone_name.return_value = "copy.qcow2"
+        self.stg.clone_volume.return_value = "copy.qcow2"
+
+    @patch("storages.views.wvmStorage")
+    def test_the_page_queues_the_copy_and_the_worker_makes_it(self, page):
+        page.return_value.is_active.return_value = False
+        page.return_value.get_size.return_value = (1, 1)
+        url = reverse("storage", args=[self.compute.pk, "pool-a"])
+        response = self.client.post(url, {"cln_volume": "1", "image": "base.qcow2", "name": "copy",
+                                          "submission_key": "key-vol00001"})
+        self.assertRedirects(response, url, fetch_redirect_response=False)
+        op = Operation.objects.get()
+        self.assertEqual((op.kind, op.intent["target"]), ("volume.clone", "copy.qcow2"))
+        self.assertIn((f"volume:{self.compute.pk}:pool-a/copy.qcow2", "exclusive"),
+                      set(Claim.objects.values_list("resource", "mode")))
+        self.stg.clone_volume.assert_not_called()
+
+        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()
+        CloneVolume().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a")))
+        op.refresh_from_db()
+        self.stg.clone_volume.assert_called_once_with("base.qcow2", "copy", None, False)
+        self.assertEqual(op.intent["created"], "copy.qcow2")
+        self.assertIn(f"key=pool%3A{self.compute.pk}%3Apool-a", self.client.get(url).content.decode())
+
+    def test_recovery(self):
+        self.stg.get_volumes.return_value = ["base.qcow2", "copy.qcow2"]
+        op = Operation(params={"compute": self.compute.pk, "pool": "pool-a"}, intent={"target": "copy.qcow2"})
+        self.assertEqual(CloneVolume().reconcile(op).state, Operation.UNCERTAIN)  # there, not proven complete
+        self.stg.get_volumes.return_value = []
+        self.assertEqual(CloneVolume().reconcile(op).state, Operation.FAILED)
+        op.intent["created"] = "copy.qcow2"
+        self.assertEqual(CloneVolume().reconcile(op).state, Operation.SUCCEEDED)

storages/views.py

@@ -11,7 +11,7 @@ from django.shortcuts import get_object_or_404, redirect, render
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 from libvirt import libvirtError
-from operations import claims
+from operations import claims, jobs
 from operations.utils import busy_message
 
 from vrtManager.connection import CONN_TCP, CONN_TLS
@@ -253,34 +253,22 @@ def storage(request, compute_id, pool):
         if "cln_volume" in request.POST:
             form = CloneImage(request.POST)
             if form.is_valid():
-                data = form.cleaned_data
-                img_name = data["name"]
-                meta_prealloc = 0
-                if img_name in conn.update_volumes():
-                    msg = _("Name of volume already in use")
-                    messages.error(request, msg)
-                if data["convert"]:
-                    format = data["format"]
-                    if data["meta_prealloc"] and data["format"] == "qcow2":
-                        meta_prealloc = True
-                else:
-                    format = None
                 try:
-                    name = conn.clone_volume(
-                        data["image"], data["name"], format, meta_prealloc
-                    )
-                    messages.success(
-                        request,
-                        _("%(image)s image cloned as %(name)s successfully")
-                        % {"image": data["image"], "name": name},
-                    )
-                    return HttpResponseRedirect(request.get_full_path())
-                except libvirtError as lib_err:
-                    messages.error(request, lib_err)
+                    op = jobs.submit("volume.clone", request.user, {"compute": compute.id, "pool": pool,
+                                     **form.cleaned_data}, request.POST.get("submission_key", ""))
+                except claims.Busy as busy:
+                    messages.error(request, busy_message(busy, request.user))
+                except (jobs.SubmissionConflict, libvirtError) as err:
+                    messages.error(request, str(err))
+                else:
+                    messages.info(request, _("The volume is being cloned; follow it in the job panel."))
+                    jobs.remember_job(request, op)
             else:
                 for msg_err in form.errors.values():
                     messages.error(request, msg_err.as_text())
+            return redirect(reverse("storage", args=[compute.id, pool]))
 
+    job_resource = claims.pool_key(compute.id, pool)
     conn.close()
 
     return render(request, "storage.html", locals())

vrtManager/storage.py

@@ -335,6 +335,14 @@ class wvmStorage(wvmConnect):
         self._createXML(xml, metadata)
         return name
 
+    def clone_name(self, target_file, vol_fmt, file_suffix="img"):
+        """The name clone_volume gives a copy: a dir pool adds the format's suffix."""
+        if self.get_type() != "dir":
+            return target_file
+        if vol_fmt in ("qcow", "qcow2"):
+            return f"{target_file}.{vol_fmt}"
+        return f"{target_file}.{file_suffix}" if file_suffix else target_file
+
     def clone_volume(
         self,
         name,
@@ -350,13 +358,7 @@ class wvmStorage(wvmConnect):
         if not vol_fmt:
             vol_fmt = self.get_volume_format_type(name)
 
-        storage_type = self.get_type()
-        if storage_type == "dir":
-            if vol_fmt in ["qcow", "qcow2"]:
-                target_file += f".{vol_fmt}"
-            else:
-                suffix = f".{file_suffix}"
-                target_file += suffix if len(suffix) > 1 else ""
+        target_file = self.clone_name(target_file, vol_fmt, file_suffix)
 
         xml = f"""
             <volume>