From d9f9ebab587de7267228e3452146ca56dc246e72 Mon Sep 17 00:00:00 2001 From: catborise Date: Sun, 25 Oct 2026 12:15:23 +0300 Subject: [PATCH] refactor(computes): the Computes page reads capacity through vrtManager The capacity read lived in its own module with raw libvirt calls and a thread pool with a deadline, beside helpers that already read the same things. It now uses them: - wvmHostDetails.get_capacity() sums the vCPUs and memory of all and of the running VMs from get_host_instances(), against the host's info. - get_storages_info() also gives each pool's free space. - Compute.capacity joins the two, like cpu_count and ram_size, and is None when the host is down or the read fails. The page reads the hosts one after another, as it already connected to each of them; a host that cannot be reached is given up after the connect timeout. computes/capacity.py and its thread handling are gone. The page looks and sorts as before. --- computes/capacity.py | 82 ----------------- computes/models.py | 17 +++- computes/templates/computes/list.html | 2 +- computes/tests/test_capacity.py | 124 +++++++------------------- computes/views.py | 13 ++- vrtManager/hostdetails.py | 17 +++- vrtManager/storage.py | 3 +- vrtManager/tests/test_host_reads.py | 41 ++++++++- 8 files changed, 114 insertions(+), 185 deletions(-) delete mode 100644 computes/capacity.py diff --git a/computes/capacity.py b/computes/capacity.py deleted file mode 100644 index d5944b22..00000000 --- a/computes/capacity.py +++ /dev/null @@ -1,82 +0,0 @@ -""" -What each compute has given its VMs: vCPUs and memory of all VMs (a -shut-off VM takes its share again when it starts) and of the running ones, -against the host's CPUs and memory, and the free space of its active -storage pools. Read live from libvirt for the Computes page. -""" - -import threading -import time -from concurrent import futures - -from libvirt import VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE, VIR_DOMAIN_SHUTOFF, libvirtError - -MAX_WORKERS = 8 - -# A libvirt call may hang: its thread outlives the page. Until it ends, its -# compute is not read again, so the threads stay at one per compute. -_reading = set() -_reading_lock = threading.Lock() - - -def read(compute): - """The compute's capacity, or None when it cannot be reached. A VM that - goes while it is read is skipped.""" - if compute.status is not True: - return None - conn = compute.connection - _arch, memory_mib, cpus, *_ = conn.getInfo() - vcpus = {"all": 0, "running": 0} - assigned = {"all": 0, "running": 0} - for dom in conn.listAllDomains(0): - try: - state, _max_memory, memory_kib, vcpu_count, _cpu_time = dom.info() - except libvirtError: - continue - for share in ("all", "running") if state != VIR_DOMAIN_SHUTOFF else ("all",): - vcpus[share] += vcpu_count - assigned[share] += memory_kib * 1024 - pools = [] - for pool in conn.listAllStoragePools(VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE): - try: - _state, size, _allocation, available = pool.info() - except libvirtError: - continue - pools.append((pool.name(), size, available)) - return {"cpus": cpus, "memory": memory_mib * 1024 * 1024, "vcpus": vcpus, "assigned": assigned, - "pools": sorted(pools)} - - -def _read_once(compute): - try: - return read(compute) - finally: - with _reading_lock: - _reading.discard(compute.id) - - -def read_all(computes, deadline=10.0): - """{compute id: capacity} of the computes read by the deadline (seconds), - in parallel; a compute down, failing, late, or still being read for an - earlier page is left out.""" - with _reading_lock: - computes = [compute for compute in computes if compute.id not in _reading] - _reading.update(compute.id for compute in computes) - pool = futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) - end = time.monotonic() + deadline - try: - reads = {pool.submit(_read_once, compute): compute.id for compute in computes} - done, _late = futures.wait(reads, timeout=max(0, end - time.monotonic())) - found = {} - for future in done: - try: - result = future.result() - except libvirtError: - continue - if result is not None: - found[reads[future]] = result - return found - finally: - pool.shutdown(wait=False, cancel_futures=True) - with _reading_lock: # a read cancelled before it started ends nothing itself - _reading.difference_update(cid for future, cid in reads.items() if future.cancelled()) diff --git a/computes/models.py b/computes/models.py index a8c13433..deb3ca45 100644 --- a/computes/models.py +++ b/computes/models.py @@ -4,11 +4,12 @@ from django.db import transaction from django.db.models import CharField, IntegerField, Model from django.utils.functional import cached_property from django.utils.translation import gettext_lazy as _ -from libvirt import virConnect +from libvirt import libvirtError, virConnect from computes.validators import validate_hostname, validate_login, validate_migration_address from vrtManager.connection import connection_manager from vrtManager.hostdetails import wvmHostDetails +from vrtManager.storage import wvmStorages class Compute(Model): @@ -87,6 +88,20 @@ class Compute(Model): def ram_usage(self): return self.proxy.get_memory_usage()["percent"] + @cached_property + def capacity(self): + """What the host has given its VMs (wvmHostDetails.get_capacity) and + its active storage pools, or None when it cannot be read.""" + if self.status is not True: + return None + try: + given = self.proxy.get_capacity() + storages = wvmStorages(self.hostname, self.login, self.password, self.type) + given["pools"] = [pool for pool in storages.get_storages_info() if pool["status"]] + except libvirtError: + return None + return given + def __str__(self): return self.name diff --git a/computes/templates/computes/list.html b/computes/templates/computes/list.html index 4b418a4a..e824498b 100644 --- a/computes/templates/computes/list.html +++ b/computes/templates/computes/list.html @@ -81,7 +81,7 @@ {% endif %} {% else %} - {% if compute.status is True %}{% trans "The capacity could not be read in time." %}{% else %}{% trans "No capacity while the host cannot be reached." %}{% endif %} + {% if compute.status is True %}{% trans "The capacity could not be read." %}{% else %}{% trans "No capacity while the host cannot be reached." %}{% endif %} {% endif %}
diff --git a/computes/tests/test_capacity.py b/computes/tests/test_capacity.py index f8b07c28..26b41c20 100644 --- a/computes/tests/test_capacity.py +++ b/computes/tests/test_capacity.py @@ -1,13 +1,8 @@ -"""What each compute has given its VMs (computes/capacity.py) and the -Computes page that shows it.""" +"""What each compute has given its VMs (Compute.capacity) and the Computes +page that shows it.""" -import threading -import time -import unittest -from types import SimpleNamespace from unittest.mock import patch -from computes import capacity from computes.models import Compute from django.contrib.auth import get_user_model from django.test import TestCase @@ -18,85 +13,33 @@ from lxml import html GIB = 1024 ** 3 -def domain(state, memory_mib, vcpus): - info = [state, memory_mib * 1024, memory_mib * 1024, vcpus, 0] - return SimpleNamespace(info=lambda: info) +def pool(name, size_gib, free_gib, status=1): + return {"name": name, "status": status, "size": size_gib * GIB, "free": free_gib * GIB} -def pool(name, capacity_gib, available_gib): - return SimpleNamespace(name=lambda: name, info=lambda: [2, capacity_gib * GIB, 0, available_gib * GIB]) - - -def compute(name, domains=(), pools=(), cpus=16, memory_mib=65536, down=False, pk=1): - connection = SimpleNamespace( - getInfo=lambda: ["x86_64", memory_mib, cpus, 2400, 1, 1, 8, 2], - listAllDomains=lambda flags=0: list(domains), - listAllStoragePools=lambda flags=0: list(pools), - ) - return SimpleNamespace(id=pk, name=name, status=Exception("down") if down else True, connection=connection) - - -class ReadTests(unittest.TestCase): +class CapacityTests(TestCase): def setUp(self): - capacity._reading.clear() # a stuck read of an earlier test may not have ended yet + self.compute = Compute(name="host-one", hostname="192.0.2.1", login="u", password="p", type=1) + self.compute.status = True - def test_a_host_gives_its_vms_cpus_and_memory(self): - c = compute("c1", [domain(1, 4096, 4), domain(1, 2048, 2), domain(5, 8192, 8)], - [pool("default", 500, 120), pool("iso", 100, 90)]) - result = capacity.read(c) - self.assertEqual(result["cpus"], 16) - self.assertEqual(result["memory"], 64 * GIB) - self.assertEqual(result["vcpus"], {"all": 14, "running": 6}) - self.assertEqual(result["assigned"], {"all": 14 * GIB, "running": 6 * GIB}) - self.assertEqual(result["pools"], [("default", 500 * GIB, 120 * GIB), ("iso", 100 * GIB, 90 * GIB)]) + @patch("computes.models.wvmStorages") + @patch("computes.models.wvmHostDetails") + def test_a_host_tells_what_it_gave_and_its_active_pools(self, details, storages): + details.return_value.get_capacity.return_value = {"cpus": 16} + storages.return_value.get_storages_info.return_value = [pool("default", 500, 120), pool("off", 10, 0, status=0)] + self.assertEqual(self.compute.capacity, {"cpus": 16, "pools": [pool("default", 500, 120)]}) - def test_a_vm_that_goes_meanwhile_is_skipped(self): - def gone(): - raise libvirtError("Domain not found") + @patch("computes.models.wvmHostDetails") + def test_a_host_whose_read_fails_has_no_capacity(self, details): + # a VM that goes while the host is read, for one + details.return_value.get_capacity.side_effect = libvirtError("Domain not found") + self.assertIsNone(self.compute.capacity) - c = compute("c1", [SimpleNamespace(info=gone), domain(1, 1024, 1)]) - self.assertEqual(capacity.read(c)["vcpus"], {"all": 1, "running": 1}) - - def test_a_down_host_has_no_capacity(self): - self.assertIsNone(capacity.read(compute("c1", down=True))) - - def test_hosts_are_read_together_by_the_deadline(self): - slow = threading.Event() - try: - stuck = compute("c2", pk=2) - stuck.connection.getInfo = lambda: slow.wait() and None - started = time.monotonic() - found = capacity.read_all([compute("c1"), stuck], deadline=0.3) - self.assertLess(time.monotonic() - started, 1) - self.assertEqual(set(found), {1}) - finally: - slow.set() - - - def test_a_host_still_being_read_is_not_read_again(self): - # a stuck read keeps its thread; a second page load must not add another - slow, calls = threading.Event(), [] - try: - stuck = compute("c2", pk=2) - stuck.connection.getInfo = lambda: calls.append(1) or slow.wait() and None - for _ in range(2): - self.assertEqual(set(capacity.read_all([compute("c1"), stuck], deadline=0.2)), {1}) - self.assertEqual(len(calls), 1) - finally: - slow.set() - - - def test_a_host_whose_read_never_started_is_read_next_time(self): - # more hosts than threads: a read still queued at the deadline is cancelled - slow = threading.Event() - try: - stuck = compute("c1") - stuck.connection.getInfo = lambda: slow.wait() and None - with patch.object(capacity, "MAX_WORKERS", 1): - self.assertEqual(capacity.read_all([stuck, compute("c2", pk=2)], deadline=0.2), {}) - self.assertEqual(set(capacity.read_all([compute("c2", pk=2)], deadline=1)), {2}) - finally: - slow.set() + @patch("computes.models.wvmHostDetails") + def test_a_down_host_is_not_read(self, details): + self.compute.status = False + self.assertIsNone(self.compute.capacity) + details.assert_not_called() class ComputesPageTests(TestCase): @@ -107,8 +50,8 @@ class ComputesPageTests(TestCase): Compute.objects.create(name="host-two", hostname="192.0.2.2", login="u", password="p", type=1) found = {c1.id: {"cpus": 16, "memory": 64 * GIB, "vcpus": {"all": 40, "running": 24}, "assigned": {"all": 48 * GIB, "running": 32 * GIB}, - "pools": [("default", 500 * GIB, 120 * GIB)]}} - with patch("computes.views.capacity.read_all", return_value=found), \ + "pools": [pool("default", 500, 120)]}} + with patch.object(Compute, "capacity", new=property(lambda c: found.get(c.id))), \ patch.object(Compute, "status", new=True): page = html.fromstring(self.client.get(reverse("computes")).content) headers = [" ".join(th.text_content().split()) for th in page.xpath("//thead//th")] @@ -121,19 +64,18 @@ class ComputesPageTests(TestCase): self.assertIn("48.0 GB given of 64.0 GB", cells[2]) self.assertIn("32.0 GB of it running", cells[2]) self.assertIn("120.0 GB free of 500.0 GB", cells[3]) - # host-two answered nothing in time: one cell says so - self.assertIn("could not be read in time", " ".join(two.text_content().split())) + # host-two could not be read: one cell says so + self.assertIn("capacity could not be read", " ".join(two.text_content().split())) def test_the_default_pool_shows_and_the_others_open_on_a_click(self): admin = get_user_model().objects.create_superuser("cap-pools", "pools@example.com", "pw") self.client.force_login(admin) - c1 = Compute.objects.create(name="host-one", hostname="192.0.2.1", login="u", password="p", type=1) - pools = [("backup", 900 * GIB, 800 * GIB), ("default", 500 * GIB, 120 * GIB), - ("iso", 100 * GIB, 90 * GIB), ("nfs", 2000 * GIB, 1000 * GIB)] - found = {c1.id: {"cpus": 16, "memory": 64 * GIB, "vcpus": {"all": 8, "running": 8}, - "assigned": {"all": 8 * GIB, "running": 8 * GIB}, "pools": pools}} - with patch("computes.views.capacity.read_all", return_value=found), \ - patch.object(Compute, "status", new=True): + Compute.objects.create(name="host-one", hostname="192.0.2.1", login="u", password="p", type=1) + # libvirt lists the pools in no order + pools = [pool("nfs", 2000, 1000), pool("default", 500, 120), pool("iso", 100, 90), pool("backup", 900, 800)] + given = {"cpus": 16, "memory": 64 * GIB, "vcpus": {"all": 8, "running": 8}, + "assigned": {"all": 8 * GIB, "running": 8 * GIB}, "pools": pools} + with patch.object(Compute, "capacity", new=given), patch.object(Compute, "status", new=True): page = html.fromstring(self.client.get(reverse("computes")).content) (storage,) = page.xpath("//tbody/tr/td[4]") shown = " ".join(" ".join(storage.xpath("./div//text()")).split()) diff --git a/computes/views.py b/computes/views.py index f51e13dc..dd678e67 100644 --- a/computes/views.py +++ b/computes/views.py @@ -23,7 +23,7 @@ from computes.forms import ( ) from computes.models import Compute -from . import capacity, utils +from . import utils # The host CPU graph sends back the CPU sample of its previous answer; usage # is taken since then when it is this recent, otherwise over one second. @@ -33,11 +33,10 @@ CPU_SAMPLE_SIGNER = signing.TimestampSigner(salt="computes.cpu-sample") @superuser_only def computes(request): - """The computes, with what each has given its VMs (computes/capacity.py).""" + """The computes, with what each has given its VMs.""" computes = list(Compute.objects.order_by("name")) - found = capacity.read_all(computes) for compute in computes: - compute.capacity = given = found.get(compute.id) + given = compute.capacity if given: given["ratio"] = given["vcpus"]["all"] / given["cpus"] if given["cpus"] else 0 # the memory bar: running and shut-off VMs' memory, and a mark at the @@ -48,9 +47,9 @@ def computes(request): "stopped": (assigned["all"] - assigned["running"]) * 100 / scale, "host": memory * 100 / scale, "over": assigned["all"] > memory} given["memory_percent"] = round(assigned["all"] * 100 / memory) if memory else 0 - pools = [{"name": name, "size": size, "free": free, - "used_percent": (size - free) * 100 / size if size else 0} - for name, size, free in given["pools"]] + pools = sorted(given["pools"], key=lambda pool: pool["name"]) + for pool in pools: + pool["used_percent"] = (pool["size"] - pool["free"]) * 100 / pool["size"] if pool["size"] else 0 # the default pool shows; the others (a host may have dozens) open on a click given["pool"] = next((pool for pool in pools if pool["name"] == "default"), pools[0] if pools else None) given["more_pools"] = [pool for pool in pools if pool is not given["pool"]] diff --git a/vrtManager/hostdetails.py b/vrtManager/hostdetails.py index 7e63a5dd..3de53687 100644 --- a/vrtManager/hostdetails.py +++ b/vrtManager/hostdetails.py @@ -1,6 +1,6 @@ import time -from libvirt import VIR_NODE_CPU_STATS_ALL_CPUS +from libvirt import VIR_DOMAIN_SHUTOFF, VIR_NODE_CPU_STATS_ALL_CPUS from vrtManager.connection import wvmConnect from vrtManager.util import get_xml_path @@ -83,3 +83,18 @@ class wvmHostDetails(wvmConnect): ) # cpu version info.append(self.wvm.getURI()) # uri return info + + def get_capacity(self): + """ + What the host has given its VMs: vCPUs and memory (bytes) of all VMs + and of the running ones, against its CPUs and memory. A shut-off VM + takes its share again when it starts. + """ + vcpus = {"all": 0, "running": 0} + assigned = {"all": 0, "running": 0} + for vm in self.get_host_instances(raw_mem_size=True).values(): + for share in ("all",) if vm["status"] == VIR_DOMAIN_SHUTOFF else ("all", "running"): + vcpus[share] += int(vm["vcpu"]) + assigned[share] += vm["memory"] + return {"cpus": self.host_info[2], "memory": self.host_info[1] * 1048576, + "vcpus": vcpus, "assigned": assigned} diff --git a/vrtManager/storage.py b/vrtManager/storage.py index 4b2a1ce0..b932b5cd 100644 --- a/vrtManager/storage.py +++ b/vrtManager/storage.py @@ -14,7 +14,7 @@ class wvmStorages(wvmConnect): stg_status = stg.isActive() stg_type = util.get_xml_path(stg.XMLDesc(0), "/pool/@type") stg_vol = len(stg.listVolumes()) if stg_status else None - stg_size = stg.info()[1] + _state, stg_size, _allocation, stg_free = stg.info() storages.append( { "name": pool, @@ -22,6 +22,7 @@ class wvmStorages(wvmConnect): "type": stg_type, "volumes": stg_vol, "size": stg_size, + "free": stg_free, } ) return storages diff --git a/vrtManager/tests/test_host_reads.py b/vrtManager/tests/test_host_reads.py index 66df2ae6..9ada59de 100644 --- a/vrtManager/tests/test_host_reads.py +++ b/vrtManager/tests/test_host_reads.py @@ -13,9 +13,19 @@ from django.conf import settings if not settings.configured: settings.configure(MAC_OUI="52:54:10") -from libvirt import VIR_CONNECT_LIST_NODE_DEVICES_CAP_NET, VIR_ERR_NO_SUPPORT, VIR_ERR_OPERATION_DENIED, VIR_ERR_XML_ERROR, libvirtError +from libvirt import ( + VIR_CONNECT_LIST_NODE_DEVICES_CAP_NET, + VIR_DOMAIN_PAUSED, + VIR_DOMAIN_RUNNING, + VIR_DOMAIN_SHUTOFF, + VIR_ERR_NO_SUPPORT, + VIR_ERR_OPERATION_DENIED, + VIR_ERR_XML_ERROR, + libvirtError, +) from vrtManager.connection import wvmConnect from vrtManager.hostdetails import wvmHostDetails +from vrtManager.storage import wvmStorages CAPABILITIES = """x86_64 hvm/usr/bin/qemu-kvm @@ -132,3 +142,32 @@ class ChangeInterfacesTestCase(unittest.TestCase): self.assertIs(conn.can_change_interfaces(), changeable) # an invalid definition: nothing can be defined by asking conn.wvm.interfaceDefineXML.assert_called_once_with("", 0) + + +class CapacityTestCase(unittest.TestCase): + def test_a_host_gives_its_vms_cpus_and_memory(self): + conn = connection(wvmHostDetails) + gib = 1024 ** 3 + # get_host_instances gives the vCPUs as text, the memory in bytes + conn.get_host_instances = MagicMock(return_value={ + "web": {"status": VIR_DOMAIN_RUNNING, "vcpu": "4", "memory": 4 * gib}, + "db": {"status": VIR_DOMAIN_PAUSED, "vcpu": "2", "memory": 2 * gib}, + "old": {"status": VIR_DOMAIN_SHUTOFF, "vcpu": "8", "memory": 8 * gib}, + }) + self.assertEqual(conn.get_capacity(), { + "cpus": 4, "memory": 2 * gib, + "vcpus": {"all": 14, "running": 6}, + "assigned": {"all": 14 * gib, "running": 6 * gib}, + }) + conn.get_host_instances.assert_called_once_with(raw_mem_size=True) + + +class StoragesInfoTestCase(unittest.TestCase): + def test_each_pool_tells_its_free_space(self): + conn = connection(wvmStorages) + pool = MagicMock(**{"isActive.return_value": 1, "XMLDesc.return_value": "", + "listVolumes.return_value": ["a.qcow2"], "info.return_value": [2, 500, 380, 120]}) + conn.get_storages = lambda: ["default"] + conn.get_storage = lambda name: pool + (info,) = conn.get_storages_info() + self.assertEqual((info["name"], info["size"], info["free"]), ("default", 500, 120))