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): """