feat!: declare the guest agent per guest OS, and share what provisioning needs #20

Merged
christianmanivong merged 1 commits from feat/guest-os-provisioning into main 2026-10-08 05:57:09 +00:00
6 changed files with 191 additions and 23 deletions
+10
View File
@@ -312,6 +312,16 @@ class ProxmoxDriver(HypervisorDriver):
...
```
A hypervisor that provisions VMs from cloud images declares, per guest OS,
the agent cloud-init installs so the hypervisor can read the new VM's IP:
`GUEST_AGENTS = {guest_os: (packages, runcmd)}`. Its keys are the guests the
driver can provision, and `create_vm_from_cloud_init(guest_os=...)` refuses
any other. A QEMU-based driver takes `provisioning.QEMU_GUEST_AGENTS`
(Linux, FreeBSD, OpenBSD). `provisioning.network_config()` writes the
network-config v2 every cloud-init flavour reads (FreeBSD's nuageinit reads
no other), and `provisioning.split_compression()` says how a packed image is
unpacked; both are the same for every hypervisor.
### OS / Linux
```python
+14 -5
View File
@@ -14,6 +14,7 @@ from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.provisioning import QEMU_GUEST_AGENTS, GuestAgents
from napalm_device_types.models import (
NICConfigDict,
NetworkTargetDict,
@@ -44,11 +45,11 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
ROLE: str = "hypervisor"
TYPE_LABEL: str = "Hypervisor"
#: What cloud-init installs and starts on a VM provisioned through this
#: driver, so the hypervisor can read the guest's IP address back.
GUEST_AGENT_PACKAGES: tuple[str, ...] = ("qemu-guest-agent",)
GUEST_AGENT_RUNCMD: tuple[str, ...] = ("systemctl enable --now qemu-guest-agent",)
#: What cloud-init installs and runs on a VM provisioned through this
#: driver, per guest OS, so the hypervisor can read the guest's IP
#: address back. The guest operating systems listed here are the ones
#: the driver can provision; netOrk reads this off the class.
GUEST_AGENTS: GuestAgents = {"linux": QEMU_GUEST_AGENTS["linux"]}
# ------------------------------------------------------------------
@@ -464,6 +465,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
disk_resize_gb: int | None = None,
storage: str | None = None,
cpu_type: str | None = None,
guest_os: str = "linux",
download_timeout: int = 300,
timeout: int = 180,
) -> VMProvisionResultDict:
@@ -512,6 +514,13 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
value other than None with ValueError; one that has it raises
ValueError for a name it does not list or that is not
``available`` on this node, before creating anything.
guest_os (string) - the operating system in the image, a key of
``GUEST_AGENTS`` (``"linux"``, ``"freebsd"``, ``"openbsd"``). The
driver sets up the VM's hardware for it (OS type, how the guest
agent is attached) and, where the guest needs it, writes the
network-config itself (``provisioning.network_config``). A
guest_os the driver does not list raises ValueError before
anything is created. Default ``"linux"``.
download_timeout (int) - maximum seconds to wait for the image download
(skipped entirely if already cached on the hypervisor). Default 300.
timeout (int) - maximum seconds to wait for the remaining provisioning
+64
View File
@@ -0,0 +1,64 @@
"""
What provisioning a VM from a cloud image needs to know about the guest.
The same for every hypervisor: which agent a guest runs so a QEMU-based
hypervisor can read its IP, the network-config cloud-init and FreeBSD's
nuageinit both read, and how a packed cloud image is unpacked. How a
hypervisor attaches, boots and talks to the VM stays in its driver.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
#: ``{guest_os: (packages, runcmd)}``: what cloud-init installs and runs so
#: the hypervisor can read the guest's IP address back.
GuestAgents = Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]]
#: The QEMU guest agent per guest OS, for hypervisors built on QEMU. Package
#: names and service commands were verified on FreeBSD 15.1 and OpenBSD 7.9
#: cloud images (NetOrk/netork#793).
QEMU_GUEST_AGENTS: GuestAgents = {
"linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
"freebsd": (
("qemu-guest-agent",),
("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
),
"openbsd": (("qemu-ga",), ("rcctl enable qemu_ga", "rcctl start qemu_ga")),
}
#: Suffix of a packed image -> the command that writes it unpacked to stdout.
_DECOMPRESSORS = {
"xz": "xz -dc",
"gz": "gzip -dc",
"bz2": "bzip2 -dc",
"zst": "zstd -dc",
}
def network_config(nics: list[tuple[str, bool]]) -> dict[str, Any] | None:
"""Network-config v2: DHCP on every ``(mac, dhcp)`` NIC that asks for it.
Version 2 because FreeBSD's nuageinit reads no other: given the v1 that
Proxmox generates, it fails and skips the rest of its first stage,
runcmd included. ``None`` when no NIC wants DHCP, so the guest is left
alone rather than told to configure nothing.
"""
ethernets = {
f"nic{i}": {"match": {"macaddress": mac.lower()}, "dhcp4": True}
for i, (mac, dhcp) in enumerate(nics)
if dhcp
}
return {"version": 2, "ethernets": ethernets} if ethernets else None
def split_compression(filename: str) -> tuple[str, str | None]:
"""``(unpacked filename, command that unpacks to stdout)`` for *filename*.
The command is None for an image that is not packed, which is then
imported as it is.
"""
stem, dot, suffix = filename.rpartition(".")
command = _DECOMPRESSORS.get(suffix.lower()) if dot and stem else None
return (stem, command) if command else (filename, None)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-device-types"
version = "2.7.0"
version = "3.0.0"
description = "Abstract device-type base classes for NAPALM drivers"
readme = "README.md"
requires-python = ">=3.10"
+102
View File
@@ -0,0 +1,102 @@
"""What provisioning a VM from a cloud image needs to know about the guest,
the same for every hypervisor (NetOrk/netork#794)."""
from __future__ import annotations
import pytest
from napalm_device_types import HypervisorDriver
from napalm_device_types.provisioning import (
QEMU_GUEST_AGENTS,
network_config,
split_compression,
)
class TestGuestAgentDeclaration:
"""netOrk's cloud-init installs the agent through which the hypervisor
reads the new VM's IP. Which agent depends on the hypervisor and on the
guest; the guests a driver lists are the ones it can provision."""
def test_the_base_provisions_linux_with_qemu_guest_agent(self):
assert HypervisorDriver.GUEST_AGENTS == {
"linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
}
def test_the_declaration_is_data_netork_reads_off_the_class(self):
assert not callable(vars(HypervisorDriver)["GUEST_AGENTS"])
def test_the_per_hypervisor_attributes_are_gone(self):
assert not hasattr(HypervisorDriver, "GUEST_AGENT_PACKAGES")
assert not hasattr(HypervisorDriver, "GUEST_AGENT_RUNCMD")
class TestQemuGuestAgents:
"""Verified on FreeBSD 15.1 and OpenBSD 7.9 cloud images (#793)."""
def test_linux_is_the_base_default(self):
assert QEMU_GUEST_AGENTS["linux"] == HypervisorDriver.GUEST_AGENTS["linux"]
def test_freebsd(self):
assert QEMU_GUEST_AGENTS["freebsd"] == (
("qemu-guest-agent",),
("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
)
def test_openbsd(self):
assert QEMU_GUEST_AGENTS["openbsd"] == (
("qemu-ga",),
("rcctl enable qemu_ga", "rcctl start qemu_ga"),
)
@pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
def test_no_bsd_guest_is_told_to_use_systemd(self, guest_os):
_, runcmd = QEMU_GUEST_AGENTS[guest_os]
assert not any("systemctl" in command for command in runcmd)
class TestNetworkConfig:
"""cloud-init's network-config v2. FreeBSD's nuageinit reads only this
version; the v1 Proxmox generates makes it skip runcmd (#793)."""
def test_dhcp_nics_matched_by_mac(self):
cfg = network_config([("00:50:56:aa:bb:cc", True), ("00:50:56:aa:bb:dd", False)])
assert cfg == {
"version": 2,
"ethernets": {
"nic0": {"match": {"macaddress": "00:50:56:aa:bb:cc"}, "dhcp4": True},
},
}
def test_no_dhcp_nic_means_no_network_config(self):
assert network_config([("00:50:56:aa:bb:cc", False)]) is None
def test_macs_are_written_in_lower_case(self):
"""Proxmox reports them upper-case; nuageinit compares them as given."""
cfg = network_config([("BC:24:11:AA:BB:02", True)])
assert cfg["ethernets"]["nic0"]["match"]["macaddress"] == "bc:24:11:aa:bb:02"
class TestSplitCompression:
"""Official FreeBSD images come packed; hypervisors import unpacked ones."""
def test_xz(self):
assert split_compression("FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz") == (
"FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2",
"xz -dc",
)
@pytest.mark.parametrize(
("name", "command"),
[("a.raw.gz", "gzip -dc"), ("a.raw.bz2", "bzip2 -dc"), ("a.qcow2.zst", "zstd -dc")],
)
def test_other_packers(self, name, command):
assert split_compression(name) == (name.rsplit(".", 1)[0], command)
@pytest.mark.parametrize(
"name", ["debian-13-genericcloud-amd64.qcow2", "ubuntu-24.04-server-cloudimg-amd64.img"]
)
def test_an_unpacked_image_is_left_as_it_is(self, name):
assert split_compression(name) == (name, None)
def test_the_suffix_is_matched_case_insensitively(self):
assert split_compression("IMAGE.QCOW2.XZ") == ("IMAGE.QCOW2", "xz -dc")
-17
View File
@@ -51,20 +51,3 @@ class TestVMConfigCarriesWhatAHardwareViewShows:
from napalm_device_types.models import VMPassthroughDict
assert get_type_hints(VMPassthroughDict) == {"slot": str, "kind": str, "config": str}
class TestGuestAgentDeclaration:
"""netOrk's cloud-init installed qemu-guest-agent on every new VM. A VMware
guest reports its IP through open-vm-tools instead; the hypervisor says
which, and netOrk stops hard-coding one of them."""
def test_default_is_qemu_guest_agent(self):
from napalm_device_types import HypervisorDriver
assert HypervisorDriver.GUEST_AGENT_PACKAGES == ("qemu-guest-agent",)
assert HypervisorDriver.GUEST_AGENT_RUNCMD == ("systemctl enable --now qemu-guest-agent",)
def test_attributes_are_not_methods(self):
from napalm_device_types import HypervisorDriver
assert not callable(vars(HypervisorDriver)["GUEST_AGENT_PACKAGES"])