feat(hypervisor): add VM provisioning interface stubs

Add three new methods to HypervisorDriver:
- create_vm_from_cloud_init(): provision VM from template with dual-NIC config
- destroy_vm(): stop and remove VM with optional disk cleanup
- get_vm_status(): poll runtime status, optionally wait for IP via guest-agent

New TypedDicts VMProvisionResultDict and VMStatusDict in models.py
document the provisioning API contract.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-07-06 21:45:52 +02:00
co-authored by Claude Haiku 4.5
parent f51b1a4b6e
commit 7f0dd789b0
2 changed files with 138 additions and 0 deletions
+121
View File
@@ -20,6 +20,8 @@ from napalm_device_types.models import (
StorageVolumeDict,
VMConfigDict,
VMDict,
VMProvisionResultDict,
VMStatusDict,
VirtualNetworkDict,
)
@@ -535,3 +537,122 @@ class HypervisorDriver(DeviceTypeDriver):
)
"""
raise NotImplementedError
# ------------------------------------------------------------------
# Virtual machine provisioning
# ------------------------------------------------------------------
def create_vm_from_cloud_init(
self,
name: str,
*,
template: str,
cpu: int,
memory: int,
mgmt_bridge: str,
mgmt_vlan_tag: int | None,
capture_bridge: str,
capture_vlan_tags: List[int],
cloud_init_config: Dict[str, Any],
ssh_public_keys: List[str] | None = None,
timeout: int = 120,
) -> VMProvisionResultDict:
"""
Create a new virtual machine from a Cloud-Init template.
Clones a pre-existing VM template, configures dual NICs (management +
capture), and injects Cloud-Init configuration via a storage snippet or
similar mechanism. The resulting VM is left in a running state.
Args:
name (string) - new VM display name
template (string) - hypervisor-internal ID/name of the template VM to clone
cpu (int) - number of virtual CPUs to assign
memory (int) - RAM to assign in megabytes
mgmt_bridge (string) - network bridge for management NIC (net0)
mgmt_vlan_tag (int | None) - VLAN tag for net0 (None → untagged)
capture_bridge (string) - network bridge for packet capture NIC (net1).
Must support ``vlan-aware`` and trunk mode.
capture_vlan_tags (list[int]) - VLAN tags to place on net1 (trunk).
Packets matching any of these VLANs are visible to the guest.
cloud_init_config (dict) - user-data dict (will be rendered to YAML).
Should include: ``bootstrap_token``, ``central_url``, ``satellite_id``,
``runcmd`` for custom startup sequence.
ssh_public_keys (list[str] | None) - SSH public keys to inject into guest.
If None or empty, no SSH key injection is performed.
timeout (int) - maximum seconds to wait for provisioning completion
(clone, config, start). Default 120.
Returns:
VMProvisionResultDict - ``{"vmid": str, "name": str, "node": str}``
vmid is the hypervisor-internal VM ID as a string (numeric for Proxmox).
Raises:
RuntimeError - if provisioning fails (storage unavailable, invalid
config, timeout, etc.)
"""
raise NotImplementedError
def destroy_vm(
self,
vmid: str,
*,
remove_disk: bool = True,
timeout: int = 60,
) -> None:
"""
Destroy a virtual machine and optionally remove its storage.
Stops the VM (if running) and purges it from the hypervisor. Optionally
also removes associated disks and ephemeral storage (e.g. Cloud-Init snippets).
Args:
vmid (string) - hypervisor-internal VM ID (as returned from ``get_vms()``
or ``create_vm_from_cloud_init()``).
remove_disk (bool) - if True (default), delete all disks and snapshot
data associated with the VM. If False, only the VM configuration
is removed; disks are left behind (rare use case).
timeout (int) - maximum seconds to wait for the destroy operation
(stop + delete). Default 60.
Raises:
RuntimeError - if the VM does not exist or destroy fails.
"""
raise NotImplementedError
def get_vm_status(
self,
vmid: str,
*,
wait_for_ip: bool = False,
timeout: int = 300,
poll_interval: int = 5,
) -> VMStatusDict:
"""
Get the current runtime status of a virtual machine.
Optionally waits for the VM to acquire an IP address on its management
NIC (net0), useful when provisioning a VM and waiting for it to boot.
Args:
vmid (string) - hypervisor-internal VM ID.
wait_for_ip (bool) - if True, poll until the VM's guest-agent reports
an IP address on the first NIC (net0, the management NIC). Useful
for ``create_vm_from_cloud_init()`` follow-up. If False, return
immediate status without waiting (may have no IP).
timeout (int) - maximum seconds to wait for IP acquisition
(if wait_for_ip=True). Raises RuntimeError if timeout is exceeded.
Default 300.
poll_interval (int) - seconds between status polls (if wait_for_ip=True).
Default 5.
Returns:
VMStatusDict - ``{"status": str, "ip_address": str, "hostname": str,
"mac_address": str}``. ip_address, hostname, and mac_address are only
present if the VM is running and has network info available.
Raises:
RuntimeError - if the VM does not exist or if wait_for_ip=True and
timeout is exceeded.
"""
raise NotImplementedError
+17
View File
@@ -704,3 +704,20 @@ class HealthMetricsDict(TypedDict):
load_5: NotRequired[float]
load_15: NotRequired[float]
interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]]
class VMProvisionResultDict(TypedDict):
"""Return value of ``create_vm_from_cloud_init()`` — provisioned VM identifier."""
vmid: str # Hypervisor-internal VM ID (string for portability across hypervisors)
name: str # VM display name
node: str # Cluster node name (empty string for standalone hypervisors)
class VMStatusDict(TypedDict):
"""Return value of ``get_vm_status()`` — runtime information of a VM."""
status: str # "running", "stopped", "paused", "suspended", etc.
ip_address: NotRequired[str] # Management NIC IP (absent if VM has no IP or is stopped)
hostname: NotRequired[str] # Hostname resolved from IP (if available)
mac_address: NotRequired[str] # MAC of management NIC