fix(qemu): Fixed missing image replacement for linked-clone nodes

This commit is contained in:
Cristi 2026-09-18 23:58:49 +03:00
parent 59a367b88d
commit da16dc9874
7 changed files with 308 additions and 27 deletions

View File

@ -64,6 +64,7 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s
qemu = Qemu.instance() qemu = Qemu.instance()
node_data = jsonable_encoder(node_data, exclude_unset=True) node_data = jsonable_encoder(node_data, exclude_unset=True)
disk_images_to_reset = set(node_data.pop("disk_images_to_reset", []))
vm = await qemu.create_node( vm = await qemu.create_node(
node_data.pop("name"), node_data.pop("name"),
str(project_id), str(project_id),
@ -80,12 +81,35 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s
# update the disk image with the backing file if provided # update the disk image with the backing file if provided
# this is needed when duplicating a node that uses backed disk images # this is needed when duplicating a node that uses backed disk images
drives = ["a", "b", "c", "d"] drives = ["a", "b", "c", "d"]
for disk_index, drive in enumerate(drives): for drive in drives:
disk_image_backing_file = node_data.get(f"hd{drive}_disk_image_backing_file") disk_image_backing_file = node_data.get(f"hd{drive}_disk_image_backing_file")
if disk_image_backing_file: if disk_image_backing_file:
log.debug(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}") log.debug(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}")
node_data[f"hd{drive}_disk_image"] = disk_image_backing_file node_data[f"hd{drive}_disk_image"] = disk_image_backing_file
# Validate every explicitly replaced disk before removing its stale
# overlay. Other unresolved disks may still make this create request fail,
# but a valid replacement applied in stages must not retain its old layer.
for drive in drives:
disk_image_property = f"hd{drive}_disk_image"
if disk_image_property in disk_images_to_reset:
replacement_image = node_data.get(disk_image_property)
if replacement_image:
vm.manager.get_abs_image_path(replacement_image, vm.working_dir)
local_disk_name = f"hd{drive}_disk.qcow2"
local_disk = os.path.join(vm.working_dir, local_disk_name)
if vm.linked_clone and os.path.exists(local_disk):
# A degraded linked clone is being assigned a new base image.
# Its old overlay depends on the unavailable base and cannot
# safely be rebased onto an arbitrary replacement. Discard it
# so start creates a fresh overlay from the selected image.
log.info(
"Removing stale linked-clone disk '%s' before using replacement image '%s'",
local_disk,
node_data.get(disk_image_property),
)
vm.delete_disk_image(local_disk_name)
for name, value in node_data.items(): for name, value in node_data.items():
if hasattr(vm, name) and getattr(vm, name) != value: if hasattr(vm, name) and getattr(vm, name) != value:
setattr(vm, name, value) setattr(vm, name, value)

View File

@ -1777,6 +1777,9 @@ class QemuVM(BaseNode):
try: try:
os.remove(disk_path) os.remove(disk_path)
md5sum_path = disk_path + ".md5sum"
if os.path.exists(md5sum_path):
os.remove(md5sum_path)
except OSError as e: except OSError as e:
raise QemuError(f"Could not delete '{disk_name}' disk image: {e}") raise QemuError(f"Could not delete '{disk_name}' disk image: {e}")

View File

@ -494,7 +494,7 @@ class Node:
def links(self): def links(self):
return self._links return self._links
async def create(self, allow_missing_image=False): async def create(self, allow_missing_image=False, disk_images_to_reset=None):
""" """
Create the node on the compute Create the node on the compute
@ -502,9 +502,13 @@ class Node:
cannot be uploaded/synced, the node is not created on the compute cannot be uploaded/synced, the node is not created on the compute
but is kept on the controller and flagged with the missing image(s) but is kept on the controller and flagged with the missing image(s)
instead of raising. instead of raising.
:param disk_images_to_reset: QEMU disk properties whose linked-clone
overlays must be recreated after replacing a missing base image.
""" """
data = self._node_data() data = self._node_data()
data["node_id"] = self._id data["node_id"] = self._id
if self._node_type == "qemu" and disk_images_to_reset:
data["disk_images_to_reset"] = sorted(disk_images_to_reset)
if self._node_type == "docker": if self._node_type == "docker":
timeout = None timeout = None
await self._add_docker_image_digest(data) await self._add_docker_image_digest(data)
@ -576,6 +580,41 @@ class Node:
raise last_missing_error raise last_missing_error
return False return False
def _find_local_image(self, image_type, image):
"""Return an image path contained in an allowed image directory."""
if not image or image_type == "docker":
return None
try:
directories = images_directories(image_type)
except NotImplementedError:
return None
requested = os.path.normpath(image)
for directory in directories:
root = os.path.realpath(directory)
candidate = os.path.realpath(os.path.join(root, requested))
try:
contained = os.path.commonpath((root, candidate)) == root
except ValueError:
contained = False
if contained and os.path.isfile(candidate):
return candidate
# Image API records expose a basename separately from their
# controller-local absolute path. Search subfolders when that
# portable basename is used by a node or remote compute.
if os.path.basename(requested) == requested:
for current_root, _, filenames in os.walk(root):
if requested in filenames:
nested_candidate = os.path.realpath(os.path.join(current_root, requested))
try:
nested_contained = os.path.commonpath((root, nested_candidate)) == root
except ValueError:
nested_contained = False
if nested_contained and os.path.isfile(nested_candidate):
return nested_candidate
return None
def _image_available(self, image_type, image): def _image_available(self, image_type, image):
""" """
Check whether an image is available on the controller (and can Check whether an image is available on the controller (and can
@ -590,14 +629,7 @@ class Node:
if image_type == "docker": if image_type == "docker":
# Docker images are not stored on the controller filesystem # Docker images are not stored on the controller filesystem
return True return True
try: return self._find_local_image(image_type, image) is not None
directories = images_directories(image_type)
except NotImplementedError:
return True
for directory in directories:
if os.path.exists(os.path.join(directory, image)):
return True
return False
def _compute_missing_images(self, fallback_image=None): def _compute_missing_images(self, fallback_image=None):
""" """
@ -641,6 +673,29 @@ class Node:
missing.append({"property": prop, "image": fallback_image, "image_type": image_type}) missing.append({"property": prop, "image": fallback_image, "image_type": image_type})
return missing return missing
def _remove_replaced_image_metadata(self, old_properties, new_properties):
"""Remove metadata that still refers to an image being replaced."""
replaced_properties = {
missing.get("property")
for missing in self._missing_images
if missing.get("property")
and old_properties.get(missing["property"]) != new_properties.get(missing["property"])
}
if self._node_type == "qemu":
for prop in replaced_properties:
new_properties.pop(f"{prop}_md5sum", None)
if prop in {"hda_disk_image", "hdb_disk_image", "hdc_disk_image", "hdd_disk_image"}:
# The QEMU create route treats a backing-file value as the
# authoritative base image. Keeping the old value would
# silently override the replacement selected by the user.
new_properties.pop(f"{prop}_backing_file", None)
elif self._node_type == "dynamips" and "image" in replaced_properties:
new_properties.pop("image_md5sum", None)
elif self._node_type == "iou" and "path" in replaced_properties:
new_properties.pop("md5sum", None)
return replaced_properties
async def update(self, **kwargs): async def update(self, **kwargs):
""" """
Update the node on the compute Update the node on the compute
@ -665,7 +720,7 @@ class Node:
# We update properties on the compute and wait for the answer from the compute node # We update properties on the compute and wait for the answer from the compute node
if prop == "properties": if prop == "properties":
compute_properties = kwargs[prop] compute_properties = copy.deepcopy(kwargs[prop])
else: else:
if ( if (
prop == "name" prop == "name"
@ -681,7 +736,10 @@ class Node:
if self.missing_image and compute_properties is not None: if self.missing_image and compute_properties is not None:
# The node was never created on the compute (missing image). Apply # The node was never created on the compute (missing image). Apply
# the new properties locally so create() can use them. # the new properties locally so create() can use them.
replaced_image_properties = self._remove_replaced_image_metadata(old_properties, compute_properties)
self._properties = compute_properties self._properties = compute_properties
else:
replaced_image_properties = set()
self._list_ports() self._list_ports()
if update_compute: if update_compute:
if self.missing_image: if self.missing_image:
@ -689,7 +747,10 @@ class Node:
# have been provided. If it is still missing the node simply # have been provided. If it is still missing the node simply
# remains in its degraded state. # remains in its degraded state.
try: try:
created = await self.create(allow_missing_image=True) created = await self.create(
allow_missing_image=True,
disk_images_to_reset=replaced_image_properties,
)
except Exception: except Exception:
# create() can reject the replacement for reasons other # create() can reject the replacement for reasons other
# than a missing image. Keep the controller state aligned # than a missing image. Keep the controller state aligned
@ -979,20 +1040,19 @@ class Node:
if self._node_type == "docker": if self._node_type == "docker":
return await self._sync_missing_docker_image(img) return await self._sync_missing_docker_image(img)
for directory in images_directories(type): image = self._find_local_image(type, img)
image = os.path.join(directory, img) if image is None:
if os.path.exists(image): return False
self.project.emit_notification("log.info", {"message": f"Uploading missing image {img}"}) self.project.emit_notification("log.info", {"message": f"Uploading missing image {img}"})
try: try:
with open(image, "rb") as f: with open(image, "rb") as f:
await self._compute.post( await self._compute.post(
f"/{self._node_type}/images/{os.path.basename(img)}", data=f, timeout=None f"/{self._node_type}/images/{os.path.basename(img)}", data=f, timeout=None
) )
except OSError as e: except OSError as e:
raise ControllerError(f"Can't upload {image}: {str(e)}") raise ControllerError(f"Can't upload {image}: {str(e)}")
self.project.emit_notification("log.info", {"message": f"Upload finished for {img}"}) self.project.emit_notification("log.info", {"message": f"Upload finished for {img}"})
return True return True
return False
async def _add_docker_image_digest(self, data): async def _add_docker_image_digest(self, data):
""" """

View File

@ -221,7 +221,10 @@ class QemuCreate(QemuBase):
Properties to create a Qemu node. Properties to create a Qemu node.
""" """
pass disk_images_to_reset: Optional[List[str]] = Field(
None,
description="Disk image properties whose stale linked-clone overlays must be recreated",
)
class QemuUpdate(QemuBase): class QemuUpdate(QemuBase):

View File

@ -20,6 +20,7 @@ import pytest_asyncio
import os import os
import stat import stat
import shutil import shutil
import uuid
from fastapi import FastAPI, status from fastapi import FastAPI, status
from httpx import AsyncClient from httpx import AsyncClient
@ -144,6 +145,128 @@ class TestQemuNodesRoutes:
assert response.json()["project_id"] == compute_project.id assert response.json()["project_id"] == compute_project.id
assert response.json()["ram"] == 1024 assert response.json()["ram"] == 1024
assert response.json()["hda_disk_image"] == "linux载.img" assert response.json()["hda_disk_image"] == "linux载.img"
async def test_qemu_create_discards_stale_overlay_for_replacement(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str
):
node_id = str(uuid.uuid4())
working_dir = os.path.join(compute_project.path, "project-files", "qemu", node_id)
os.makedirs(working_dir)
stale_overlay = os.path.join(working_dir, "hda_disk.qcow2")
with open(stale_overlay, "w+") as f:
f.write("stale linked clone")
stale_checksum = stale_overlay + ".md5sum"
with open(stale_checksum, "w+") as f:
f.write("0" * 32)
params = {
**base_params,
"node_id": node_id,
"linked_clone": True,
"hda_disk_image": "linux载.img",
"disk_images_to_reset": ["hda_disk_image"],
}
response = await compute_client.post(
app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["hda_disk_image"] == "linux载.img"
assert not os.path.exists(stale_overlay)
assert not os.path.exists(stale_checksum)
async def test_qemu_create_preserves_existing_overlay_without_reset_marker(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str
):
node_id = str(uuid.uuid4())
working_dir = os.path.join(compute_project.path, "project-files", "qemu", node_id)
os.makedirs(working_dir)
overlay = os.path.join(working_dir, "hda_disk.qcow2")
shutil.copy("tests/resources/empty8G.qcow2", overlay)
params = {
**base_params,
"node_id": node_id,
"linked_clone": True,
"hda_disk_image": "linux载.img",
}
response = await compute_client.post(
app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params
)
assert response.status_code == status.HTTP_201_CREATED
assert os.path.exists(overlay)
async def test_qemu_create_preserves_overlay_when_replacement_is_missing(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str
):
node_id = str(uuid.uuid4())
working_dir = os.path.join(compute_project.path, "project-files", "qemu", node_id)
os.makedirs(working_dir)
overlay = os.path.join(working_dir, "hda_disk.qcow2")
shutil.copy("tests/resources/empty8G.qcow2", overlay)
params = {
**base_params,
"node_id": node_id,
"linked_clone": True,
"hda_disk_image": "does-not-exist.qcow2",
"disk_images_to_reset": ["hda_disk_image"],
}
response = await compute_client.post(
app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params
)
assert response.status_code == status.HTTP_409_CONFLICT
assert os.path.exists(overlay)
async def test_qemu_create_resets_valid_replacement_when_another_disk_is_missing(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str
):
node_id = str(uuid.uuid4())
working_dir = os.path.join(compute_project.path, "project-files", "qemu", node_id)
os.makedirs(working_dir)
stale_overlay = os.path.join(working_dir, "hda_disk.qcow2")
shutil.copy("tests/resources/empty8G.qcow2", stale_overlay)
params = {
**base_params,
"node_id": node_id,
"linked_clone": True,
"hda_disk_image": "linux载.img",
"hdb_disk_image": "does-not-exist.qcow2",
"disk_images_to_reset": ["hda_disk_image"],
}
response = await compute_client.post(
app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params
)
assert response.status_code == status.HTTP_409_CONFLICT
assert not os.path.exists(stale_overlay)
@pytest.mark.parametrize( @pytest.mark.parametrize(

View File

@ -806,6 +806,19 @@ def test_hda_disk_image(vm, images_dir):
assert vm.hda_disk_image == force_unix_path(os.path.join(images_dir, "QEMU", "test2")) assert vm.hda_disk_image == force_unix_path(os.path.join(images_dir, "QEMU", "test2"))
def test_delete_disk_image_removes_cached_checksum(vm):
disk_path = os.path.join(vm.working_dir, "hda_disk.qcow2")
checksum_path = disk_path + ".md5sum"
open(disk_path, "w+").close()
open(checksum_path, "w+").close()
vm.delete_disk_image("hda_disk.qcow2")
assert not os.path.exists(disk_path)
assert not os.path.exists(checksum_path)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hda_disk_image_non_linked_clone(vm, images_dir, compute_project, manager, fake_qemu_binary): async def test_hda_disk_image_non_linked_clone(vm, images_dir, compute_project, manager, fake_qemu_binary):
""" """

View File

@ -584,6 +584,45 @@ async def test_update_recreates_missing_image_node(project, compute, node):
assert project.restore_deferred_links.called is True assert project.restore_deferred_links.called is True
@pytest.mark.asyncio
async def test_update_replaces_qemu_linked_clone_backing_image(project, compute):
"""Stale linked-clone metadata must not override a replacement image."""
qemu_node = Node(
project,
compute,
"old-qemu",
node_id=str(uuid.uuid4()),
node_type="qemu",
properties={
"hda_disk_image": "hda_disk.qcow2",
"hda_disk_image_backing_file": "missing.qcow2",
"hda_disk_image_md5sum": "old-checksum",
},
)
qemu_node._missing_images = [
{"property": "hda_disk_image", "image": "missing.qcow2", "image_type": "qemu"}
]
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
project.restore_deferred_links = AsyncioMagicMock()
await qemu_node.update(
properties={
**qemu_node.properties,
"hda_disk_image": "replacement.qcow2",
}
)
request_data = compute.post.call_args.kwargs["data"]
assert request_data["hda_disk_image"] == "replacement.qcow2"
assert request_data["disk_images_to_reset"] == ["hda_disk_image"]
assert "hda_disk_image_backing_file" not in request_data
assert "hda_disk_image_md5sum" not in request_data
assert qemu_node.missing_image is False
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_update_missing_image_node_rolls_back_after_create_error(project, compute, node): async def test_update_missing_image_node_rolls_back_after_create_error(project, compute, node):
original_properties = {"image": "missing.image"} original_properties = {"image": "missing.image"}
@ -1005,6 +1044,22 @@ async def test_upload_missing_image(compute, controller, images_dir):
compute.post.assert_called_with("/qemu/images/linux.img", data=ANY, timeout=None) compute.post.assert_called_with("/qemu/images/linux.img", data=ANY, timeout=None)
@pytest.mark.asyncio
async def test_upload_missing_image_from_nested_directory(compute, controller, images_dir):
project = Project(str(uuid.uuid4()), controller=controller)
node = Node(project, compute, "demo",
node_id=str(uuid.uuid4()),
node_type="qemu",
properties={"hda_disk_image": "nested.img"})
nested_dir = os.path.join(images_dir, "vendor")
os.makedirs(nested_dir)
open(os.path.join(nested_dir, "nested.img"), "w+").close()
assert await node._upload_missing_image("qemu", "nested.img") is True
compute.post.assert_called_with("/qemu/images/nested.img", data=ANY, timeout=None)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_sync_missing_docker_image_from_controller_daemon(compute, controller): async def test_sync_missing_docker_image_from_controller_daemon(compute, controller):