feat(instances): VM create runs as a job, also in the API

computes/templates/computes/instances.html

@@ -1,6 +1,7 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% load static %}
+{% load operations %}
 {% load bootstrap_icons %}
 {% block title %}{% trans "Instances" %} - {{ compute.name }}{% endblock %}
 {% block style %}
@@ -19,6 +20,7 @@
 {% endblock page_heading_extra %}
 
 {% block content %}
+<div id="job-panel" aria-live="polite"></div>
 <div class="row">
     <div class="col-lg-12">
         <nav aria-label="breadcrumb">
@@ -110,6 +112,8 @@
 </div>
 {% 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>
 {% endblock %}

computes/views.py

@@ -8,6 +8,7 @@ from django.urls import reverse
 from django.utils import timezone
 from instances.models import Instance
 from libvirt import libvirtError
+from operations import claims
 from vrtManager.connection import wvmConnect
 from vrtManager.hostdetails import wvmHostDetails
 
@@ -80,7 +81,8 @@ def instances(request, compute_id):
     )
 
     return render(
-        request, "computes/instances.html", {"compute": compute, "instances": instances}
+        request, "computes/instances.html",
+        {"compute": compute, "instances": instances, "job_resource": claims.compute_key(compute.id)},
     )
 
 

instances/api/serializers.py

@@ -58,6 +58,7 @@ class CreateInstanceSerializer(serializers.ModelSerializer):
     cache_mode = serializers.CharField(initial="none")
     virtio = serializers.BooleanField(initial=True)
     qemu_ga = serializers.BooleanField(initial=True)
+    submission_key = serializers.CharField(write_only=True)
 
     def validate(self, attrs):
         try:
@@ -82,9 +83,9 @@ class CreateInstanceSerializer(serializers.ModelSerializer):
             "meta_prealloc",
             "virtio",
             "qemu_ga",
-            "console_pass",
             "video",
             "listener_addr",
+            "submission_key",
         ]
 
 

instances/api/viewsets.py

@@ -1,9 +1,9 @@
 from admin.permissions import IsSuperUser
-from appsettings.settings import app_settings
 from computes import utils
 from computes.models import Compute
 from django.http import Http404
 from django.shortcuts import get_object_or_404
+from django.urls import reverse
 from instances.models import Flavor, Instance
 from instances.utils import migrate_instance
 from instances.views import destroy as instance_destroy
@@ -17,12 +17,11 @@ from instances.views import (
     suspend,
 )
 from drf_spectacular.utils import extend_schema
-from libvirt import libvirtError
+from operations import jobs
+from operations.api.serializers import OperationSerializer
 from rest_framework import permissions, status, viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
-from vrtManager import util
-from vrtManager.create import wvmCreate
 from webvirtcloud.serializers import StatusSerializer
 
 from .serializers import (
@@ -227,94 +226,15 @@ class CreateInstanceViewSet(viewsets.ViewSet):
     serializer_class = CreateInstanceSerializer
     queryset = ""
 
-    @extend_schema(responses=StatusSerializer)
+    @extend_schema(responses={202: OperationSerializer, 400: dict})
     def create(self, request, compute_pk=None, arch=None, machine=None):
-        serializer = CreateInstanceSerializer(
+        serializer = CreateInstanceSerializer(data=request.data,
-            data=request.data,
+                                              context={"compute_pk": compute_pk, "arch": arch, "machine": machine})
-            context={"compute_pk": compute_pk, "arch": arch, "machine": machine},
+        if not serializer.is_valid():
-        )
-        if serializer.is_valid():
-            volume_list = []
-            default_bus = app_settings.INSTANCE_VOLUME_DEFAULT_BUS
-            default_io = app_settings.INSTANCE_VOLUME_DEFAULT_IO
-            default_discard = app_settings.INSTANCE_VOLUME_DEFAULT_DISCARD
-            default_zeroes = app_settings.INSTANCE_VOLUME_DEFAULT_DETECT_ZEROES
-            default_scsi_disk_model = (
-                app_settings.INSTANCE_VOLUME_DEFAULT_SCSI_CONTROLLER
-            )
-            default_disk_format = app_settings.INSTANCE_VOLUME_DEFAULT_FORMAT
-            default_disk_owner_uid = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID)
-            default_disk_owner_gid = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID)
-            compute = Compute.objects.get(pk=compute_pk)
-            conn = wvmCreate(
-                compute.hostname,
-                compute.login,
-                compute.password,
-                compute.type,
-            )
-
-            path = None  # the volume this request allocated
-            try:
-                path = conn.create_volume(
-                    serializer.validated_data["storage"],
-                    serializer.validated_data["name"],
-                    serializer.validated_data["hdd_size"],
-                    default_disk_format,
-                    serializer.validated_data["meta_prealloc"],
-                    default_disk_owner_uid,
-                    default_disk_owner_gid,
-                )
-                volume = {}
-                volume["device"] = "disk"
-                volume["path"] = path
-                volume["type"] = conn.get_volume_format_type(path)
-                volume["cache_mode"] = serializer.validated_data["cache_mode"]
-                volume["bus"] = default_bus
-                if volume["bus"] == "scsi":
-                    volume["scsi_model"] = default_scsi_disk_model
-                volume["discard_mode"] = default_discard
-                volume["detect_zeroes_mode"] = default_zeroes
-                volume["io_mode"] = default_io
-
-                volume_list.append(volume)
-
-                with utils.libvirt_compute_lock(compute):
-                    conn.create_instance(
-                        name=serializer.validated_data["name"],
-                        memory=serializer.validated_data["memory"],
-                        vcpu=serializer.validated_data["vcpu"],
-                        vcpu_mode=serializer.validated_data["vcpu_mode"],
-                        uuid=util.randomUUID(),
-                        arch=arch,
-                        machine=machine,
-                        firmware={},
-                        volumes=volume_list,
-                        networks=serializer.validated_data["networks"],
-                        nwfilter=serializer.validated_data["nwfilter"],
-                        virtio=serializer.validated_data["virtio"],
-                        listener_addr=serializer.validated_data["listener_addr"],
-                        video=serializer.validated_data["video"],
-                        console_pass=serializer.validated_data["console_pass"],
-                        mac=serializer.validated_data["mac"],
-                        qemu_ga=serializer.validated_data["qemu_ga"],
-                    )
-            except Exception as err:
-                # A failed definition must not leave the new disk behind
-                if path:
-                    try:
-                        conn.delete_volume(path)
-                    except libvirtError:
-                        return Response(
-                            {"status": f"{err}; remove {path} by hand"}, status=status.HTTP_400_BAD_REQUEST
-                        )
-                if not isinstance(err, libvirtError):
-                    raise
-                return Response({"status": str(err)}, status=status.HTTP_400_BAD_REQUEST)
-            # After the definition: the VM uses the volume, so no rollback here.
-            # The refresh claims the compute itself and skips it when busy; the
-            # VM exists either way, so a busy compute must not refuse the request.
-            utils.refresh_instance_database(compute)
-            msg = f"Instance {serializer.validated_data['name']} is created"
-            return Response({"status": msg})
-        else:
             return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+        data = dict(serializer.validated_data)
