1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00

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.
This commit is contained in:
catborise 2026-10-25 12:15:23 +03:00
parent 1436ed13af
commit d9f9ebab58
8 changed files with 114 additions and 185 deletions

View file

@ -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())

View file

@ -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

View file

@ -81,7 +81,7 @@
{% endif %}
</td>
{% else %}
<td colspan="3" class="small text-body-secondary">{% 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 %}</td>
<td colspan="3" class="small text-body-secondary">{% if compute.status is True %}{% trans "The capacity could not be read." %}{% else %}{% trans "No capacity while the host cannot be reached." %}{% endif %}</td>
{% endif %}
<td class="text-end">
<div class="btn-group">

View file

@ -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())

View file

@ -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"]]

View file

@ -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}

View file

@ -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

View file

@ -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 = """<capabilities><host><cpu><arch>x86_64</arch></cpu></host>
<guest><os_type>hvm</os_type><arch name='x86_64'><emulator>/usr/bin/qemu-kvm</emulator>
@ -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("<interface/>", 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": "<pool type='dir'/>",
"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))