From 003ee5d6348ce99273fcf463ea428778b28e4853 Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Mon, 13 Jul 2026 02:20:54 +0300 Subject: [PATCH] feat: implement QEMU disk resize and move --- README.md | 4 ++ app/api/registry.py | 19 ++++++ app/handlers/qemu.py | 92 +++++++++++++++++++++++++++ app/main.py | 1 + app/tasks/qemu.py | 46 ++++++++++++++ docs/compatibility-0.1.0.md | 4 +- docs/original-prompt-gap-plan.md | 4 +- evidence/pve-9.2.3-0.1.0.json | 12 ++++ tests/compatibility/test_proxmoxer.py | 13 +++- tests/unit/test_compatible_io.py | 12 ++++ tests/unit/test_qemu_handlers.py | 18 +++++- tests/unit/test_qemu_task.py | 21 ++++++ 12 files changed, 239 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index bc8bd6c..16cf48b 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,10 @@ 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. +Indexed contract fields such as `scsi[n]` accept their concrete Proxmox names +(`scsi0`, `scsi1`, and so on). QEMU disk growth is synchronous and rejects +shrinking; disk moves are durable tasks that update normalized disk metadata and +the preserved VM configuration. Token lifecycle is available at `/access/users/{userid}/token[/{tokenid}]`. A generated secret is returned only diff --git a/app/api/registry.py b/app/api/registry.py index 9a53c68..11fdc36 100644 --- a/app/api/registry.py +++ b/app/api/registry.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any, Literal, cast @@ -226,9 +227,16 @@ async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]: supplied.update(dict(parse_qsl((await request.body()).decode()))) definitions = {parameter.name: parameter.definition for parameter in method.parameters} + indexed = { + re.compile("^" + re.escape(name).replace(r"\[n\]", r"\d+") + "$"): definition + for name, definition in definitions.items() + if "[n]" in name + } errors: dict[str, str] = {} parsed: dict[str, Any] = {} for name, definition in definitions.items(): + if "[n]" in name: + continue if name not in supplied: if definition.optional: if definition.default is not None: @@ -240,7 +248,18 @@ async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]: parsed[name] = _coerce(supplied[name], definition) except (TypeError, ValueError) as exc: errors[name] = str(exc) + indexed_names: set[str] = set() for name in supplied.keys() - definitions.keys(): + indexed_definition = next( + (candidate for pattern, candidate in indexed.items() if pattern.fullmatch(name)), None + ) + if indexed_definition is not None: + indexed_names.add(name) + try: + parsed[name] = _coerce(supplied[name], indexed_definition) + except (TypeError, ValueError) as exc: + errors[name] = str(exc) + for name in supplied.keys() - definitions.keys() - indexed_names: if name not in request.path_params: errors[name] = "property is not defined in schema" if errors: diff --git a/app/handlers/qemu.py b/app/handlers/qemu.py index 954cf91..d8033f1 100644 --- a/app/handlers/qemu.py +++ b/app/handlers/qemu.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re from collections.abc import Mapping from typing import Any, cast @@ -385,6 +386,61 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: }, ) + async def resize(request: Request, inputs: dict[str, Any]) -> None: + values = _values(inputs) + node, vmid, disk = str(values["node"]), str(values["vmid"]), str(values["disk"]) + resource = await _qemu_resource(request, node, vmid) + config = _state(resource["config"]) + if disk not in config: + raise ApiError(400, f"disk {disk} does not exist") + current = _disk_size_bytes(str(config[disk])) + size = _resize_bytes(str(values["size"]), current) + config[disk] = _replace_disk_size(str(config[disk]), size) + status = await _database(request).pool.execute( + """UPDATE virtual_machines SET config=$2::jsonb + WHERE resource_id=$1""", + resource["id"], + json.dumps(config, sort_keys=True), + ) + if status != "UPDATE 1": + raise ApiError(409, "configuration changed concurrently") + await _database(request).pool.execute( + """UPDATE resources SET state=state || $2::jsonb,version=version+1, + updated_at=now() WHERE id=$1""", + resource["id"], + json.dumps({disk: config[disk]}, sort_keys=True), + ) + await _database(request).pool.execute( + """INSERT INTO vm_disks(id,resource_id,device,storage_id,size_bytes) + VALUES(gen_random_uuid(),$1,$2,$3,$4) + ON CONFLICT(resource_id,device) DO UPDATE SET size_bytes=EXCLUDED.size_bytes""", + resource["id"], + disk, + str(config[disk]).split(":", 1)[0], + size, + ) + + async def move_disk(request: Request, inputs: dict[str, Any]) -> str: + values = _values(inputs) + node, vmid, disk = str(values["node"]), str(values["vmid"]), str(values["disk"]) + resource = await _qemu_resource(request, node, vmid) + if disk not in _state(resource["config"]): + raise ApiError(400, f"disk {disk} does not exist") + return await _create_task( + request, + node=node, + vmid=vmid, + task_type="qemu-move-disk", + payload={ + "resource_id": str(resource["id"]), + "disk": disk, + "storage": str(values.get("storage") or "local-lvm"), + "target_vmid": int(values.get("target-vmid") or vmid), + "target_disk": str(values.get("target-disk") or disk), + "delete": bool(values.get("delete", True)), + }, + ) + 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"]) @@ -448,6 +504,8 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: 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}/qemu/{vmid}/resize", "PUT", resize) + registry.register("/nodes/{node}/qemu/{vmid}/move_disk", "POST", move_disk) 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) @@ -473,6 +531,7 @@ async def _create_task( "qemu-snapshot-rollback": "qmrollback", "qemu-clone": "qmclone", "qemu-migrate": "qmigrate", + "qemu-move-disk": "qmmove", }[task_type] upid = str(Upid(node, pid, pid, timestamp, worker_type, vmid, str(request.state.principal))) try: @@ -514,3 +573,36 @@ async def _snapshot(request: Request, values: dict[str, Any]) -> Any: if row is None: raise ApiError(404, "snapshot does not exist") return row + + +_SIZE_RE = re.compile(r"^(?P\d+)(?P[KMGT]?)$", re.IGNORECASE) + + +def _size_bytes(value: str) -> int: + match = _SIZE_RE.fullmatch(value.strip()) + if match is None: + raise ApiError(400, f"invalid disk size: {value}") + units = {"": 1, "K": 2**10, "M": 2**20, "G": 2**30, "T": 2**40} + return int(match.group("value")) * units[match.group("unit").upper()] + + +def _disk_size_bytes(value: str) -> int: + for part in value.split(","): + if part.startswith("size="): + return _size_bytes(part.removeprefix("size=")) + return 0 + + +def _resize_bytes(value: str, current: int) -> int: + if value.startswith("+"): + return current + _size_bytes(value[1:]) + result = _size_bytes(value) + if result < current: + raise ApiError(400, "shrinking disks is not supported") + return result + + +def _replace_disk_size(value: str, size: int) -> str: + parts = [part for part in value.split(",") if not part.startswith("size=")] + parts.append(f"size={size // 2**30}G" if size % 2**30 == 0 else f"size={size}") + return ",".join(parts) diff --git a/app/main.py b/app/main.py index 7ae79fd..c46c660 100644 --- a/app/main.py +++ b/app/main.py @@ -52,6 +52,7 @@ def create_app( "qemu-resume": handler, "qemu-shutdown": handler, "qemu-migrate": handler, + "qemu-move-disk": 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 03c834f..f2cb257 100644 --- a/app/tasks/qemu.py +++ b/app/tasks/qemu.py @@ -31,6 +31,8 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler: ) if operation == "migrate": return await _migrate(repository, task, resource_id, clock) + if operation == "move-disk": + return await _move_disk(repository, task, resource_id) async with repository.pool.acquire() as connection: row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id) if row is None: @@ -278,5 +280,49 @@ async def _migrate( return {"node": target, "status": str(transition.after)} +async def _move_disk( + repository: TaskRepository, task: Task, resource_id: uuid.UUID +) -> dict[str, Any]: + disk = str(task.payload["disk"]) + target_disk = str(task.payload["target_disk"]) + storage = str(task.payload["storage"]) + async with repository.pool.acquire() as connection: + async with connection.transaction(): + row = await connection.fetchrow( + "SELECT config FROM virtual_machines WHERE resource_id=$1", resource_id + ) + if row is None: + raise ValueError("resource disappeared") + config = _object(row["config"]) + if disk not in config: + raise ValueError("disk disappeared") + original = str(config[disk]) + suffix = original.split(":", 1)[1] if ":" in original else original + config[target_disk] = f"{storage}:{suffix}" + if bool(task.payload.get("delete", True)) and target_disk != disk: + config.pop(disk, None) + await connection.execute( + "UPDATE virtual_machines SET config=$2::jsonb WHERE resource_id=$1", + resource_id, + json.dumps(config, sort_keys=True), + ) + await connection.execute( + """UPDATE resources SET state=state || $2::jsonb, version=version+1, + updated_at=now() WHERE id=$1""", + resource_id, + json.dumps({target_disk: config[target_disk]}, sort_keys=True), + ) + await connection.execute( + """UPDATE vm_disks SET device=$2,storage_id=$3 + WHERE resource_id=$1 AND device=$4""", + resource_id, + target_disk, + storage, + disk, + ) + await repository.append_log(task.id, f"disk {disk} moved to {storage}") + return {"disk": target_disk, "storage": storage} + + 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 a080713..3c041a6 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 | 37 | 5.48% | Handler registry and unit/integration tests | -| Schema-only or explicitly unsupported | 638 | 94.52% | Default 501 fallback | +| Stateful semantics implemented on current main | 39 | 5.78% | Handler registry and unit/integration tests | +| Schema-only or explicitly unsupported | 636 | 94.22% | 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 7520f9a..78a1f86 100644 --- a/docs/original-prompt-gap-plan.md +++ b/docs/original-prompt-gap-plan.md @@ -59,8 +59,8 @@ plaintext password, ticket, CSRF token, or token secret reaches storage/logs. - [x] Shutdown, reboot, reset, suspend and resume. - [x] Snapshots and rollback. - [x] Clone and local migration. -- [ ] Remote migration, resize and move disk, selected agent endpoints, - pending/status data. +- [x] Resize and move disk. +- [ ] Remote migration, selected agent endpoints and 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 bcb6854..462a658 100644 --- a/evidence/pve-9.2.3-0.1.0.json +++ b/evidence/pve-9.2.3-0.1.0.json @@ -164,6 +164,18 @@ "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}/resize", + "verb": "PUT", + "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_compatible_io.py", "tests/unit/test_qemu_handlers.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/move_disk", + "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 56fe7de..4bfb218 100644 --- a/tests/compatibility/test_proxmoxer.py +++ b/tests/compatibility/test_proxmoxer.py @@ -94,7 +94,11 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: assert hidden.value.status_code == 403 create_upid = proxmox.nodes("pve1").qemu.post( - vmid=150, name="created-by-proxmoxer", cores=2, memory=1024 + vmid=150, + name="created-by-proxmoxer", + cores=2, + memory=1024, + scsi0="local-lvm:vm-150-disk-0,size=8G", ) with pytest.raises(ResourceException) as duplicate_create: proxmox.nodes("pve1").qemu.post(vmid=150, name="duplicate") @@ -110,6 +114,13 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: assert wait_task(proxmox, update_upid)["exitstatus"] == "OK" assert proxmox.nodes("pve1").qemu("150").config.get()["name"] == "async-update" + disk_api = proxmox.nodes("pve1").qemu("150") + assert disk_api.resize.put(disk="scsi0", size="+2G") is None + assert "size=10G" in disk_api.config.get()["scsi0"] + move_upid = disk_api.move_disk.post(disk="scsi0", storage="local") + assert wait_task(proxmox, move_upid)["exitstatus"] == "OK" + assert disk_api.config.get()["scsi0"].startswith("local:") + snapshots = proxmox.nodes("pve1").qemu("150").snapshot snapshot_upid = snapshots.post(snapname="baseline", description="before change") assert wait_task(proxmox, snapshot_upid)["exitstatus"] == "OK" diff --git a/tests/unit/test_compatible_io.py b/tests/unit/test_compatible_io.py index e75d3b7..a3aeaf3 100644 --- a/tests/unit/test_compatible_io.py +++ b/tests/unit/test_compatible_io.py @@ -23,6 +23,7 @@ async def client_for(tmp_path: Path) -> AsyncClient: Parameter(name="node", definition=Schema(type="string")), Parameter(name="count", definition=Schema(type="integer", minimum=1)), Parameter(name="force", definition=Schema(type="boolean", optional=True)), + Parameter(name="scsi[n]", definition=Schema(type="string", optional=True)), ), returns=Schema(type="null"), checksum="1" * 64, @@ -41,6 +42,8 @@ async def client_for(tmp_path: Path) -> AsyncClient: async def handler(_request: Request, inputs: dict[str, Any]) -> None: assert inputs["values"]["count"] >= 1 + if "scsi0" in inputs["values"]: + assert inputs["values"]["scsi0"] == "local:disk,size=8G" return None handlers.register("/nodes/{node}/test", "POST", handler) @@ -96,3 +99,12 @@ async def test_non_object_json_is_rejected_without_fastapi_body(tmp_path: Path) assert response.status_code == 400 assert response.json()["errors"] == {"body": "expected an object"} assert "detail" not in response.json() + + +async def test_indexed_contract_parameter_accepts_concrete_device(tmp_path: Path) -> None: + async with await client_for(tmp_path) as client: + response = await client.post( + "/api2/json/nodes/pve/test", json={"count": 1, "scsi0": "local:disk,size=8G"} + ) + + assert response.status_code == 200 diff --git a/tests/unit/test_qemu_handlers.py b/tests/unit/test_qemu_handlers.py index 1ed9368..b5b977d 100644 --- a/tests/unit/test_qemu_handlers.py +++ b/tests/unit/test_qemu_handlers.py @@ -66,7 +66,11 @@ class QemuPool: return {"state": '{"status":"stopped"}', "config": '{"name":"vm"}'} if "SELECT r.id, r.state" in sql: status = "running" if self.running else "stopped" - return {"id": self.resource_id, "state": f'{{"status":"{status}"}}'} + return { + "id": self.resource_id, + "state": f'{{"status":"{status}"}}', + "config": '{"scsi0":"local-lvm:vm-150-disk-0,size=8G"}', + } if "SELECT r.id, r.state, v.config" in sql: return {"id": self.resource_id, "state": '{"status":"stopped"}', "config": "{}"} if "SELECT s.* FROM snapshots" in sql: @@ -292,7 +296,9 @@ async def test_qemu_clone_and_migrate_handlers(monkeypatch: pytest.MonkeyPatch) 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 + resize = registry.get("/nodes/{node}/qemu/{vmid}/resize", "PUT") + move = registry.get("/nodes/{node}/qemu/{vmid}/move_disk", "POST") + assert clone and migrate_get and migrate and resize and move clone_upid = await clone( http_request, inputs(node="pve1", vmid=150, newid=151, name="clone", full=True) @@ -307,6 +313,14 @@ async def test_qemu_clone_and_migrate_handlers(monkeypatch: pytest.MonkeyPatch) ) assert migrate_upid.startswith("UPID:pve1:") assert tasks[-1]["task_type"] == "qemu-migrate" + assert ( + await resize(http_request, inputs(node="pve1", vmid=150, disk="scsi0", size="+2G")) is None + ) + move_upid = await move( + http_request, inputs(node="pve1", vmid=150, disk="scsi0", storage="local") + ) + assert move_upid.startswith("UPID:pve1:") + assert tasks[-1]["task_type"] == "qemu-move-disk" with pytest.raises(ApiError) as same_node: await migrate(http_request, inputs(node="pve1", vmid=150, target="pve1")) diff --git a/tests/unit/test_qemu_task.py b/tests/unit/test_qemu_task.py index 1905eca..a779247 100644 --- a/tests/unit/test_qemu_task.py +++ b/tests/unit/test_qemu_task.py @@ -84,6 +84,8 @@ class CrudConnection: return {"state": '{"status":"stopped","name":"old"}', "config": '{"name":"old"}'} if "SELECT state FROM resources" in sql: return {"state": '{"status":"stopped","name":"old"}'} + if "SELECT config FROM virtual_machines" in sql: + return {"config": '{"scsi0":"local-lvm:vm-150-disk-0,size=10G"}'} if "FROM snapshots" in sql: return { "state": ( @@ -255,9 +257,28 @@ async def test_qemu_worker_clone_and_migrate_are_persistent() -> None: 1, ) ) + moved = await handler( + Task( + uuid.uuid4(), + "UPID:move", + "qemu-move-disk", + "running", + { + "resource_id": str(resource_id), + "disk": "scsi0", + "target_disk": "scsi0", + "storage": "local", + "delete": True, + }, + 0, + False, + 1, + ) + ) assert cloned == {"vmid": 151, "node": "pve1"} assert migrated == {"node": "pve2", "status": "stopped"} + assert moved == {"disk": "scsi0", "storage": "local"} 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)