+        key = data.pop("submission_key")
+        op = jobs.submit("instance.create", request.user, {**data, "compute": compute_pk, "arch": arch,
+                                                           "machine": machine}, key)
+        return Response(OperationSerializer(op, context={"request": request}).data, status=status.HTTP_202_ACCEPTED,
+                        headers={"Location": reverse("operation-detail", args=[op.pk])})

instances/forms.py

@@ -58,7 +58,6 @@ class NewVMForm(forms.ModelForm):
     # virtio = forms.BooleanField(required=False)
     # qemu_ga = forms.BooleanField(required=False)
     # mac = forms.CharField(required=False)
-    # console_pass = forms.CharField(required=False, empty_value="", widget=forms.PasswordInput())
     # graphics = forms.CharField(error_messages={'required': _('Please select a graphics type')})
     # video = forms.CharField(error_messages={'required': _('Please select a video driver')})
     # listener_addr = forms.ChoiceField(required=True, widget=forms.RadioSelect, choices=QEMU_CONSOLE_LISTENER_ADDRESSES)

instances/jobs.py

@@ -9,6 +9,7 @@ from libvirt import (VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE, VIR_DOMAIN_SNAPSHOT_
                      libvirtError)
 from vrtManager import util
 from vrtManager.connection import wvmConnect
+from vrtManager.create import wvmCreate
 from vrtManager.instance import wvmInstance
 
 from accounts.models import UserInstance
@@ -281,3 +282,150 @@ class CloneInstance(Kind):
         compute, _sep, source = op.resource_label.rpartition("/")
         message = f"Create a clone of '{source}'"
         return compute, op.intent["name"], message if state == Operation.SUCCEEDED else f"{message} ({state})"
