Author SHA1 Message Date
christianmanivong b5c40019af feat: a new VM's CPU model can be chosen, from a list the hypervisor offers
create_vm_from_cloud_init takes cpu_type. Proxmox gives a VM created without
one the kvm64 model, which has no AVX, so MongoDB 5.0 and later do not start
there, and netOrk's graylog role failed on every VM it provisioned
(netork#494). Which model is right depends on the cluster: host cannot
live-migrate between different CPUs, x86-64-v3 does not start on a CPU older
than Haswell. So the caller chooses.

get_vm_cpu_types() is the new, optional listing behind that choice. Each
VMCpuTypeDict names the model, says what it is for, lists the /proc/cpuinfo
flags the guest gets (so a caller can ask "does this give AVX?" without
knowing model names), whether the node at hand can run it, and which one is
the default. A hypervisor whose VMs have no per-VM CPU model (VMware sets CPU
compatibility per cluster) does not implement it and must reject any
cpu_type other than None.

Both declarations sit under TYPE_CHECKING like the rest of the contract, so
hasattr stays a truthful capability probe; the tests read the signature from
the source.
2026-10-04 12:02:38 +02:00
4 changed files with 111 additions and 0 deletions
+6
View File
@@ -247,6 +247,12 @@ class ProxmoxDriver(HypervisorDriver):
def create_vm_snapshot(self, name, snapshot, description="", include_memory=False):
...
def get_vm_cpu_types(self):
# optional — return List[VMCpuTypeDict]: the CPU models a new VM may get
# on this node, each with its cpuinfo flags and whether the node can run
# it; the name goes to create_vm_from_cloud_init(cpu_type=...)
...
```
### OS / Linux
+29
View File
@@ -21,6 +21,7 @@ from napalm_device_types.models import (
StorageTargetDict,
StorageVolumeDict,
VMConfigDict,
VMCpuTypeDict,
VMDict,
VMProvisionResultDict,
VMStatusDict,
@@ -462,6 +463,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
ssh_public_keys: List[str] | None = None,
disk_resize_gb: int | None = None,
storage: str | None = None,
cpu_type: str | None = None,
download_timeout: int = 300,
timeout: int = 180,
) -> VMProvisionResultDict:
@@ -503,6 +505,13 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
disk on (a name returned by ``get_image_storages()``). If None,
the driver auto-detects the first enabled, node-available storage
whose content includes "images".
cpu_type (string | None) - virtual CPU model for the VM (a name
returned by ``get_vm_cpu_types()``). If None, the driver uses the
entry that listing marks ``default``. A driver without
``get_vm_cpu_types()`` offers no choice and must reject any
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.
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
@@ -618,3 +627,23 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
as ``create_vm_from_cloud_init``'s ``storage`` argument.
"""
...
def get_vm_cpu_types(self) -> List[VMCpuTypeDict]:
"""
List the virtual CPU models a new VM may be given on the node it will
be created on.
Optional: a hypervisor whose VMs have no per-VM CPU model (VMware
sets CPU compatibility per cluster) does not implement it, and a
caller then offers no choice.
Every model the driver knows is listed, also those this node's CPU
cannot run -- marked ``available: False`` -- so a picker can show why
an option is missing. Exactly one entry is ``default``: the model
``create_vm_from_cloud_init`` uses when ``cpu_type`` is None.
Returns:
List[VMCpuTypeDict] - each entry's ``name`` is directly usable as
``create_vm_from_cloud_init``'s ``cpu_type`` argument.
"""
...
+18
View File
@@ -912,6 +912,24 @@ class StorageTargetDict(TypedDict):
available_gb: float # Free capacity in gigabytes
class VMCpuTypeDict(TypedDict):
"""A virtual CPU model a new VM may be given (``create_vm_from_cloud_init``'s
``cpu_type`` argument), judged against the specific node the VM will be
created on.
``features`` uses the flag names of Linux's ``/proc/cpuinfo`` (``avx``,
``avx2``, ``aes`` ...), so a caller can ask "does this model give the guest
AVX?" without knowing the hypervisor's model names. A model that passes the
host CPU through lists that CPU's own flags.
"""
name: str # Model name, usable directly as create_vm_from_cloud_init(cpu_type=...)
description: str # One line on what the model is for, for a picker
features: List[str] # cpuinfo flags the guest is guaranteed to see
available: bool # False when this node's CPU cannot run the model
default: bool # The model create_vm_from_cloud_init uses when cpu_type is None
# ---------------------------------------------------------------------------
# Ping sweep (shared across device types)
# ---------------------------------------------------------------------------
+58
View File
@@ -0,0 +1,58 @@
"""A new VM's virtual CPU model can be chosen, from a list the hypervisor offers.
Proxmox gives a VM created without a ``cpu`` argument the ``kvm64`` model,
which has no AVX -- and MongoDB 5.0 and later will not start without it. Which
model is right depends on the cluster (``host`` cannot live-migrate between
different CPUs, ``x86-64-v3`` does not start on a CPU older than Haswell), so
the caller chooses, from entries that say what each model provides and whether
the node at hand can run it.
The declarations sit under ``TYPE_CHECKING`` (see test_role_contracts), so the
signature is read from the source rather than from the class.
"""
from __future__ import annotations
import ast
import inspect
from typing import List, get_type_hints
import napalm_device_types.hypervisor as hypervisor_module
from napalm_device_types import HypervisorDriver
from napalm_device_types.models import VMCpuTypeDict
def _declared(name: str) -> ast.FunctionDef:
tree = ast.parse(inspect.getsource(hypervisor_module))
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name == name:
return node
raise AssertionError(f"HypervisorDriver does not declare {name}()")
class TestCreateVmTakesACpuType:
def test_cpu_type_is_an_optional_keyword(self):
fn = _declared("create_vm_from_cloud_init")
kwonly = {arg.arg: default for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults)}
assert "cpu_type" in kwonly
default = kwonly["cpu_type"]
assert isinstance(default, ast.Constant) and default.value is None
class TestCpuTypeListing:
def test_is_declared(self):
assert _declared("get_vm_cpu_types").returns is not None
def test_absent_until_a_driver_implements_it(self):
"""netOrk probes capabilities with hasattr; a hypervisor without a
choice of CPU model must not seem to offer one."""
assert not hasattr(HypervisorDriver, "get_vm_cpu_types")
def test_entry_shape(self):
assert get_type_hints(VMCpuTypeDict) == {
"name": str,
"description": str,
"features": List[str],
"available": bool,
"default": bool,
}