From 9264cdcba94e78c1bdf20e9a27f9390546ba36d0 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 7 Jul 2026 10:37:36 +0200 Subject: [PATCH] feat(vm_provision_mixin): create_vm_from_cloud_init downloads cloud images directly Replaces the template-clone flow with: create empty VM shell, download the cloud image on the node (cached by filename, optional checksum verification), qm importdisk, attach as scsi0. NIC config, snippet upload, ssh keys, disk resize, and start remain unchanged (already generic). New helpers: _run_node_command (strict SSH exec with custom timeout and non-zero-exit detection, unlike the best-effort _exec_ssh_command), _download_cloud_image (idempotent download + checksum check), _find_default_image_storage (content=images discovery, mirrors the existing snippet-storage discovery). 24 tests pass (10 new: _run_node_command x2, _download_cloud_image x4, plus rewrites of the 4 existing create_vm_from_cloud_init tests for the new flow). --- napalm_proxmox/vm_provision_mixin.py | 187 +++++++++++++++++++++++---- tests/test_vm_provision_mixin.py | 186 +++++++++++++++++++++----- 2 files changed, 316 insertions(+), 57 deletions(-) diff --git a/napalm_proxmox/vm_provision_mixin.py b/napalm_proxmox/vm_provision_mixin.py index dd6106a..64d339c 100644 --- a/napalm_proxmox/vm_provision_mixin.py +++ b/napalm_proxmox/vm_provision_mixin.py @@ -12,10 +12,105 @@ from napalm_device_types.models import NetworkTargetDict, VMProvisionResultDict, _logger = logging.getLogger(__name__) +# Downloaded cloud images are cached here on the hypervisor node, keyed by +# filename, so provisioning multiple VMs from the same image only pays the +# download cost once. +_IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images" + class ProxmoxVMProvisionMixin: """Mixin to add VM provisioning to ProxmoxDriver.""" + def _run_node_command(self, command: str, timeout: int) -> str: + """ + Execute a shell command on the Proxmox node via SSH, raising on failure. + + Unlike ``_exec_ssh_command`` (best-effort, fixed timeout, swallows + errors), this is for critical provisioning steps — image download, + disk import — where a non-zero exit or a caller-specific timeout must + surface as a hard failure rather than an empty string. + """ + import paramiko + + if self._ssh_client is None: + ssh_user = self._ssh_username or self.username + ssh_pass = self._ssh_password or self.password + ssh_pkey = None + if self._ssh_key and not ssh_pass: + from io import StringIO as _StringIO + + ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key)) + self._ssh_client = paramiko.SSHClient() + self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + connect_kwargs: Dict[str, Any] = { + "hostname": self.hostname, + "port": 22, + "username": ssh_user, + "timeout": self.timeout, + } + if ssh_pkey: + connect_kwargs["pkey"] = ssh_pkey + else: + connect_kwargs["password"] = ssh_pass + self._ssh_client.connect(**connect_kwargs) + + _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout) + exit_status = stdout.channel.recv_exit_status() + out = stdout.read().decode().strip() + err = stderr.read().decode().strip() + if exit_status != 0: + raise RuntimeError(f"Command failed (exit {exit_status}): {command}\n{err or out}") + return out + + def _download_cloud_image( + self, image_url: str, image_checksum: str | None, timeout: int + ) -> str: + """ + Download image_url to the node's image cache dir if not already present. + + Returns the local path on the hypervisor node. Verifies image_checksum + (format ":", e.g. "sha256:abc123...") if given, re-downloading + is left to the caller's next attempt if verification fails. + """ + filename = image_url.rstrip("/").rsplit("/", 1)[-1] + local_path = f"{_IMAGE_CACHE_DIR}/{filename}" + + exists = self._run_node_command( + f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} && echo EXISTS || echo MISSING", + timeout=30, + ) + if "EXISTS" not in exists: + _logger.info(f"Downloading cloud image {image_url} -> {local_path}") + self._run_node_command( + f"wget -q -O {local_path}.tmp '{image_url}' && mv {local_path}.tmp {local_path}", + timeout=timeout, + ) + + if image_checksum: + algo, _, expected = image_checksum.partition(":") + algo = (algo or "sha256").lower() + actual = self._run_node_command( + f"{algo}sum {local_path} | awk '{{print $1}}'", timeout=60 + ) + if actual.lower() != expected.lower(): + # Remove the bad file so a retry re-downloads instead of reusing it. + self._run_node_command(f"rm -f {local_path}", timeout=30) + raise RuntimeError( + f"Checksum mismatch for {image_url}: expected {expected}, got {actual}" + ) + + return local_path + + def _find_default_image_storage(self) -> str: + """Find a storage suitable for VM root disks (content includes 'images').""" + for storage in self._api.storage.get(): + content = storage.get("content", "") + if "images" in content and storage.get("enabled"): + return storage["storage"] + raise ValueError( + "No storage with content='images' found. Configure a storage for VM disks." + ) + def _wait_for_task(self, upid: str, timeout: int = 120) -> None: """ Poll a Proxmox task until completion. @@ -51,68 +146,104 @@ class ProxmoxVMProvisionMixin: self, name: str, *, - template: str, + image_url: str, cpu: int, memory: int, nics: List[Dict[str, Any]], cloud_init_config: Dict[str, Any], + image_checksum: str | None = None, ssh_public_keys: List[str] | None = None, disk_resize_gb: int | None = None, + download_timeout: int = 300, timeout: int = 180, ) -> VMProvisionResultDict: """ - Create a new VM from a Cloud-Init template via Proxmox API. + Create a new VM from a downloaded cloud image via Proxmox API. Steps: 1. Get next available VMID from cluster - 2. Clone template VM (full clone, new VMID) - 3. Configure CPU, memory, and network interfaces - 4. Verify snippet storage exists - 5. Render cloud-init config to YAML and upload - 6. Set Cloud-Init config references and SSH keys - 7. Optionally resize root disk - 8. Start the VM - 9. Return VMID, name, node + 2. Create an empty VM shell (no clone — no pre-existing template needed) + 3. Download the cloud image on the node (cached by filename) and + import it as the VM's root disk + 4. Configure CPU, memory, and network interfaces + 5. Verify snippet storage exists + 6. Render cloud-init config to YAML and upload + 7. Set Cloud-Init config references and SSH keys + 8. Optionally resize root disk + 9. Start the VM + 10. Return VMID, name, node Args: name: new VM display name - template: template VMID/name to clone from + image_url: URL of the cloud image to download and use as root disk cpu: number of vCPUs memory: RAM in MB nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag) cloud_init_config: user-data dict (will be YAML-rendered) + image_checksum: expected ":" checksum of the image, verified + after download (None = no verification) ssh_public_keys: SSH public keys to inject disk_resize_gb: resize root disk to this size (None = no resize) - timeout: max seconds for provisioning + download_timeout: max seconds for the image download (skipped if cached) + timeout: max seconds for the remaining provisioning steps Returns: {"vmid": str, "name": str, "node": str} Raises: - RuntimeError: provisioning failure (clone, config, timeout, etc.) + RuntimeError: provisioning failure (download, import, config, timeout, etc.) ValueError: invalid storage or configuration """ try: - _logger.info(f"Creating VM '{name}' from template {template}") + _logger.info(f"Creating VM '{name}' from image {image_url}") # Step 1: Get next VMID next_vmid = self._api.cluster.nextid.get() vmid = int(next_vmid) _logger.info(f"Allocated VMID {vmid}") - # Step 2: Clone template - _logger.info(f"Cloning template {template} → VMID {vmid}") - clone_upid = self._node_api().qemu(template).clone.post( - newid=vmid, - full=1, + # Step 2: Create empty VM shell (no disks yet) + _logger.info(f"Creating VM shell {vmid}") + self._node_api().qemu.post( + vmid=vmid, name=name, + memory=memory, + cores=cpu, + ostype="l26", + scsihw="virtio-scsi-pci", ) - self._wait_for_task(clone_upid, timeout=timeout) - # Step 3: Configure CPU, memory, and NICs - _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)") + # Step 3: Download cloud image (cached) and import as root disk + local_path = self._download_cloud_image( + image_url, image_checksum, timeout=download_timeout + ) + image_storage = self._find_default_image_storage() - config_args = {"cores": cpu, "memory": memory} + _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}") + self._run_node_command( + f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2", + timeout=timeout, + ) + + # Proxmox leaves the imported disk as an "unusedN" reference — find + # it and attach it as the boot disk. + imported_config = self._node_api().qemu(vmid).config.get() + unused_value = next( + (v for k, v in imported_config.items() if k.startswith("unused")), None + ) + if not unused_value: + raise RuntimeError( + f"Disk import for VM {vmid} did not produce an unused disk reference" + ) + self._node_api().qemu(vmid).config.post( + scsi0=f"{unused_value},discard=on", + boot="order=scsi0", + ) + + # Step 4: Configure network interfaces (CPU/memory already set at shell creation) + _logger.info(f"Configuring {len(nics)} NIC(s) for VM {vmid}") + + config_args: Dict[str, Any] = {} # Build NIC config strings generically for i, nic in enumerate(nics): @@ -134,7 +265,7 @@ class ProxmoxVMProvisionMixin: self._node_api().qemu(vmid).config.post(**config_args) - # Step 4: Verify snippet storage exists + # Step 5: Verify snippet storage exists _logger.info("Checking for snippet storage...") storages = self._api.storage.get() snippet_storage = None @@ -152,7 +283,7 @@ class ProxmoxVMProvisionMixin: ) _logger.info(f"Using snippet storage: {snippet_storage}") - # Step 5: Render and upload Cloud-Init config + # Step 6: Render and upload Cloud-Init config _logger.info(f"Rendering Cloud-Init config for VMID {vmid}") user_data_yaml = "#cloud-config\n" + yaml.dump( @@ -169,7 +300,7 @@ class ProxmoxVMProvisionMixin: data=user_data_yaml, ) - # Step 6: Configure Cloud-Init references and SSH keys + # Step 7: Configure Cloud-Init references and SSH keys _logger.info(f"Setting Cloud-Init config for VM {vmid}") cloud_init_args = { @@ -191,7 +322,7 @@ class ProxmoxVMProvisionMixin: self._node_api().qemu(vmid).config.post(**cloud_init_args) - # Step 7: Optionally resize root disk + # Step 8: Optionally resize root disk if disk_resize_gb is not None: _logger.info(f"Resizing root disk to {disk_resize_gb}GB") # Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first) @@ -212,7 +343,7 @@ class ProxmoxVMProvisionMixin: except Exception as e: _logger.warning(f"Failed to resize disk: {e}, continuing anyway") - # Step 8: Start the VM + # Step 9: Start the VM _logger.info(f"Starting VM {vmid}") start_upid = self._node_api().qemu(vmid).status.start.post() self._wait_for_task(start_upid, timeout=timeout) diff --git a/tests/test_vm_provision_mixin.py b/tests/test_vm_provision_mixin.py index c030c3f..bb64015 100644 --- a/tests/test_vm_provision_mixin.py +++ b/tests/test_vm_provision_mixin.py @@ -75,19 +75,25 @@ def test_create_vm_from_cloud_init_single_nic(): mock_api = MagicMock() mock_api.cluster.nextid.get.return_value = 101 mock_api.storage.get.return_value = [ - {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1} + {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}, + {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}, ] mock_node = MagicMock() mixin._api = mock_api mixin._node_api = MagicMock(return_value=mock_node) + mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2") + mixin._run_node_command = MagicMock(return_value="") # Mock VM operations mock_vm = MagicMock() mock_node.qemu.return_value = mock_vm - mock_vm.clone.post.return_value = "UPID:pve1:123:clone" + mock_node.qemu.post.return_value = None mock_vm.config.post.return_value = None - mock_vm.config.get.return_value = {"scsi0": "local:100/vm-101-disk-0.raw"} + mock_vm.config.get.return_value = { + "unused0": "local-lvm:vm-101-disk-0", + "scsi0": "local-lvm:vm-101-disk-0", + } mock_vm.status.start.post.return_value = "UPID:pve1:124:start" # Mock task completion @@ -103,7 +109,7 @@ def test_create_vm_from_cloud_init_single_nic(): with patch("time.sleep"): result = mixin.create_vm_from_cloud_init( name="test-vm", - template="100", + image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", cpu=2, memory=2048, nics=[{"bridge": "vmbr0", "vlan_tag": 10}], @@ -114,13 +120,15 @@ def test_create_vm_from_cloud_init_single_nic(): assert result["vmid"] == "101" assert result["name"] == "test-vm" assert result["node"] == "pve1" - assert mock_vm.clone.post.called + assert mock_node.qemu.post.called + mixin._download_cloud_image.assert_called_once() # Verify NIC config was set correctly: net0 with tag=10, DHCP enabled - config_call_args = mock_vm.config.post.call_args_list[0] # First call (cores/memory/net0) - assert "net0" in config_call_args[1] - assert "tag=10" in config_call_args[1]["net0"] - assert "vmbr0" in config_call_args[1]["net0"] + net_call_args = next( + c for c in mock_vm.config.post.call_args_list if "net0" in c[1] + ) + assert "tag=10" in net_call_args[1]["net0"] + assert "vmbr0" in net_call_args[1]["net0"] def test_create_vm_from_cloud_init_dual_nic_trunk(): @@ -132,19 +140,25 @@ def test_create_vm_from_cloud_init_dual_nic_trunk(): mock_api = MagicMock() mock_api.cluster.nextid.get.return_value = 102 mock_api.storage.get.return_value = [ - {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1} + {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}, + {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}, ] mock_node = MagicMock() mixin._api = mock_api mixin._node_api = MagicMock(return_value=mock_node) + mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2") + mixin._run_node_command = MagicMock(return_value="") # Mock VM operations mock_vm = MagicMock() mock_node.qemu.return_value = mock_vm - mock_vm.clone.post.return_value = "UPID:pve1:125:clone" + mock_node.qemu.post.return_value = None mock_vm.config.post.return_value = None - mock_vm.config.get.return_value = {"scsi0": "local:100/vm-102-disk-0.raw"} + mock_vm.config.get.return_value = { + "unused0": "local-lvm:vm-102-disk-0", + "scsi0": "local-lvm:vm-102-disk-0", + } mock_vm.status.start.post.return_value = "UPID:pve1:126:start" # Mock task completion @@ -160,7 +174,7 @@ def test_create_vm_from_cloud_init_dual_nic_trunk(): with patch("time.sleep"): result = mixin.create_vm_from_cloud_init( name="wireshark-sat-1", - template="100", + image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", cpu=4, memory=4096, nics=[ @@ -175,16 +189,18 @@ def test_create_vm_from_cloud_init_dual_nic_trunk(): assert result["name"] == "wireshark-sat-1" # Verify both NICs configured - config_call_args = mock_vm.config.post.call_args_list[0] - assert "net0" in config_call_args[1] - assert "net1" in config_call_args[1] - assert "tag=10" in config_call_args[1]["net0"] - assert "trunks=20;30" in config_call_args[1]["net1"] - assert "vmbr1" in config_call_args[1]["net1"] + net_call_args = next( + c for c in mock_vm.config.post.call_args_list if "net0" in c[1] + ) + assert "net1" in net_call_args[1] + assert "tag=10" in net_call_args[1]["net0"] + assert "trunks=20;30" in net_call_args[1]["net1"] + assert "vmbr1" in net_call_args[1]["net1"] # Verify DHCP config: ipconfig0 yes, ipconfig1 no - cloud_init_call_args = mock_vm.config.post.call_args_list[1] # Second call (ipconfig) - assert "ipconfig0" in cloud_init_call_args[1] + cloud_init_call_args = next( + c for c in mock_vm.config.post.call_args_list if "ipconfig0" in c[1] + ) assert cloud_init_call_args[1]["ipconfig0"] == "ip=dhcp" assert "ipconfig1" not in cloud_init_call_args[1] # net1 has no DHCP @@ -194,7 +210,7 @@ def test_create_vm_missing_snippet_storage(): mixin = ProxmoxVMProvisionMixin() mixin._node_name = "pve1" - # Mock API with no snippet storage + # Mock API with images storage but no snippet storage mock_api = MagicMock() mock_api.cluster.nextid.get.return_value = 101 mock_api.storage.get.return_value = [ @@ -203,13 +219,15 @@ def test_create_vm_missing_snippet_storage(): mixin._api = mock_api - # Mock node for clone operation (so we get to the storage check) mock_node = MagicMock() mixin._node_api = MagicMock(return_value=mock_node) + mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2") + mixin._run_node_command = MagicMock(return_value="") mock_vm = MagicMock() mock_node.qemu.return_value = mock_vm - mock_vm.clone.post.return_value = "UPID:pve1:123:clone" + mock_node.qemu.post.return_value = None mock_vm.config.post.return_value = None + mock_vm.config.get.return_value = {"unused0": "local-lvm:vm-101-disk-0"} # Mock task to allow clone to complete mock_task = MagicMock() @@ -220,7 +238,7 @@ def test_create_vm_missing_snippet_storage(): with patch("napalm_proxmox.vm_provision_mixin.time.sleep"): mixin.create_vm_from_cloud_init( name="test-vm", - template="100", + image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", cpu=2, memory=2048, nics=[{"bridge": "vmbr0"}], @@ -237,19 +255,25 @@ def test_create_vm_with_disk_resize(): mock_api = MagicMock() mock_api.cluster.nextid.get.return_value = 103 mock_api.storage.get.return_value = [ - {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1} + {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}, + {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}, ] mock_node = MagicMock() mixin._api = mock_api mixin._node_api = MagicMock(return_value=mock_node) + mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2") + mixin._run_node_command = MagicMock(return_value="") # Mock VM operations mock_vm = MagicMock() mock_node.qemu.return_value = mock_vm - mock_vm.clone.post.return_value = "UPID:pve1:127:clone" + mock_node.qemu.post.return_value = None mock_vm.config.post.return_value = None - mock_vm.config.get.return_value = {"scsi0": "local:100/vm-103-disk-0.raw"} + mock_vm.config.get.return_value = { + "unused0": "local-lvm:vm-103-disk-0", + "scsi0": "local-lvm:vm-103-disk-0", + } mock_vm.resize.put.return_value = None mock_vm.status.start.post.return_value = "UPID:pve1:128:start" @@ -266,7 +290,7 @@ def test_create_vm_with_disk_resize(): with patch("time.sleep"): result = mixin.create_vm_from_cloud_init( name="big-vm", - template="100", + image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", cpu=2, memory=2048, nics=[{"bridge": "vmbr0"}], @@ -511,3 +535,107 @@ def test_get_network_targets_bridge_has_no_fixed_vlan_tag(): targets = mixin.get_network_targets() assert "fixed_vlan_tag" not in targets[0] or targets[0]["fixed_vlan_tag"] is None + + +# --------------------------------------------------------------------------- +# _run_node_command +# --------------------------------------------------------------------------- + + +def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = ""): + """Build a mock (stdin, stdout, stderr) tuple as returned by exec_command.""" + mock_stdin = MagicMock() + mock_stdout = MagicMock() + mock_stderr = MagicMock() + mock_stdout.channel.recv_exit_status.return_value = exit_status + mock_stdout.read.return_value = stdout_text.encode() + mock_stderr.read.return_value = stderr_text.encode() + return mock_stdin, mock_stdout, mock_stderr + + +def test_run_node_command_success_returns_stdout(): + mixin = ProxmoxVMProvisionMixin() + mixin._ssh_client = MagicMock() + mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n") + + result = mixin._run_node_command("echo hello", timeout=10) + + assert result == "hello" + + +def test_run_node_command_nonzero_exit_raises(): + mixin = ProxmoxVMProvisionMixin() + mixin._ssh_client = MagicMock() + mixin._ssh_client.exec_command.return_value = _mock_ssh_exec( + 1, stderr_text="No such file or directory" + ) + + with pytest.raises(RuntimeError, match="No such file or directory"): + mixin._run_node_command("cat /nonexistent", timeout=10) + + +# --------------------------------------------------------------------------- +# _download_cloud_image +# --------------------------------------------------------------------------- + + +def test_download_cloud_image_skips_download_when_cached(): + mixin = ProxmoxVMProvisionMixin() + mixin._run_node_command = MagicMock(return_value="EXISTS") + + path = mixin._download_cloud_image( + "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", + None, + timeout=300, + ) + + assert path.endswith("debian-12-genericcloud-amd64.qcow2") + # Only the existence check ran — no wget call + assert mixin._run_node_command.call_count == 1 + assert "wget" not in mixin._run_node_command.call_args_list[0][0][0] + + +def test_download_cloud_image_downloads_when_missing(): + mixin = ProxmoxVMProvisionMixin() + mixin._run_node_command = MagicMock(return_value="MISSING") + + mixin._download_cloud_image( + "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", + None, + timeout=300, + ) + + commands = [c[0][0] for c in mixin._run_node_command.call_args_list] + assert any("wget" in cmd for cmd in commands) + + +def test_download_cloud_image_verifies_matching_checksum(): + mixin = ProxmoxVMProvisionMixin() + mixin._run_node_command = MagicMock( + side_effect=["EXISTS", "abc123"] # existence check, then checksum + ) + + path = mixin._download_cloud_image( + "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", + "sha256:abc123", + timeout=300, + ) + + assert path.endswith("debian-12-genericcloud-amd64.qcow2") + + +def test_download_cloud_image_checksum_mismatch_raises_and_removes_file(): + mixin = ProxmoxVMProvisionMixin() + mixin._run_node_command = MagicMock( + side_effect=["EXISTS", "wrong-checksum", ""] # existence, checksum, rm + ) + + with pytest.raises(RuntimeError, match="Checksum mismatch"): + mixin._download_cloud_image( + "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2", + "sha256:abc123", + timeout=300, + ) + + commands = [c[0][0] for c in mixin._run_node_command.call_args_list] + assert any(cmd.startswith("rm -f") for cmd in commands)