+
+
+@register
+class CreateInstance(Kind):
+    name = "instance.create"
+    label = gettext_noop("Create VM")
+    time_limit = 12 * 3600  # a copy cannot be stopped: over time is only noted
+
+    def normalize(self, raw):
+        # what the API leaves out means what create_instance did by default
+        params = {"template": "", "images": "", "firmware": "", "net_model": "virtio", "add_cdrom": "sata",
+                  "add_input": "default", "meta_prealloc": False, "hdd_size": 0}
+        params.update({k: v for k, v in raw.items() if k not in ("submission_key", "create", "csrfmiddlewaretoken")})
+        params.update(compute=int(raw["compute"]), devices=[list(d) for d in raw.get("devices", [])])
+        if "UEFI" in params["firmware"] and "secboot" in params["firmware"].partition(":")[2]:
+            params["machine"] = "q35"  # a secure boot loader needs SMM, on q35 only
+        return params
+
+    def validate(self, user, params):
+        from instances.views import CDROM_BUSES
+
+        if not user.is_superuser:
+            raise PermissionDenied
+        data = params
+        c = Compute.objects.get(pk=data["compute"])
+        conn = wvmCreate(c.hostname, c.login, c.password, c.type)
+        cdrom_buses = [bus for bus in conn.get_disk_bus_types(data["arch"], data["machine"]) if bus in CDROM_BUSES]
+        if data["name"] in conn.get_instances():
+            raise util.OperationError(_("A virtual machine with this name already exists"))
+        if Instance.objects.filter(name__exact=data["name"]):
+            raise util.OperationError(_("There is an instance with same name. Remove it and try again!"))
+        if not (data["hdd_size"] or data["template"] or data["images"]):
+            raise util.OperationError(_("First you need to create or select an image"))
+        if not data["hdd_size"] and data["template"] and conn.get_volume_path(data["name"] + ".img", data["storage"]):
+            raise util.OperationError(_("Image has already exist. Please check volumes or change instance name"))
+        if data["cache_mode"] not in conn.get_cache_modes():
+            raise util.OperationError(_("Invalid cache mode"))
+        if data["add_cdrom"] != "None" and data["add_cdrom"] not in cdrom_buses:
+            raise util.OperationError(_("A CD-ROM cannot use the %(bus)s bus") % {"bus": data["add_cdrom"]})
+        try:
+            utils.nic_macs(data["mac"], data["networks"])
+        except ValueError as err:
+            raise util.OperationError(str(err))
+        if "UEFI" in data["firmware"] and ":" not in data["firmware"]:
+            raise util.OperationError(_("Invalid firmware"))
+
+    def plan(self, user, params):
+        compute = Compute.objects.get(pk=params["compute"])
+        return Plan(claims=vm_claims(compute, params["name"]), resource=claims.compute_key(compute.pk),
+                    label=f"{compute.name}/{params['name']}",
+                    intent={"compute": compute.pk, "uuid": util.randomUUID(), "name": params["name"]})
+
+    def run(self, op, ctx):
+        data = op.params
+        c = Compute.objects.get(pk=data["compute"])
+        conn = wvmCreate(c.hostname, c.login, c.password, c.type)
+        meta_prealloc = bool(data["meta_prealloc"])
+        default_bus = app_settings.INSTANCE_VOLUME_DEFAULT_BUS
+        default_disk_format = app_settings.INSTANCE_VOLUME_DEFAULT_FORMAT
+        default_disk_owner_uid = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID)
+        default_disk_owner_gid = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID)
+
+        firmware = dict()
+        if "UEFI" in data["firmware"]:
+            firmware["loader"] = data["firmware"].split(":")[1].strip()
+            firmware["secure"] = "yes" if "secboot" in firmware["loader"] else "no"
+            firmware["readonly"] = "yes"
+            firmware["type"] = "pflash"
+
+        def volume(path, device="disk", bus=default_bus):
+            return {
+                "path": path,
+                "type": conn.get_volume_format_type(path),
+                "device": device,
+                "bus": bus,
+                "scsi_model": app_settings.INSTANCE_VOLUME_DEFAULT_SCSI_CONTROLLER if bus == "scsi" else None,
+                "cache_mode": data["cache_mode"],
+                "discard_mode": app_settings.INSTANCE_VOLUME_DEFAULT_DISCARD,
+                "detect_zeroes_mode": app_settings.INSTANCE_VOLUME_DEFAULT_DETECT_ZEROES,
+                "io_mode": app_settings.INSTANCE_VOLUME_DEFAULT_IO,
+            }
+
+        ctx.record_intent()
+        ctx.set_phase(gettext_noop("Copying disks"))
+        created = []  # volumes this job allocated
+        volume_list = []
+        try:
+            if data["hdd_size"]:
+                path = conn.create_volume(data["storage"], data["name"], data["hdd_size"], default_disk_format,
+                                          meta_prealloc, default_disk_owner_uid, default_disk_owner_gid)
+                created.append(path)
+                volume_list.append(volume(path))
+            elif data["template"]:
+                path = conn.clone_from_template(data["name"], conn.get_volume_path(data["template"]), data["storage"],
+                                                meta_prealloc, default_disk_owner_uid, default_disk_owner_gid)
+                created.append(path)
+                volume_list.append(volume(path))
+            else:
+                for vol, (device, bus) in zip(data["images"].split(","), data["devices"]):
+                    volume_list.append(volume(conn.get_volume_path(vol), device, bus))
+
+            ctx.set_phase(gettext_noop("Defining the VM"))
+            conn.create_instance(
+                name=data["name"],
+                memory=data["memory"],
+                vcpu=data["vcpu"],
+                vcpu_mode=data["vcpu_mode"],
+                uuid=op.intent["uuid"],
+                arch=data["arch"],
+                machine=data["machine"],
+                firmware=firmware,
+                volumes=volume_list,
+                networks=data["networks"],
+                virtio=data["virtio"],
+                listener_addr=data["listener_addr"],
+                nwfilter=data["nwfilter"],
+                net_model="virtio" if data["net_model"] == "default" else data["net_model"],
+                video=data["video"],
+                console_pass="",
+                mac=data["mac"],
+                qemu_ga=data["qemu_ga"],
+                add_cdrom=data["add_cdrom"],
+                add_input=data["add_input"],
+            )
+        except Exception as err:
+            # the VM is not defined: its new volumes go
+            leftovers = []
+            for path in created:
+                try:
+                    conn.delete_volume(path)
+                except libvirtError:
+                    leftovers.append(path)
+            if leftovers:
+                raise util.OperationError(f"{err}. Remove these volumes by hand: {', '.join(leftovers)}") from err
+            raise
+        record_vm(op, owner=False)
+        return {}
+
+    def reconcile(self, op):
+        data = op.params
+        names = [data["name"] + ext for ext in ("", ".img", ".qcow", ".qcow2")]
+        return vm_outcome(op, owner=False, volumes=names if data["hdd_size"] or data["template"] else [])
+
+    def log_entry(self, op, state):
+        compute, _sep, name = op.resource_label.rpartition("/")
+        message = "Instance is created"
+        return compute, name, message if state == Operation.SUCCEEDED else f"{message} ({state})"

instances/models.py

@@ -373,7 +373,6 @@ class CreateInstance(models.Model):
     virtio = models.BooleanField(default=True)
     qemu_ga = models.BooleanField(default=False)
     mac = models.CharField(max_length=17, blank=True)
-    console_pass = models.CharField(max_length=64, blank=True)
     add_cdrom = models.CharField(max_length=16)
     add_input = models.CharField(max_length=16)
     video = models.CharField(

instances/templates/create_instance_w2.html

@@ -1,6 +1,7 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% load static %}
+{% load operations %}
 {% load bootstrap_icons %}
 {% block title %}{% trans "Create new instance" %}{% endblock %}
 
@@ -97,7 +98,7 @@
                                                             <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
                                                         </div>
                                                         <div class="modal-body">
-                                                            <form method="post" role="form" aria-label="Create instance from flavor form">{% csrf_token %}
+                                                            <form method="post" role="form" aria-label="Create instance from flavor form">{% csrf_token %}{% submission_key %}
                                                                 <div class="row">
                                                                     <label class="col-sm-4 col-form-label">{% trans "Name" %}</label>
                                                                     <div class="col-sm-7">
@@ -240,12 +241,6 @@
                                                                             </select>
                                                                         </div>
                                                                     </div>
