From 95562544eb319e9242caf5958bd0fd41364e9705 Mon Sep 17 00:00:00 2001
From: catborise <catborise@gmail.com>
Date: Sat, 31 Oct 2026 19:56:02 +0300
Subject: [PATCH] feat(instances): Add PCIe slots for a VM that has none free

An existing q35 VM whose PCIe root ports are all used takes no new
virtio disk or network device while it runs, and the only way to give
it more was editing its XML.

The Disks tab of a PCIe machine now counts its slots, "PCIe slots: 0
free of 7", in red when none is free, with what its next start will
have when that differs, and Add PCIe slots adds four root ports to its
persistent definition: libvirt numbers and places them, and a running
VM gets them at its next start. The no-slot advice of Add disk and Add
network device points at it.
---
 instances/models.py                           | 10 +++
 .../templates/instances/settings_tab.html     | 15 ++++
 instances/tests/test_pci_slots.py             | 79 +++++++++++++++++++
 instances/tests/test_pending_edits.py         | 13 +++
 instances/urls.py                             |  1 +
 instances/views.py                            | 34 +++++++-
 vrtManager/instance.py                        | 20 +++++
 7 files changed, 168 insertions(+), 4 deletions(-)

diff --git a/instances/models.py b/instances/models.py
index 91370ce9..a840a3df 100644
--- a/instances/models.py
+++ b/instances/models.py
@@ -161,6 +161,16 @@ class Instance(models.Model):
             net["target"] = live_targets.get(net["mac"], "")
         return networks
 
