From 41b76b5472ae6b7f9b201ad17a5e27b2af256bad Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Mon, 13 Jul 2026 02:14:00 +0300 Subject: [PATCH] feat: implement QEMU clone and local migration --- README.md | 3 ++ app/handlers/qemu.py | 68 +++++++++++++++++++++++ app/main.py | 2 + app/tasks/qemu.py | 77 +++++++++++++++++++++++++++ docs/compatibility-0.1.0.md | 4 +- docs/original-prompt-gap-plan.md | 5 +- evidence/pve-9.2.3-0.1.0.json | 18 +++++++ tests/compatibility/test_proxmoxer.py | 13 +++++ tests/unit/test_qemu_handlers.py | 49 +++++++++++++++++ tests/unit/test_qemu_task.py | 44 +++++++++++++++ 10 files changed, 279 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index c8f9b02..bc8bd6c 100644 --- a/README.md +++ b/README.md @@ -85,6 +85,9 @@ QEMU snapshots are durable PostgreSQL records. Create, delete, and rollback use UPID tasks and the same per-VM lock as other mutations; rollback restores both the captured VM configuration and runtime state. Snapshot listing, inspection, and description updates are available through the native Proxmox API paths. +Full QEMU clones copy configuration into a new stopped VM, and local migration +moves a VM between seeded nodes through an explicit migrating state. Both are +durable UPID operations with VMID collision and target-node validation. Token lifecycle is available at `/access/users/{userid}/token[/{tokenid}]`. A generated secret is returned only diff --git a/app/handlers/qemu.py b/app/handlers/qemu.py index b961897..954cf91 100644 --- a/app/handlers/qemu.py +++ b/app/handlers/qemu.py @@ -322,6 +322,69 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: async def snapshot_rollback(request: Request, inputs: dict[str, Any]) -> str: return await snapshot_task("snapshot-rollback", request, inputs) + async def clone(request: Request, inputs: dict[str, Any]) -> str: + values = _values(inputs) + node, vmid, newid = str(values["node"]), str(values["vmid"]), str(values["newid"]) + source = await _qemu_resource(request, node, vmid) + if await _database(request).pool.fetchval( + """SELECT EXISTS(SELECT 1 FROM resources + WHERE external_id=$1 AND kind IN ('qemu','lxc'))""", + newid, + ): + raise ApiError(409, "VMID already exists") + target = str(values.get("target") or node) + if not await _database(request).pool.fetchval( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE name=$1)", target + ): + raise ApiError(404, "target node does not exist") + return await _create_task( + request, + node=target, + vmid=newid, + task_type="qemu-clone", + payload={ + "source_resource_id": str(source["id"]), + "source_vmid": vmid, + "node": target, + "vmid": int(newid), + "name": values.get("name"), + "description": values.get("description"), + "full": bool(values.get("full", False)), + }, + ) + + async def migrate_preconditions(request: Request, inputs: dict[str, Any]) -> dict[str, Any]: + values = _values(inputs) + await _qemu_resource(request, str(values["node"]), str(values["vmid"])) + target = str(values["target"]) + exists = await _database(request).pool.fetchval( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE name=$1)", target + ) + if not exists: + raise ApiError(404, "target node does not exist") + return {"local_disks": [], "local_resources": [], "running": False} + + async def migrate(request: Request, inputs: dict[str, Any]) -> str: + values = _values(inputs) + node, vmid, target = str(values["node"]), str(values["vmid"]), str(values["target"]) + resource = await _qemu_resource(request, node, vmid) + if target == node: + raise ApiError(400, "target node is the same as source node") + await migrate_preconditions(request, inputs) + return await _create_task( + request, + node=node, + vmid=vmid, + task_type="qemu-migrate", + payload={ + "resource_id": str(resource["id"]), + "node": node, + "target": target, + "vmid": vmid, + "online": bool(values.get("online", False)), + }, + ) + async def task_list(request: Request, inputs: dict[str, Any]) -> list[dict[str, Any]]: tasks = await TaskRepository(_database(request).pool).list_for_node( str(_values(inputs)["node"]) @@ -382,6 +445,9 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: registry.register( "/nodes/{node}/qemu/{vmid}/snapshot/{snapname}/rollback", "POST", snapshot_rollback ) + registry.register("/nodes/{node}/qemu/{vmid}/clone", "POST", clone) + registry.register("/nodes/{node}/qemu/{vmid}/migrate", "GET", migrate_preconditions) + registry.register("/nodes/{node}/qemu/{vmid}/migrate", "POST", migrate) registry.register("/nodes/{node}/tasks", "GET", task_list) registry.register("/nodes/{node}/tasks/{upid}/status", "GET", task_status) registry.register("/nodes/{node}/tasks/{upid}/log", "GET", task_log) @@ -405,6 +471,8 @@ async def _create_task( "qemu-snapshot-create": "qmsnapshot", "qemu-snapshot-delete": "qmdelsnapshot", "qemu-snapshot-rollback": "qmrollback", + "qemu-clone": "qmclone", + "qemu-migrate": "qmigrate", }[task_type] upid = str(Upid(node, pid, pid, timestamp, worker_type, vmid, str(request.state.principal))) try: diff --git a/app/main.py b/app/main.py index 6fabca6..7ae79fd 100644 --- a/app/main.py +++ b/app/main.py @@ -44,12 +44,14 @@ def create_app( repository, "simulator-worker", { + "qemu-clone": handler, "qemu-create": handler, "qemu-delete": handler, "qemu-reboot": handler, "qemu-reset": handler, "qemu-resume": handler, "qemu-shutdown": handler, + "qemu-migrate": handler, "qemu-snapshot-create": handler, "qemu-snapshot-delete": handler, "qemu-snapshot-rollback": handler, diff --git a/app/tasks/qemu.py b/app/tasks/qemu.py index 13b1cb6..03c834f 100644 --- a/app/tasks/qemu.py +++ b/app/tasks/qemu.py @@ -18,6 +18,8 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler: operation = task.task_type.removeprefix("qemu-") if operation == "create": return await _create(repository, task) + if operation == "clone": + return await _clone(repository, task) resource_id = uuid.UUID(str(task.payload["resource_id"])) if operation == "update": return await _update(repository, task, resource_id) @@ -27,6 +29,8 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler: return await _snapshot( repository, task, resource_id, operation.removeprefix("snapshot-") ) + if operation == "migrate": + return await _migrate(repository, task, resource_id, clock) async with repository.pool.acquire() as connection: row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id) if row is None: @@ -201,5 +205,78 @@ async def _snapshot( return {"snapshot": name, "operation": operation} +async def _clone(repository: TaskRepository, task: Task) -> dict[str, Any]: + source_id = uuid.UUID(str(task.payload["source_resource_id"])) + target_id = uuid.uuid4() + node, vmid = str(task.payload["node"]), int(task.payload["vmid"]) + async with repository.pool.acquire() as connection: + async with connection.transaction(): + source = await connection.fetchrow( + """SELECT r.state, v.config FROM resources r + JOIN virtual_machines v ON v.resource_id=r.id WHERE r.id=$1""", + source_id, + ) + target = await connection.fetchrow( + "SELECT id, cluster_id FROM nodes WHERE name=$1", node + ) + if source is None or target is None: + raise ValueError("clone source or target disappeared") + config = _object(source["config"]) + if task.payload.get("name") is not None: + config["name"] = task.payload["name"] + state = {**_object(source["state"]), **config, "status": "stopped"} + await connection.execute( + """INSERT INTO resources(id,node_id,cluster_id,kind,external_id,state,metadata) + VALUES($1,$2,$3,'qemu',$4,$5::jsonb,'{}'::jsonb)""", + target_id, + target["id"], + target["cluster_id"], + str(vmid), + json.dumps(state), + ) + await connection.execute( + """INSERT INTO virtual_machines(resource_id,cluster_id,vmid,config) + VALUES($1,$2,$3,$4::jsonb)""", + target_id, + target["cluster_id"], + vmid, + json.dumps(config), + ) + await repository.append_log(task.id, f"VM cloned to {vmid}") + return {"vmid": vmid, "node": node} + + +async def _migrate( + repository: TaskRepository, task: Task, resource_id: uuid.UUID, clock: Clock +) -> dict[str, Any]: + target = str(task.payload["target"]) + async with repository.pool.acquire() as connection: + row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id) + if row is None: + raise ValueError("resource disappeared") + state = _object(row["state"]) + transition = plan_transition(VmState(str(state["status"])), "migrate") + state["status"] = transition.intermediate + await connection.execute( + "UPDATE resources SET state=$2::jsonb WHERE id=$1", resource_id, json.dumps(state) + ) + await repository.append_log(task.id, f"migration to {target} started") + await clock.sleep(1.0) + async with repository.pool.acquire() as connection: + node = await connection.fetchrow("SELECT id FROM nodes WHERE name=$1", target) + if node is None: + raise ValueError("target node disappeared") + state["status"] = transition.after + await connection.execute( + """UPDATE resources SET node_id=$2,state=$3::jsonb,version=version+1, + updated_at=now() WHERE id=$1""", + resource_id, + node["id"], + json.dumps(state), + ) + await repository.append_log(task.id, f"migration to {target} completed") + return {"node": target, "status": str(transition.after)} + + def _object(value: object) -> dict[str, Any]: return json.loads(value) if isinstance(value, str) else dict(cast(Mapping[str, Any], value)) diff --git a/docs/compatibility-0.1.0.md b/docs/compatibility-0.1.0.md index 78179e3..a080713 100644 --- a/docs/compatibility-0.1.0.md +++ b/docs/compatibility-0.1.0.md @@ -9,8 +9,8 @@ Proxmox compatibility. | Level | Methods | Contract share | Evidence | |---|---:|---:|---| | Declared and dynamically routed | 675 | 100% | Imported immutable API Viewer artifact | -| Stateful semantics implemented on current main | 34 | 5.04% | Handler registry and unit/integration tests | -| Schema-only or explicitly unsupported | 641 | 94.96% | Default 501 fallback | +| Stateful semantics implemented on current main | 37 | 5.48% | Handler registry and unit/integration tests | +| Schema-only or explicitly unsupported | 638 | 94.52% | Default 501 fallback | | proxmoxer smoke exercised | 9 | 1.33% | Unmodified proxmoxer 2.3 compatibility test | The smoke set is `POST /access/ticket`, `GET /version`, `GET /nodes`, diff --git a/docs/original-prompt-gap-plan.md b/docs/original-prompt-gap-plan.md index 5e5bbd9..7520f9a 100644 --- a/docs/original-prompt-gap-plan.md +++ b/docs/original-prompt-gap-plan.md @@ -58,8 +58,9 @@ plaintext password, ticket, CSRF token, or token secret reaches storage/logs. - [x] Create, synchronous/asynchronous update, and delete. - [x] Shutdown, reboot, reset, suspend and resume. - [x] Snapshots and rollback. -- [ ] Clone, local/remote migration, resize and move disk, selected agent - endpoints, pending/status data. +- [x] Clone and local migration. +- [ ] Remote migration, resize and move disk, selected agent endpoints, + pending/status data. - [ ] Persist normalized CPU/memory/common fields plus unknown PVE parameters in JSONB; simulate usage, uptime, PID, IO/network, lock, template, QMP, HA and guest-agent availability. diff --git a/evidence/pve-9.2.3-0.1.0.json b/evidence/pve-9.2.3-0.1.0.json index 96823ef..bcb6854 100644 --- a/evidence/pve-9.2.3-0.1.0.json +++ b/evidence/pve-9.2.3-0.1.0.json @@ -146,6 +146,24 @@ "verb": "POST", "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_task.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/clone", + "verb": "POST", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/migrate", + "verb": "GET", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/migrate", + "verb": "POST", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"] } ] } diff --git a/tests/compatibility/test_proxmoxer.py b/tests/compatibility/test_proxmoxer.py index 38f0fbc..56fe7de 100644 --- a/tests/compatibility/test_proxmoxer.py +++ b/tests/compatibility/test_proxmoxer.py @@ -126,6 +126,19 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: assert wait_task(proxmox, snapshot_delete_upid)["exitstatus"] == "OK" assert not snapshots.get() + clone_upid = ( + proxmox.nodes("pve1").qemu("150").clone.post(newid=151, name="clone-by-proxmoxer", full=1) + ) + assert wait_task(proxmox, clone_upid)["exitstatus"] == "OK" + assert proxmox.nodes("pve1").qemu("151").config.get()["name"] == "clone-by-proxmoxer" + migration = proxmox.nodes("pve1").qemu("151").migrate + assert migration.get(target="pve2")["local_disks"] == [] + migrate_upid = migration.post(target="pve2", online=0) + assert wait_task(proxmox, migrate_upid)["exitstatus"] == "OK" + assert proxmox.nodes("pve2").qemu("151").config.get()["name"] == "clone-by-proxmoxer" + clone_delete_upid = proxmox.nodes("pve2").qemu("151").delete() + assert wait_task(proxmox, clone_delete_upid)["exitstatus"] == "OK" + delete_upid = proxmox.nodes("pve1").qemu("150").delete() assert wait_task(proxmox, delete_upid)["exitstatus"] == "OK" with pytest.raises(ResourceException) as deleted_vm: diff --git a/tests/unit/test_qemu_handlers.py b/tests/unit/test_qemu_handlers.py index 9410db5..1ed9368 100644 --- a/tests/unit/test_qemu_handlers.py +++ b/tests/unit/test_qemu_handlers.py @@ -262,3 +262,52 @@ async def test_qemu_snapshot_handlers(monkeypatch: pytest.MonkeyPatch) -> None: assert (await rollback(http_request, common)).startswith("UPID:pve1:") assert (await delete(http_request, common)).startswith("UPID:pve1:") assert tasks == ["qemu-snapshot-create", "qemu-snapshot-rollback", "qemu-snapshot-delete"] + + +async def test_qemu_clone_and_migrate_handlers(monkeypatch: pytest.MonkeyPatch) -> None: + tasks: list[dict[str, Any]] = [] + + class FakeTaskRepository: + def __init__(self, pool: object) -> None: + del pool + + async def create(self, **kwargs: Any) -> Task: + tasks.append(kwargs) + return Task( + uuid.uuid4(), + str(kwargs["upid"]), + str(kwargs["task_type"]), + "queued", + {}, + 0, + False, + 0, + ) + + monkeypatch.setattr("app.handlers.qemu.TaskRepository", FakeTaskRepository) + registry = HandlerRegistry() + register_qemu_handlers(registry) + pool = QemuPool() + http_request = request(pool) + clone = registry.get("/nodes/{node}/qemu/{vmid}/clone", "POST") + migrate_get = registry.get("/nodes/{node}/qemu/{vmid}/migrate", "GET") + migrate = registry.get("/nodes/{node}/qemu/{vmid}/migrate", "POST") + assert clone and migrate_get and migrate + + clone_upid = await clone( + http_request, inputs(node="pve1", vmid=150, newid=151, name="clone", full=True) + ) + assert clone_upid.startswith("UPID:pve1:") + assert tasks[-1]["task_type"] == "qemu-clone" + assert (await migrate_get(http_request, inputs(node="pve1", vmid=150, target="pve2")))[ + "local_disks" + ] == [] + migrate_upid = await migrate( + http_request, inputs(node="pve1", vmid=150, target="pve2", online=False) + ) + assert migrate_upid.startswith("UPID:pve1:") + assert tasks[-1]["task_type"] == "qemu-migrate" + + with pytest.raises(ApiError) as same_node: + await migrate(http_request, inputs(node="pve1", vmid=150, target="pve1")) + assert same_node.value.status_code == 400 diff --git a/tests/unit/test_qemu_task.py b/tests/unit/test_qemu_task.py index 8de6527..1905eca 100644 --- a/tests/unit/test_qemu_task.py +++ b/tests/unit/test_qemu_task.py @@ -82,6 +82,8 @@ class CrudConnection: return {"id": uuid.uuid4(), "cluster_id": uuid.uuid4()} if "JOIN virtual_machines" in sql: return {"state": '{"status":"stopped","name":"old"}', "config": '{"name":"old"}'} + if "SELECT state FROM resources" in sql: + return {"state": '{"status":"stopped","name":"old"}'} if "FROM snapshots" in sql: return { "state": ( @@ -217,3 +219,45 @@ async def test_qemu_worker_snapshot_create_rollback_and_delete_are_persistent() assert any("INSERT INTO snapshots" in command for command in commands) assert any("UPDATE virtual_machines" in command for command in commands) assert any("DELETE FROM snapshots" in command for command in commands) + + +async def test_qemu_worker_clone_and_migrate_are_persistent() -> None: + repository = CrudRepository() + handler = qemu_handler(cast(TaskRepository, repository), cast(Clock, ImmediateClock())) + resource_id = uuid.uuid4() + + cloned = await handler( + Task( + uuid.uuid4(), + "UPID:clone", + "qemu-clone", + "running", + { + "source_resource_id": str(resource_id), + "node": "pve1", + "vmid": 151, + "name": "clone", + }, + 0, + False, + 1, + ) + ) + migrated = await handler( + Task( + uuid.uuid4(), + "UPID:migrate", + "qemu-migrate", + "running", + {"resource_id": str(resource_id), "target": "pve2"}, + 0, + False, + 1, + ) + ) + + assert cloned == {"vmid": 151, "node": "pve1"} + assert migrated == {"node": "pve2", "status": "stopped"} + commands = repository.connection.commands + assert any("INSERT INTO resources" in command for command in commands) + assert any("node_id=$2" in command for command in commands)