-                                                                    <div class="row">
-                                                                        <label class="col-sm-4 col-form-label">{% trans "Console Password" %}</label>
-                                                                        <div class="col-sm-7">
-                                                                            <input class="form-control" type="password" name="console_pass" placeholder="{% trans "Console Password" %}" maxlength="64">
-                                                                        </div>
-                                                                    </div>
                                                                     <div class="row">
                                                                         <label class="col-sm-4 col-form-label">{% trans "Add CD-Rom" %}</label>
                                                                         <div class="col-sm-7">
@@ -323,7 +318,7 @@
 
             <div role="tabpanel" class="tab-pane tab-pane-bordered" id="addCustom">
                 <div>
-                    <form method="post" role="form" aria-label="Create custom instance form">{% csrf_token %}
+                    <form method="post" role="form" aria-label="Create custom instance form">{% csrf_token %}{% submission_key %}
                     <div class="row">
                         <label class="col-sm-3 col-form-label">{% trans "Name" %}</label>
                         <div class="col-sm-7">
@@ -493,12 +488,6 @@
                                 </select>
                             </div>
                         </div>
-                        <div class="row">
-                            <label class="col-sm-3 col-form-label">{% trans "Console Password" %}</label>
-                            <div class="col-sm-7">
-                                <input type="password" class="form-control" name="console_pass" placeholder="{% trans "Console Password" %}" maxlength="64">
-                            </div>
-                        </div>
                         <div class="row">
                             <label class="col-sm-3 col-form-label">{% trans "Add CD-Rom" %}</label>
                             <div class="col-sm-7">
@@ -543,7 +532,7 @@
                     <div class="row">
                         <div class="d-grid col-sm-7 offset-3">
                         {% if storages %}
-                            <button type="submit" class="btn btn-primary" name="create" formnovalidate onclick="showPleaseWaitDialog()" value="1">
+                            <button type="submit" class="btn btn-primary" name="create" formnovalidate value="1">
                                 {% trans "Create" %}
                             </button>
                         {% else %}
@@ -560,7 +549,7 @@
 
             <div role="tabpanel" class="tab-pane tab-pane-bordered" id="add-from-temp">
                 <div>
-                    <form method="post" role="form" aria-label="Create instance from template form">{% csrf_token %}
+                    <form method="post" role="form" aria-label="Create instance from template form">{% csrf_token %}{% submission_key %}
                         <div class="row">
                             <label class="col-sm-3 col-form-label">{% trans "Name" %}</label>
                             <div class="col-sm-7">
@@ -733,12 +722,6 @@
                                     </select>
                                 </div>
                             </div>
-                            <div class="row">
-                                <label class="col-sm-3 col-form-label">{% trans "Console Password" %}</label>
-                                <div class="col-sm-7">
-                                    <input type="password" class="form-control" name="console_pass" placeholder="{% trans "Console Password" %}" maxlength="64">
-                                </div>
-                            </div>
                             <div class="row">
                                 <label class="col-sm-3 col-form-label">{% trans "Add CD-Rom" %}</label>
                                 <div class="col-sm-7">
@@ -783,7 +766,7 @@
                         <div class="row">
                             <div class="d-grid col-sm-7 offset-3">
                             {% if storages %}
-                                <button type="submit" class="btn btn-primary" name="create" value="1" formnovalidate onclick="showPleaseWaitDialog()">
+                                <button type="submit" class="btn btn-primary" name="create" value="1" formnovalidate>
                                     {% trans "Create" %}
                                 </button>
                             {% else %}

instances/tests/test_create_instance.py

@@ -1,39 +1,61 @@
 """Regression tests for the instance creation request."""
 
-from unittest.mock import patch
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 from computes.models import Compute
 from django.contrib.auth import get_user_model
 from django.test import TestCase
 from django.urls import reverse
 from instances.api.serializers import CreateInstanceSerializer
+from instances.jobs import CreateInstance
 from instances.models import Instance
 from instances.utils import nic_macs
+from operations.jobs import Context
+from operations.models import Operation, acquire
 
 
+wvm_create = MagicMock()  # the page and the job reach the same host
+
+
+def run(op):
+    """The queued job, run as the worker runs it."""
+    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()
+    CreateInstance().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a")))
+
+
+@patch("instances.jobs.wvmCreate", new=wvm_create)
 class CreateInstanceViewTests(TestCase):
-    @patch("instances.views.wvmCreate")
+    @patch("instances.views.wvmCreate", new=wvm_create)
     @patch("instances.views.NewVMForm")
-    def test_successful_create_redirects_after_saving_instance(self, form_class, connection_class):
+    def test_successful_create_redirects_to_the_compute_instances_with_a_queued_job(self, form_class):
-        response, conn = self.create(form_class, connection_class)
+        response, conn = self.create(form_class)
 
-        instance = Instance.objects.get(name="new-vm")
         self.assertRedirects(
-            response, reverse("instances:instance", args=[instance.id]),
+            response, reverse("instances", args=[Compute.objects.get().id]),
             fetch_redirect_response=False,
         )
+        op = Operation.objects.get()
+        self.assertEqual((op.kind, op.state), ("instance.create", Operation.QUEUED))
+        conn.create_instance.assert_not_called()
+        run(op)
         conn.create_instance.assert_called_once()
+        self.assertEqual(conn.create_instance.call_args.kwargs["volumes"][0]["bus"], "virtio")
+        self.assertTrue(Instance.objects.filter(name="new-vm").exists())
 
-    @patch("instances.views.wvmCreate")
+    @patch("instances.views.wvmCreate", new=wvm_create)
     @patch("instances.views.NewVMForm")
-    def test_secure_boot_loader_on_q35_requests_secure_boot(self, form_class, connection_class):
+    def test_secure_boot_loader_on_q35_requests_secure_boot(self, form_class):
-        self.create(form_class, connection_class, firmware="UEFI x86_64: /usr/share/edk2/ovmf/OVMF_CODE.secboot.fd")
+        _, conn = self.create(form_class, firmware="UEFI x86_64: /usr/share/edk2/ovmf/OVMF_CODE.secboot.fd")
-        firmware = connection_class.return_value.create_instance.call_args.kwargs["firmware"]
+        run(Operation.objects.get())
+        firmware = conn.create_instance.call_args.kwargs["firmware"]
         self.assertEqual(firmware["secure"], "yes")
 
