j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
Code Issues Projects Releases Wiki Activity

fix(instances): the IP search asks a guest agent with a timeout

Browse source 
The IP search and the VM report read each running VM's addresses and
OS from its guest agent with interfaceAddresses() and guestInfo(),
which wait for a hung agent indefinitely. The search answered by its
deadline, but the thread stayed blocked, counted against the search
thread limit and held up the worker's exit.

They now use the 10 s agent command the VM page uses, shared as
vrtManager.instance.agent_command() and agent_interfaces(). With the
agent frozen, a search answers in 5 s from the leases and the ARP
table, counts the VM unread, and its thread ends about 5 s later.
This commit is contained in:
catborise 2026-10-25 16:04:07 +03:00
parent 400b19cf42
commit 9c86e75c36
3 changed files with 81 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

21
instances/addresses.py
View file

@ -16,7 +16,6 @@ from urllib.parse import urlparse
from libvirt import (
VIR_CONNECT_LIST_DOMAINS_ACTIVE,
VIR_DOMAIN_GUEST_INFO_OS,
VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP,
VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_LEASE,
@ -24,6 +23,7 @@ from libvirt import (
libvirtError,
)
from lxml import etree
from vrtManager.instance import agent_command, agent_interfaces
# in the order a match lists its sources
SOURCES = {
@ -33,10 +33,10 @@ SOURCES = {
}
# A search asks the hosts in parallel, at most MAX_WORKERS threads at once.
# A guest agent may never answer (libvirt waits without limit by default):
# its thread stays blocked after the search answers (and holds up the
# worker's exit until it does), so a search starts only while its threads
# fit in MAX_RUNNING search threads in this process.
# A read may outlast the search (a guest agent has AGENT_TIMEOUT seconds, a
# host its connection's timeouts): its thread goes on after the search
# answers, so a search starts only while its threads fit in MAX_RUNNING
# search threads in this process.
MAX_WORKERS = 8
MAX_RUNNING = 16
_running = 0
@ -85,7 +85,10 @@ def read(dom, source):
"""(MAC, address) pairs the source reports for the domain's interfaces
that have a MAC (an agent's tunnels have none); None if it fails."""
try:
answer = dom.interfaceAddresses(source)
if source == VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT:
answer = agent_interfaces(dom)
else:
answer = dom.interfaceAddresses(source)
except libvirtError:
return None
return [(iface["hwaddr"].lower(), ip_address(addr["addr"]))
@ -140,12 +143,12 @@ def _matching(dom, vm, source, macs, network):
def _read_os(dom):
"""The guest agent's OS as (name, family); None if it gives none."""
try:
info = dom.guestInfo(VIR_DOMAIN_GUEST_INFO_OS)
info = agent_command(dom, "guest-get-osinfo")
except libvirtError:
return None
if not info.get("os.id"):
if not info.get("id"):
return None
return info.get("os.pretty-name") or info.get("os.name") or info["os.id"], os_family(info["os.id"])
return info.get("pretty-name") or info.get("name") or info["id"], os_family(info["id"])
def _read_host(target, network, everything, mac):

40
instances/tests/test_addresses.py
View file

@ -1,13 +1,16 @@
"""Finding the VMs that use an address (instances/addresses.py), with libvirt
answers faked per source."""
import json
import threading
import time
import unittest
import unittest.mock
from ipaddress import ip_address, ip_network
from types import SimpleNamespace
from instances import addresses
from vrtManager import instance as vrt_instance
from libvirt import (
VIR_CONNECT_LIST_DOMAINS_ACTIVE,
VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT,
@ -19,6 +22,21 @@ from libvirt import (
AGENT, ARP, LEASE = (VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT, VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP,
VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_LEASE)
# the guest agent is asked through qemuAgentCommand, which needs a real domain:
# the fake domains answer it themselves
_agent_patch = unittest.mock.patch.object(
vrt_instance, "qemuAgentCommand",
lambda dom, request, timeout, flags: json.dumps({"return": dom.agent(json.loads(request)["execute"], timeout)}))
def setUpModule():
_agent_patch.start()
def tearDownModule():
_agent_patch.stop()
MAC1 = "52:54:00:00:00:01"
MAC2 = "52:54:00:00:00:02"
@ -42,6 +60,7 @@ class FakeDomain:
def __init__(self, answers, uuid="uuid-a", name="vm-a", xml=None, state=1, os=None):
self.answers, self.uuid, self.vname, self.state_, self.os = answers, uuid, name, state, os
self.agent_timeouts = []
self.xml = xml or domain_xml(MAC1, agent="connected")
def interfaceAddresses(self, source):
@ -79,6 +98,19 @@ class FakeDomain:
def XMLDesc(self, flags=0):
return self.xml
def agent(self, command, timeout):
"""The guest agent's answer to a command, from the same answers in
the agent's own form; the timeouts asked with are kept."""
self.agent_timeouts.append(timeout)
if command == "guest-get-osinfo":
return {key.removeprefix("os."): value for key, value in self.guestInfo().items()}
interfaces = []
for name, nic in self.interfaceAddresses(AGENT).items():
ips = [{"ip-address-type": "ipv4" if a["type"] == 0 else "ipv6", "ip-address": a["addr"],
"prefix": a["prefix"]} for a in nic.get("addrs") or []]
interfaces.append({"name": name, "hardware-address": nic.get("hwaddr"), "ip-addresses": ips})
return interfaces
def compute(name, *domains, pk=1, down=False):
def listing(flags=0):
@ -288,6 +320,14 @@ class InventoryTests(unittest.TestCase):
self.assertEqual((win["name"], win["state"], win["addresses"]), ("win", 5, []))
self.assertEqual((win["os"], win["os_family"]), ("win 2k22", "Windows"))
def test_the_agent_is_asked_with_a_timeout(self):
# interfaceAddresses() and guestInfo() would wait for a hung agent
# indefinitely, and the search's thread with it
dom = FakeDomain({AGENT: {"eth0": iface(MAC1, (0, "192.0.2.10", 24))}},
os={"os.id": "almalinux", "os.pretty-name": "AlmaLinux 10.0"})
addresses.inventory([addresses.Target(compute("c1", dom), None)])
self.assertEqual(dom.agent_timeouts, [10, 10])
def test_an_os_without_an_agent_or_metadata_is_unknown(self):
dom = vm("a", MAC1, "192.0.2.10", agent=None)
(row,) = addresses.inventory([addresses.Target(compute("c1", dom), None)])["vms"]

57
vrtManager/instance.py
View file

@ -362,6 +362,29 @@ class wvmInstances(wvmConnect):
) from error
def agent_command(dom, command, arguments=None):
"""The answer of one guest agent command that waits AGENT_TIMEOUT seconds
at most: libvirt's own calls for it wait for a hung agent indefinitely,
and the domain's agent timeout is shared by every agent call."""
request = {"execute": command}
if arguments is not None:
request["arguments"] = arguments
return json.loads(qemuAgentCommand(dom, json.dumps(request), AGENT_TIMEOUT, 0))["return"]
def agent_interfaces(dom):
"""The guest's interfaces from its agent, in interfaceAddresses()'s form,
which waits for a hung agent indefinitely."""
interfaces = {}
for iface in agent_command(dom, "guest-network-get-interfaces"):
addrs = []
for address in iface.get("ip-addresses", []):
kind = 0 if address["ip-address-type"] == "ipv4" else 1
addrs.append({"type": kind, "addr": address["ip-address"], "prefix": address["prefix"]})
interfaces[iface["name"]] = {"hwaddr": iface.get("hardware-address"), "addrs": addrs}
return interfaces
class wvmInstance(wvmConnect):
# set by cached_reads(): libvirt answers reused within the block
_read_cache = None
@ -374,9 +397,9 @@ class wvmInstance(wvmConnect):
self.instance = self.wvm.lookupByUUIDString(uuid) if uuid else self.get_instance(vname)
def osinfo(self):
info = self._agent_command("guest-get-osinfo")
info.update(self._agent_command("guest-get-timezone"))
info.update(self._agent_command("guest-get-host-name"))
info = agent_command(self.instance, "guest-get-osinfo")
info.update(agent_command(self.instance, "guest-get-timezone"))
info.update(agent_command(self.instance, "guest-get-host-name"))
return info
def start(self, flags=0):
@ -575,18 +598,6 @@ class wvmInstance(wvmConnect):
return self.instance.interfaceAddresses(source)
return {}
def _agent_interfaces(self):
"""The guest's interfaces from its agent, in interfaceAddresses()'s
form, which waits for a hung agent indefinitely."""
interfaces = {}
for iface in self._agent_command("guest-network-get-interfaces"):
addrs = []
for address in iface.get("ip-addresses", []):
kind = 0 if address["ip-address-type"] == "ipv4" else 1
addrs.append({"type": kind, "addr": address["ip-address"], "prefix": address["prefix"]})
interfaces[iface["name"]] = {"hwaddr": iface.get("hardware-address"), "addrs": addrs}
return interfaces
def refresh_interface_addresses(self):
self._ip_cache = {"qemuga": {}, "arp": {}}
@ -595,7 +606,7 @@ class wvmInstance(wvmConnect):
if self.is_agent_ready():
with contextlib.suppress(libvirtError):
self._ip_cache["qemuga"] = self._agent_interfaces()
self._ip_cache["qemuga"] = agent_interfaces(self.instance)
self._ip_cache["arp"] = self._get_interface_addresses(VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP)
@ -2509,26 +2520,16 @@ class wvmInstance(wvmConnect):
return util.get_xml_path(self._XMLDesc(0), func=_get_agent)
def _agent_command(self, command, arguments=None):
"""The answer of one guest agent command that waits AGENT_TIMEOUT
seconds at most: libvirt's own calls for it wait for a hung agent
indefinitely, and the domain's agent timeout is shared by every agent
call."""
request = {"execute": command}
if arguments is not None:
request["arguments"] = arguments
return json.loads(qemuAgentCommand(self.instance, json.dumps(request), AGENT_TIMEOUT, 0))["return"]
def set_user_password(self, user, crypted):
"""Set a user's password in the running guest through its agent;
crypted is a crypt(3) hash."""
password = base64.b64encode(crypted.encode()).decode()
self._agent_command("guest-set-user-password", {"username": user, "password": password, "crypted": True})
agent_command(self.instance, "guest-set-user-password", {"username": user, "password": password, "crypted": True})
def add_authorized_key(self, user, key):
"""Add an SSH public key to the user's authorized keys in the running
guest through its agent."""
self._agent_command("guest-ssh-add-authorized-keys", {"username": user, "keys": [key], "reset": False})
agent_command(self.instance, "guest-ssh-add-authorized-keys", {"username": user, "keys": [key], "reset": False})
def is_agent_ready(self):
"""