+    @cached_property
+    def pcie_slots(self) -> tuple[int, int] | None:
+        """(PCIe slots, free ones) the VM has now; None: not a PCIe machine."""
+        return self.proxy.get_pcie_slots()
+
+    @cached_property
+    def config_pcie_slots(self) -> tuple[int, int] | None:
+        """The same at its next start, which added slots wait for while it runs."""
+        return self.proxy.get_pcie_slots(config=True)
+
     @cached_property
     def config_disks(self) -> list[dict]:
         """Disks of the persistent definition, which the disk list, its edit
diff --git a/instances/templates/instances/settings_tab.html b/instances/templates/instances/settings_tab.html
index db38abaa..26c576fb 100644
--- a/instances/templates/instances/settings_tab.html
+++ b/instances/templates/instances/settings_tab.html
@@ -274,6 +274,21 @@
                 {% endfor %}
                 </div>
                 </div>
+                {% if instance.pcie_slots %}
+                {# a PCIe machine gives each virtio disk and network device a slot; a running VM gets no new one #}
+                {% with total=instance.pcie_slots.0 free=instance.pcie_slots.1 next=instance.config_pcie_slots %}
+                <div class="d-flex flex-wrap align-items-center justify-content-between gap-2 small mt-3">
+                    <span class="{% if not free %}text-danger{% else %}text-body-secondary{% endif %}">
+                        {% blocktrans %}PCIe slots: {{ free }} free of {{ total }}{% endblocktrans %}{% if next and next.1 != free %} · {% blocktrans with next_free=next.1 %}{{ next_free }} free from the next start{% endblocktrans %}{% endif %}
+                    </span>
+                    {% if request.user.is_superuser %}
+                    <form action="{% url 'instances:add_pcie_slots' instance.id %}" method="post">{% csrf_token %}
+                        <button type="submit" class="btn btn-sm btn-outline-secondary">{% trans "Add PCIe slots" %}</button>
+                    </form>
+                    {% endif %}
+                </div>
+                {% endwith %}
+                {% endif %}
             </section>
         </div>
         <div role="tabpanel" class="tab-pane" id="network">
diff --git a/instances/tests/test_pci_slots.py b/instances/tests/test_pci_slots.py
index 0efeac87..32a30739 100644
--- a/instances/tests/test_pci_slots.py
+++ b/instances/tests/test_pci_slots.py
@@ -102,3 +102,82 @@ class NoPciSlotTests(TestCase):
                                        NO_SLOT)
         self.assertEqual(len(texts), 1)
         self.assertIn("shut it down", texts[0])
+
+
+DOMAIN = """<domain><devices>
+<controller type='pci' index='0' model='pcie-root'/>
+<controller type='pci' index='1' model='pcie-root-port'><address type='pci' bus='0x00' slot='0x02'/></controller>
+<controller type='pci' index='2' model='pcie-root-port'><address type='pci' bus='0x00' slot='0x02' function='0x1'/></controller>
+<controller type='pci' index='3' model='pcie-root-port'><address type='pci' bus='0x00' slot='0x02' function='0x2'/></controller>
+<interface type='network'><address type='pci' bus='0x01' slot='0x00'/></interface>
+<disk type='file' device='disk'><address type='pci' bus='0x03' slot='0x00'/></disk>
+<video><address type='pci' bus='0x00' slot='0x01'/></video>
+</devices></domain>"""
+
+
+def proxy(xml=DOMAIN):
+    from vrtManager.instance import wvmInstance
+
+    p = wvmInstance.__new__(wvmInstance)
+    p._XMLDesc = MagicMock(return_value=xml)
+    p._defineXML = MagicMock()
+    return p
+
+
+class PcieSlotsTests(SimpleTestCase):
+    def test_a_port_no_device_sits_behind_is_free(self):
+        self.assertEqual(proxy().get_pcie_slots(), (3, 1))
+
+    def test_a_port_numbered_above_nine_is_matched_to_its_hex_bus(self):
+        ports = "".join(f"<controller type='pci' index='{i}' model='pcie-root-port'/>" for i in range(1, 13))
+        xml = (f"<domain><devices>{ports}<disk type='file' device='disk'><address type='pci' bus='0x0a'/></disk>"
+               "<interface type='network'><address type='pci' bus='0x0c'/></interface></devices></domain>")
+        self.assertEqual(proxy(xml).get_pcie_slots(), (12, 10))
+
+    def test_a_machine_without_ports_has_none(self):
+        self.assertIsNone(proxy("<domain><devices><video/></devices></domain>").get_pcie_slots())
+
+    def test_added_ports_are_bare_for_libvirt_to_number(self):
+        p = proxy()
+        p.add_pcie_slots(4)
+        (defined,), _ = p._defineXML.call_args
+        ports = etree.fromstring(defined).findall("devices/controller[@model='pcie-root-port']")
+        self.assertEqual([port.get("index") for port in ports], ["1", "2", "3", None, None, None, None])
+
+
+class AddPcieSlotsViewTests(TestCase):
+    def setUp(self):
+        self.admin = get_user_model().objects.create_superuser("pa-admin", "pa@example.com", "x")
+        compute = Compute.objects.create(name="pa", hostname="127.0.0.1:1", login="root", password="", type=1)
+        self.instance = Instance.objects.create(compute=compute, name="pa-vm",
+                                                uuid="aaaaaaaa-8888-2222-3333-444444444444")
+        self.url = reverse("instances:add_pcie_slots", args=[self.instance.id])
+
+    def post(self, status=1, slots=(7, 0), user=None):
+        self.client.force_login(user or self.admin)
+        with patch("instances.models.wvmInstance") as wvm:
+            wvm.return_value.get_status.return_value = status
+            wvm.return_value.get_pcie_slots.return_value = slots
+            response = self.client.post(self.url, HTTP_REFERER=reverse("instances:instance", args=[self.instance.id]))
+        return response, wvm.return_value.add_pcie_slots, [str(m) for m in get_messages(response.wsgi_request)]
+
+    def test_a_running_vm_gets_them_at_its_next_start(self):
+        _response, add, texts = self.post(status=1)
+        add.assert_called_once_with(4)
+        self.assertEqual(texts, ["4 PCIe slots added: the VM gets them at its next start"])
+        self.assertTrue(Logs.objects.get(instance="pa-vm").success)
+
+    def test_a_shut_off_vm_has_them_at_once(self):
+        _response, _add, texts = self.post(status=5)
+        self.assertEqual(texts, ["4 PCIe slots added"])
+
+    def test_a_machine_without_pcie_slots_is_refused(self):
+        _response, add, texts = self.post(slots=None)
+        add.assert_not_called()
+        self.assertEqual(texts, ["The VM's machine has no PCIe slots to add"])
+
+    def test_only_a_superuser_adds_them(self):
+        user = get_user_model().objects.create_user("pa-user", password="x")
+        response, add, _texts = self.post(user=user)
+        self.assertEqual(response.status_code, 403)
+        add.assert_not_called()
diff --git a/instances/tests/test_pending_edits.py b/instances/tests/test_pending_edits.py
index 6b802886..7d27cbfa 100644
--- a/instances/tests/test_pending_edits.py
+++ b/instances/tests/test_pending_edits.py
@@ -154,6 +154,19 @@ class PendingEditsTestCase(TestCase):
                 doc = self.page(live=disks, config=disks, get_storages=(pools,) * 2)
                 self.assertEqual(doc.xpath("//div[@id='movedisk0']//button[@type='submit']/@disabled"), disabled)
 
+    def test_the_disks_tab_counts_free_pcie_slots_now_and_from_the_next_start(self):
+        doc = self.page(get_pcie_slots=((7, 0), (12, 5)))
+        (form,) = doc.xpath("//form[@action=$a]", a=reverse("instances:add_pcie_slots", args=[self.instance.id]))
+        line = " ".join(form.getparent().xpath("./span")[0].text_content().split())
+        self.assertEqual(line, "PCIe slots: 0 free of 7 · 5 free from the next start")
+        # a persistent definition without root ports shows the running VM's count only
+        doc = self.page(get_pcie_slots=((7, 2), None))
+        (form,) = doc.xpath("//form[@action=$a]", a=reverse("instances:add_pcie_slots", args=[self.instance.id]))
+        self.assertEqual(" ".join(form.getparent().xpath("./span")[0].text_content().split()), "PCIe slots: 2 free of 7")
+        # not a PCIe machine: no line
+        doc = self.page(get_pcie_slots=(None, None))
+        self.assertFalse(doc.xpath("//form[@action=$a]", a=reverse("instances:add_pcie_slots", args=[self.instance.id])))
+
     def test_the_clone_form_lists_the_persistent_devices(self):
         nic = {"type": "network", "mac": "52:54:10:00:00:01", "nic": "default", "state": "up", "model": "virtio",
                "ipv4": None, "ipv6": None, "filterref": "", "target": "", "inbound": [], "outbound": []}
diff --git a/instances/urls.py b/instances/urls.py
index ac7170da..092c44d8 100755
--- a/instances/urls.py
+++ b/instances/urls.py
@@ -35,6 +35,7 @@ urlpatterns = [
     path("<int:pk>/edit_volume/", views.edit_volume, name="edit_volume"),
     path("<int:pk>/detach_vol/", views.detach_vol, name="detach_vol"),
     path("<int:pk>/add_cdrom/", views.add_cdrom, name="add_cdrom"),
+    path("<int:pk>/add_pcie_slots/", views.add_pcie_slots, name="add_pcie_slots"),
     path("<int:pk>/detach_cdrom/<str:dev>/", views.detach_cdrom, name="detach_cdrom"),
     path("<int:pk>/unmount_iso/", views.unmount_iso, name="unmount_iso"),
     path("<int:pk>/mount_iso/", views.mount_iso, name="mount_iso"),
diff --git a/instances/views.py b/instances/views.py
index c04afb13..3e45c706 100755
--- a/instances/views.py
+++ b/instances/views.py
@@ -715,10 +715,12 @@ def invalid_disk_options(
 PCI_SLOTS_FULL = "No more available PCI slots"
 # a disk on the scsi bus needs no slot of its own when the VM already has a SCSI controller
 DISK_NO_PCI_SLOT = _("The running VM has no free PCI slot for another disk on this bus: choose the scsi bus "
-                     "if the VM has a SCSI controller, or shut the VM down and add the disk then. A new volume "
-                     "stays in its pool, to attach as an existing volume.")
+                     "if the VM has a SCSI controller, or shut the VM down and add the disk then; Add PCIe slots "
+                     "on the Disks tab gives it free ones from its next start. A new volume stays in its pool, "
+                     "to attach as an existing volume.")
 SCSI_NO_PCI_SLOT = _("The running VM has no free PCI slot for the SCSI controller the disk needs: shut the VM "
-                     "down and add the disk then. A new volume stays in its pool, to attach as an existing volume.")
+                     "down and add the disk then; Add PCIe slots on the Disks tab gives it free ones from its next "
+                     "start. A new volume stays in its pool, to attach as an existing volume.")
 
 
 def no_pci_slot(request, err, advice):
@@ -1334,6 +1336,29 @@ def add_new_vol(request, pk):
     return _back(request, pk, "disks")
 
 
+# the PCIe slots the Add slots button adds at a time
+PCIE_SLOTS_ADDED = 4
+
+
+@require_POST
+@superuser_only
+@serialize_instance_mutation
+def add_pcie_slots(request, pk):
+    """Free PCIe slots for devices added later: a running VM gets them at its next start."""
+    instance = get_instance(request.user, pk, perm_type="change")
+    if instance.pcie_slots is None:
+        messages.error(request, _("The VM's machine has no PCIe slots to add"))
+        return _back(request, pk, "disks")
+    with audit_vm(request, instance, f"add {PCIE_SLOTS_ADDED} PCIe slots"):
+        instance.proxy.add_pcie_slots(PCIE_SLOTS_ADDED)
+    if instance.proxy.get_status() == 5:
+        messages.success(request, _("%(count)s PCIe slots added") % {"count": PCIE_SLOTS_ADDED})
+    else:
+        messages.success(request, _("%(count)s PCIe slots added: the VM gets them at its next start")
+                         % {"count": PCIE_SLOTS_ADDED})
+    return _back(request, pk, "disks")
+
+
 @require_POST
 @superuser_only
 @serialize_instance_mutation
@@ -1947,7 +1972,8 @@ def add_network(request, pk):
             instance.proxy.add_network(mac, source, source_type, model=model, nwfilter=nwfilter)
     except libvirtError as err:
         if not no_pci_slot(request, err, _("The running VM has no free PCI slot for another network device: "
-                                           "shut it down and add the device then.")):
+                                           "shut it down and add the device then; Add PCIe slots on the Disks "
+                                           "tab gives it free ones from its next start.")):
             raise
     return _back(request, pk, "network")
 
diff --git a/vrtManager/instance.py b/vrtManager/instance.py
index d84eaa53..00c6bebe 100644
--- a/vrtManager/instance.py
+++ b/vrtManager/instance.py
@@ -706,6 +706,26 @@ class wvmInstance(wvmConnect):
     def get_machine_type(self):
         return util.get_xml_path(self._XMLDesc(0), "/domain/os/type/@machine")
 
+    def get_pcie_slots(self, config=False):
+        """(PCIe root ports, free ones) of the running or the persistent
+        definition, None when it has none (not a PCIe machine): each PCIe
+        device takes a port, the one whose index is its address's bus."""
+        tree = etree.fromstring(self._XMLDesc(PERSISTENT_XML if config else 0))
+        ports = tree.xpath("./devices/controller[@type='pci'][@model='pcie-root-port']/@index")
+        if not ports:
+            return None
+        used = {int(bus, 16) for bus in tree.xpath("./devices/*/address[@type='pci']/@bus")}
+        return len(ports), sum(int(index) not in used for index in ports)
+
+    def add_pcie_slots(self, count):
+        """Add count PCIe root ports to the persistent definition, which libvirt
+        numbers and places; a running VM gets them at its next start."""
+        tree = etree.fromstring(self._XMLDesc(PERSISTENT_XML))
+        devices = tree.find("devices")
+        for _ in range(count):
+            etree.SubElement(devices, "controller", type="pci", model="pcie-root-port")
+        self._defineXML(etree.tostring(tree).decode())
+
     def get_dom_emulator(self):
         return util.get_xml_path(self._XMLDesc(0), "/domain/devices/emulator")