-    @patch("instances.views.wvmCreate")
+    @patch("instances.views.wvmCreate", new=wvm_create)
     @patch("instances.views.NewVMForm")
-    def test_invalid_devices_are_refused_before_anything_is_created(self, form_class, connection_class):
+    def test_invalid_devices_are_refused_before_anything_is_created(self, form_class):
         for data in (
             {"add_cdrom": "virtio"},  # no ejectable media
             {"add_cdrom": "fdc"},
@@ -43,14 +65,13 @@ class CreateInstanceViewTests(TestCase):
         ):
             with self.subTest(data):
                 Instance.objects.all().delete()
-                response, conn = self.create(form_class, connection_class, **data)
+                response, conn = self.create(form_class, **data)
                 self.assertEqual(response.status_code, 200)
-                self.assertFalse(Instance.objects.exists())
+                self.assertFalse(Operation.objects.exists())
                 conn.create_volume.assert_not_called()
                 conn.create_instance.assert_not_called()
-                connection_class.reset_mock()
 
-    def create(self, form_class, connection_class, **data):
+    def create(self, form_class, **data):
         user = get_user_model().objects.filter(username="create-test-admin").first() or (
             get_user_model().objects.create_superuser(
                 username="create-test-admin", password="password", email="create@example.com"
@@ -61,7 +82,8 @@ class CreateInstanceViewTests(TestCase):
             name="create-test-compute", hostname="localhost", type=4,
             login="", password="",
         )
-        conn = connection_class.return_value
+        wvm_create.reset_mock()
+        conn = wvm_create.return_value
         conn.get_instances.return_value = []
         conn.get_video_models.return_value = ["vga"]
         conn.get_cache_modes.return_value = {"default": "Default"}
@@ -86,7 +108,7 @@ class CreateInstanceViewTests(TestCase):
             "firmware": "BIOS", "net_model": "virtio", "memory": 1024,
             "vcpu": 1, "vcpu_mode": "host-model", "networks": "default",
             "virtio": True, "listener_addr": "0.0.0.0", "nwfilter": "",
-            "graphics": "vnc", "video": "vga", "console_pass": "",
+            "graphics": "vnc", "video": "vga",
             "mac": "52:54:00:00:00:01", "qemu_ga": False,
             "add_cdrom": "sata", "add_input": "default",
             **data,
@@ -94,7 +116,7 @@ class CreateInstanceViewTests(TestCase):
 
         response = self.client.post(
             reverse("instances:create_instance", args=[compute.id, "x86_64", "q35"]),
-            {"create": "1"},
+            {"create": "1", "device0": "disk", "bus0": "virtio", "submission_key": "key-create01"},
         )
         return response, conn
 
@@ -140,7 +162,7 @@ class NicMacsTestCase(TestCase):
             "name": "api-vm", "vcpu": 1, "vcpu_mode": "host-model", "memory": 512,
             "networks": "default", "mac": "52:54:00:00:00:01,52:54:00:00:00:02", "nwfilter": "",
             "storage": "default", "hdd_size": 1, "cache_mode": "none", "meta_prealloc": False,
-            "virtio": True, "qemu_ga": True, "console_pass": "", "video": "vga", "listener_addr": "0.0.0.0",
+            "virtio": True, "qemu_ga": True, "video": "vga", "listener_addr": "0.0.0.0",
         })
         self.assertFalse(serializer.is_valid())
         self.assertIn("mac", serializer.errors)

instances/tests/test_create_job.py

