From 9c86e75c369cb471349ad712f95950c57567f115 Mon Sep 17 00:00:00 2001 From: catborise Date: Sun, 25 Oct 2026 16:04:07 +0300 Subject: [PATCH] fix(instances): the IP search asks a guest agent with a timeout 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. --- instances/addresses.py | 21 +++++++----- instances/tests/test_addresses.py | 40 ++++++++++++++++++++++ vrtManager/instance.py | 57 ++++++++++++++++--------------- 3 files changed, 81 insertions(+), 37 deletions(-) diff --git a/instances/addresses.py b/instances/addresses.py index 4c669a06..c6ea0edf 100644 --- a/instances/addresses.py +++ b/instances/addresses.py @@ -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): diff --git a/instances/tests/test_addresses.py b/instances/tests/test_addresses.py index 53cad012..6f71995b 100644 --- a/instances/tests/test_addresses.py +++ b/instances/tests/test_addresses.py @@ -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"] diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 3cd3862b..d197c257 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -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): """