@@ -0,0 +1,78 @@
+"""VM create runs as a job, from the page and from the API."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import libvirt
+
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+
+from appsettings.settings import get_settings
+from computes.models import Compute
+from instances.jobs import CreateInstance
+from instances.models import Instance
+from operations.jobs import Context
+from operations.models import Operation, acquire
+
+BODY = {"name": "api-vm", "vcpu": 1, "vcpu_mode": "host-model", "memory": 512, "networks": "default", "mac": "",
+        "nwfilter": "", "storage": "default", "hdd_size": 1, "cache_mode": "none", "meta_prealloc": False,
+        "virtio": True, "qemu_ga": True, "video": "vga", "listener_addr": "0.0.0.0", "submission_key": "key-api00001"}
+
+
+class CreateJobTestCase(TransactionTestCase):
+    def setUp(self):
+        get_settings()
+        admin = get_user_model().objects.create_superuser("create_admin", "a@example.com", "pw")
+        self.client.force_login(admin)
+        self.compute = Compute.objects.create(name="create-c", hostname="192.0.2.40", login="root", password="", type=1)
+        patcher = patch("instances.jobs.wvmCreate")
+        self.conn = patcher.start().return_value
+        self.addCleanup(patcher.stop)
+        self.conn.get_instances.return_value = []
+        self.conn.get_cache_modes.return_value = {"none": "Disable cache"}
+        self.conn.get_disk_bus_types.return_value = ["virtio", "sata"]
+        self.conn.create_volume.return_value = "/srv/default/api-vm.qcow2"
+
+    def test_the_api_queues_the_create_and_the_worker_defines_and_records_it(self):
+        url = f"/api/v1/computes/{self.compute.pk}/instances/create/x86_64/q35/"
+        response = self.client.post(url, BODY, content_type="application/json")
+        self.assertEqual(response.status_code, 202)
+        op = Operation.objects.get()
+        self.assertEqual(response["Location"], f"/api/v1/operations/{op.pk}/")
+        self.conn.create_instance.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()
+        CreateInstance().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a")))
+        self.assertEqual(self.conn.create_instance.call_args.kwargs["uuid"], op.intent["uuid"])
+        self.assertTrue(Instance.objects.filter(uuid=op.intent["uuid"], name="api-vm").exists())
+
+    def test_a_refused_create_queues_nothing(self):
+        url = f"/api/v1/computes/{self.compute.pk}/instances/create/x86_64/q35/"
+        response = self.client.post(url, {**BODY, "cache_mode": "writeback"}, content_type="application/json")
+        self.assertEqual(response.status_code, 400)
+        self.assertFalse(Operation.objects.exists())
+
+    def test_a_resubmitted_key_with_other_fields_is_a_conflict(self):
+        url = f"/api/v1/computes/{self.compute.pk}/instances/create/x86_64/q35/"
+        self.client.post(url, BODY, content_type="application/json")
+        response = self.client.post(url, {**BODY, "memory": 1024}, content_type="application/json")
+        self.assertEqual(response.status_code, 409)
+        self.assertEqual(Operation.objects.count(), 1)
+
+    @patch("instances.jobs.wvmConnect")
+    def test_a_create_that_left_nothing_failed(self, conn):
+        op = Operation(params={**BODY, "template": ""},
+                       intent={"compute": self.compute.pk, "uuid": "abababab-0000-1111-2222-333333333333",
+                               "name": "api-vm"})
+        missing = libvirt.libvirtError("no domain")
+        missing.err = (libvirt.VIR_ERR_NO_DOMAIN, libvirt.VIR_FROM_QEMU, "no domain")
+        conn.return_value.wvm.lookupByUUIDString.side_effect = missing
+        pool = MagicMock()
+        pool.listVolumes.return_value = ["other.qcow2"]
+        conn.return_value.wvm.listAllStoragePools.return_value = [pool]
+        self.assertEqual(CreateInstance().reconcile(op).state, Operation.FAILED)
+        pool.listVolumes.return_value = ["api-vm.qcow2"]
+        self.assertEqual(CreateInstance().reconcile(op).state, Operation.UNCERTAIN)

instances/tests/test_create_rollback.py

@@ -1,17 +1,23 @@
+from types import SimpleNamespace
 from unittest.mock import patch
 
+from appsettings.settings import get_settings
 from computes.models import Compute
 from django.contrib.auth import get_user_model
 from django.test import TestCase
-from django.urls import reverse
 from libvirt import libvirtError
 
-FORM = {
+from instances.jobs import CreateInstance
+from operations import jobs
+from operations.jobs import Context
+from operations.models import Operation, acquire
+
+PARAMS = {
     "name": "rb-vm", "firmware": "BIOS", "vcpu": 1, "vcpu_mode": "host-model", "memory": 128,
     "hdd_size": 1, "storage": "default", "mac": "52:54:00:aa:04:01", "networks": "default",
     "nwfilter": "", "net_model": "virtio", "cache_mode": "none", "video": "vga",
-    "listener_addr": "0.0.0.0", "console_pass": "", "add_cdrom": "None", "add_input": "None",
+    "listener_addr": "0.0.0.0", "add_cdrom": "None", "add_input": "None", "virtio": True, "qemu_ga": False,
-    "create": True,
+    "arch": "x86_64", "machine": "q35",
 }
 
 
@@ -19,30 +25,32 @@ class CreateRollbackTestCase(TestCase):
     """Volumes are removed only while the VM is not defined yet."""
 
     def setUp(self):
-        admin = get_user_model().objects.create_superuser("rb_admin", "r@example.com", "pw")
+        get_settings()
-        self.client.force_login(admin)
+        self.admin = get_user_model().objects.create_superuser("rb_admin", "r@example.com", "pw")
-        self.client.raise_request_exception = False
+        self.compute = Compute.objects.create(name="rb", hostname="192.0.2.41", login="", password="", type=4)
-        self.compute = Compute.objects.create(name="rb", hostname="127.0.0.1", login="", password="", type=4)
 
-    def post(self, **conn_setup):
+    def run_job(self, **conn_setup):
-        with patch("instances.views.wvmCreate") as wvm:
+        with patch("instances.jobs.wvmCreate") as wvm:
             conn = wvm.return_value
             conn.get_cache_modes.return_value = {"none": "Disable cache"}
             conn.get_instances.return_value = []
-            conn.get_storages.return_value = ["default"]
-            conn.get_networks.return_value = ["default"]
             conn.create_volume.return_value = "/pool/rb-vm.qcow2"
+            op = jobs.submit("instance.create", self.admin, {**PARAMS, "compute": self.compute.pk}, "key-rollback1")
+            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()
             for attr, value in conn_setup.items():
                 getattr(conn, attr).side_effect = value
-            self.client.post(reverse("instances:create_instance", args=[self.compute.id, "x86_64", "q35"]), FORM)
+            with self.assertRaises(Exception):
+                CreateInstance().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a")))
         return conn
 
     def test_failed_definition_removes_the_new_volume(self):
-        conn = self.post(create_instance=libvirtError("rejected"))
+        conn = self.run_job(create_instance=libvirtError("rejected"))
         conn.delete_volume.assert_called_once_with("/pool/rb-vm.qcow2")
 
     def test_failure_after_the_definition_keeps_the_volume(self):
-        with patch("instances.views.Instance.objects.get_or_create", side_effect=RuntimeError("db down")):
+        with patch("instances.jobs.Instance.objects.get_or_create", side_effect=RuntimeError("db down")):
-            conn = self.post()
+            conn = self.run_job()
         conn.create_instance.assert_called_once()
         conn.delete_volume.assert_not_called()

instances/tests/test_live_create.py

@@ -1,4 +1,5 @@
-"""Live checks that a failed VM creation leaves no volumes behind.
+"""Live checks of VM creation: the queued job defines the VM, and a failed
+one leaves no volumes behind.
 
 Runs only with TEST_LIBVIRT_HOST set (see instances/tests/livetest.py).
 """
@@ -9,17 +10,22 @@ import unittest
 import libvirt
 from computes.models import Compute
 from django.contrib.auth import get_user_model
-from django.test import TestCase
+from django.test import TransactionTestCase
 from django.urls import reverse
 from lxml import etree
 from vrtManager.connection import connection_manager
 
+from operations.models import Operation
+from operations.tests import fakes
+from operations.worker import Worker
+
 from . import livetest
+from .test_live_jobs import wait_for
 
 P = livetest.PREFIX
 
 
-class LiveCreateTestCase(TestCase):
+class LiveCreateTestCase(TransactionTestCase):
     @classmethod
     def setUpClass(cls):
         if not livetest.enabled():
@@ -70,15 +76,26 @@ class LiveCreateTestCase(TestCase):
             "cache_mode": "none",
             "video": "vga",
             "listener_addr": "0.0.0.0",
-            "console_pass": "",
             "add_cdrom": "None",
             "add_input": "None",
             "create": True,
+            "submission_key": "key-" + name,
         }
         data.update(overrides)
-        return self.client.post(
+        response = self.client.post(
             reverse("instances:create_instance", args=[self.compute.id, "x86_64", "q35"]), data
         )
+        self.work()
+        return response
+
+    def work(self):
+        """Run the queued jobs with an in-thread worker."""
+        worker = Worker(concurrency=1, tick=0.1, heartbeat_every=1, progress_every=1, exit_on_lost_lease=False)
+        thread = fakes.start_worker(worker)
+        try:
+            self.assertTrue(wait_for(lambda: not Operation.objects.filter(state__in=Operation.ACTIVE).exists(), 180))
+        finally:
+            fakes.stop_worker(worker, thread)
 
     def volumes(self):
         pool = self.conn.storagePoolLookupByName(livetest.POOL)
@@ -184,15 +201,16 @@ class LiveCreateTestCase(TestCase):
                 "meta_prealloc": False,
                 "virtio": True,
                 "qemu_ga": False,
-                "console_pass": "",
                 "listener_addr": "0.0.0.0",
                 "video": "no-such-model",
                 "add_cdrom": "None",
                 "add_input": "None",
+                "submission_key": "key-cr-api",
             },
             content_type="application/json",
         )
-        self.assertEqual(response.status_code, 400, response.content)
+        self.assertEqual(response.status_code, 202, response.content)
+        self.work()
         self.assertEqual(self.volumes(), set(), "a volume was left behind")
         self.assertFalse(self.domain_exists(P + "cr-api"))
 

instances/tests/test_views.py

@@ -13,7 +13,10 @@ from django.test import RequestFactory, TestCase
 from vrtManager.create import wvmCreate
 from vrtManager.util import randomUUID
 
+from operations.models import Operation
+
 from . import livetest
+from .test_create_instance import run
 from instances.models import Flavor, Instance
 from instances.utils import refr
 
@@ -193,15 +196,16 @@ class InstancesTestCase(TestCase):
                 "graphics": "vnc",
                 "video": "vga",
                 "listener_addr": "0.0.0.0",
-                "console_pass": "",
                 "qemu_ga": False,
                 "virtio": True,
                 "add_cdrom": "None",
                 "add_input": "None",
                 "create": True,
+                "submission_key": "key-create-destroy",
             },
         )
         self.assertEqual(response.status_code, 302)
+        run(Operation.objects.get(kind="instance.create"))
 
         instance_qs: Instance = Instance.objects.filter(name="wvc-test-new")
         self.assertEqual(len(instance_qs), 1)

instances/views.py

@@ -2162,164 +2162,29 @@ def create_instance(request, compute_id, arch, machine):
                     form = NewVMForm(request.POST)
                     if form.is_valid():
                         data = form.cleaned_data
-                        meta_prealloc = bool(data["meta_prealloc"])
-
-                        # Everything is checked before any storage is allocated.
-                        if data["name"] in instances:
-                            raise util.OperationError(_("A virtual machine with this name already exists"))
-                        if Instance.objects.filter(name__exact=data["name"]):
-                            raise util.OperationError(
-                                _("There is an instance with same name. Remove it and try again!")
-                            )
                         if data["hdd_size"] and not data["mac"]:
                             raise util.OperationError(_("No Virtual Machine MAC has been entered"))
-                        if not (data["hdd_size"] or data["template"] or data["images"]):
+                        images = data["images"].split(",") if data["images"] else []
-                            raise util.OperationError(_("First you need to create or select an image"))
+                        raw = {**data, "compute": compute.id, "arch": arch, "machine": machine,
-                        if (
+                               "devices": [[request.POST.get(f"device{i}", ""), request.POST.get(f"bus{i}", "")]
-                            not data["hdd_size"]
+                                           for i in range(len(images))]}
-                            and data["template"]
-                            and conn.get_volume_path(data["name"] + ".img", data["storage"])
-                        ):
-                            raise util.OperationError(
-                                _("Image has already exist. Please check volumes or change instance name")
-                            )
-                        if data["cache_mode"] not in conn.get_cache_modes():
-                            raise util.OperationError(_("Invalid cache mode"))
-                        if data["add_cdrom"] != "None" and data["add_cdrom"] not in cdrom_buses:
-                            raise util.OperationError(
-                                _("A CD-ROM cannot use the %(bus)s bus") % {"bus": data["add_cdrom"]}
-                            )
                         try:
-                            utils.nic_macs(data["mac"], data["networks"])
+                            op = jobs.submit("instance.create", request.user, raw,
-                        except ValueError as err:
+                                             request.POST.get("submission_key", ""))
-                            raise util.OperationError(str(err))
+                        except claims.Busy as busy:
-
+                            messages.error(request, busy_message(busy, request.user))
-                        firmware = dict()
+                        except jobs.SubmissionConflict as err:
-                        if "UEFI" in data["firmware"]:
+                            messages.error(request, str(err))
-                            if ":" not in data["firmware"]:
-                                raise util.OperationError(_("Invalid firmware"))
-                            firmware["loader"] = data["firmware"].split(":")[1].strip()
-                            firmware["secure"] = "no"
-                            firmware["readonly"] = "yes"
-                            firmware["type"] = "pflash"
-                            if "secboot" in firmware["loader"]:
-                                # A secure boot loader needs secure='yes' and SMM, on q35 only
-                                if machine != "q35":
-                                    messages.warning(
-                                        request,
-                                        "Changing machine type from '%s' to 'q35' "
-                                        "which is required for UEFI secure boot." % machine,
-                                    )
-                                    machine = "q35"
-                                firmware["secure"] = "yes"
-
-                        if data["net_model"] == "default":
-                            data["net_model"] = "virtio"
-
-                        def volume(path, device="disk", bus=default_bus):
-                            return {
-                                "path": path,
-                                "type": conn.get_volume_format_type(path),
-                                "device": device,
-                                "bus": bus,
-                                "scsi_model": default_scsi_disk_model if bus == "scsi" else None,
-                                "cache_mode": data["cache_mode"],
-                                "discard_mode": default_discard,
-                                "detect_zeroes_mode": default_zeroes,
-                                "io_mode": default_io,
-                            }
-
-                        created = []  # volumes this request allocated
-                        defined = False
-                        volume_list = []
-                        uuid = util.randomUUID()
-                        try:
-                            if data["hdd_size"]:
-                                path = conn.create_volume(
-                                    data["storage"],
-                                    data["name"],
-                                    data["hdd_size"],
-                                    default_disk_format,
-                                    meta_prealloc,
-                                    default_disk_owner_uid,
-                                    default_disk_owner_gid,
-                                )
-                                created.append(path)
-                                volume_list.append(volume(path))
-                            elif data["template"]:
-                                path = conn.clone_from_template(
-                                    data["name"],
-                                    conn.get_volume_path(data["template"]),
-                                    data["storage"],
-                                    meta_prealloc,
-                                    default_disk_owner_uid,
-                                    default_disk_owner_gid,
-                                )
-                                created.append(path)
-                                volume_list.append(volume(path))
-                            else:
-                                for idx, vol in enumerate(data["images"].split(",")):
-                                    volume_list.append(volume(
-                                        conn.get_volume_path(vol),
-                                        request.POST.get("device" + str(idx), ""),
-                                        request.POST.get("bus" + str(idx), ""),
-                                    ))
-
-                            with libvirt_compute_lock(compute):
-                                conn.create_instance(
-                                    name=data["name"],
-                                    memory=data["memory"],
-                                    vcpu=data["vcpu"],
-                                    vcpu_mode=data["vcpu_mode"],
-                                    uuid=uuid,
-                                    arch=arch,
-                                    machine=machine,
-                                    firmware=firmware,
-                                    volumes=volume_list,
-                                    networks=data["networks"],
-                                    virtio=data["virtio"],
-                                    listener_addr=data["listener_addr"],
-                                    nwfilter=data["nwfilter"],
-                                    net_model=data["net_model"],
-                                    video=data["video"],
-                                    console_pass=data["console_pass"],
-                                    mac=data["mac"],
-                                    qemu_ga=data["qemu_ga"],
-                                    add_cdrom=data["add_cdrom"],
-                                    add_input=data["add_input"],
-                                )
-                                defined = True  # from here on the VM uses the volumes
-                                create_instance = Instance.objects.get_or_create(
-                                    compute_id=compute_id, uuid=uuid, defaults={"name": data["name"]}
-                                )[0]
-                                if create_instance.name != data["name"]:
-                                    create_instance.name = data["name"]
-                                    create_instance.save(update_fields=["name"])
-                        except Exception as err:
-                            leftovers = []
-                            for path in [] if defined else created:
-                                try:
-                                    conn.delete_volume(path)
-                                except libvirtError:
-                                    leftovers.append(path)
-                            if leftovers:
-                                messages.error(
-                                    request,
-                                    _("Remove these volumes by hand: %(paths)s") % {"paths": ", ".join(leftovers)},
-                                )
-                            if not isinstance(err, libvirtError):
-                                raise  # unexpected: the volumes are gone, the error stays visible
-                            messages.error(request, err)
                         else:
-                            msg = _("Instance is created")
+                            if op.params["machine"] != machine:
-                            messages.success(request, msg)
+                                messages.warning(
-                            addlogmsg(
+                                    request,
-                                request.user.username,
+                                    "Changing machine type from '%s' to 'q35' "
-                                create_instance.compute.name,
+                                    "which is required for UEFI secure boot." % machine,
-                                create_instance.name,
+                                )
-                                msg,
+                            messages.info(request, _("The VM is being created; follow it in the job panel."))
-                            )
+                            jobs.remember_job(request, op)
-                            return redirect(reverse("instances:instance", args=[create_instance.id]))
+                            return redirect(reverse("instances", args=[compute.id]))
             conn.close()
     except libvirtError as lib_err:
         messages.error(request, lib_err)

webvirtcloud/middleware.py

@@ -3,7 +3,7 @@ from django.http import JsonResponse
 from django.shortcuts import redirect, render
 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, public_error_text, summary
 
 API_PREFIX = "/api/"
@@ -37,6 +37,8 @@ class ExceptionMiddleware:
                 return JsonResponse({"detail": busy_message(exception, request.user),
                                      "operation": summary(exception.blocker, request.user) if exception.blocker else None},
                                     status=409)
+            if isinstance(exception, jobs.SubmissionConflict):
+                return JsonResponse({"detail": str(exception)}, status=409)
             if isinstance(exception, TimeoutError):
                 return JsonResponse({"detail": "Operation is busy. Please retry."}, status=409)
             return None