diff --git a/backend/provisioner.py b/backend/provisioner.py index c6ccb4c..1044364 100644 --- a/backend/provisioner.py +++ b/backend/provisioner.py @@ -1,15 +1,19 @@ """ -VM provisioning on the Proxmox host. +VM provisioning on the Proxmox host — real qm integration. -This module is a stub — the real Proxmox integration (qm create, qm importdisk, -qm set) runs on the backend host (srv2). Fill in the actual Proxmox commands -when you deploy the backend to the host. +Runs qemu-img convert, virt-resize (optional), qm create / importdisk / set. +Jobs run in background threads with status tracked in memory. """ from __future__ import annotations +import json +import logging +import os +import re +import shutil import subprocess -import uuid +import threading import time from pathlib import Path from typing import Optional @@ -18,78 +22,73 @@ from models import ( JobSubmissionRequest, JobStatusResponse, JobStatus, + DiskType, ) +logger = logging.getLogger("backend.provisioner") + +STAGING_IN = Path("/mnt/converter/in") STAGING_OUT = Path("/mnt/converter/out") +# In-memory job tracking — survives as long as the process runs +_jobs: dict[str, dict] = {} +_jobs_lock = threading.Lock() + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- def submit_job(req: JobSubmissionRequest) -> JobStatusResponse: - """Submit a conversion + provisioning job. Returns initial status.""" + """Queue a conversion job and start it in a background thread.""" job_id = f"job_{req.vmid}_{int(time.time())}" - # Create output staging directory out_dir = STAGING_OUT / str(req.vmid) out_dir.mkdir(parents=True, exist_ok=True) - # In the real backend, this would: - # 1. Convert the source image to QCOW2 via qemu-img convert - # 2. Resize if target_disk_size_gb is set - # 3. Create the VM with qm create - # 4. Import the disk: qm importdisk - # 5. Configure boot: qm set --boot order=scsi0 --bios ovmf - # 6. Auto-start if configured: qm set --onboot 1 + with _jobs_lock: + _jobs[job_id] = { + "vmid": req.vmid, + "status": JobStatus.QUEUED, + "progress": 0, + "message": "Queued", + "error": None, + "request": req, + "source_files": [], + "disk_paths": [], + } - # Stub: simulate a job being queued + thread = threading.Thread(target=_process_job, args=(job_id, req), daemon=True) + thread.start() + + logger.info("Job %s queued for VM %d — thread started", job_id, req.vmid) return JobStatusResponse( job_id=job_id, vmid=req.vmid, status=JobStatus.QUEUED, progress_percentage=0, - message="Job queued for processing", - error_details=None, + message="Job accepted and queued.", ) def get_job_status(job_id: str) -> JobStatusResponse: - """Query the current status of a job. - - In the real backend, this would check the actual conversion process - (qemu-img progress, qm task status, etc.). - - For now, returns a simulated progressing status to demonstrate the flow. - """ - # Stub: simulate progress advancing - parts = job_id.split("_") - vmid = int(parts[1]) if len(parts) >= 2 else 0 - ts = int(parts[2]) if len(parts) >= 3 else 0 - elapsed = int(time.time()) - ts - - if elapsed < 5: - progress = min(int(elapsed * 10), 40) - status = JobStatus.PROCESSING_CONVERSION - msg = "Converting disk image..." - elif elapsed < 10: - progress = 40 + min(int((elapsed - 5) * 10), 50) - status = JobStatus.IMPORTING_STORAGE - msg = "Importing disk to Proxmox storage..." - else: - progress = 100 - status = JobStatus.COMPLETED - msg = "VM created successfully" - + """Return current status of a job from the in-memory store.""" + with _jobs_lock: + job = _jobs.get(job_id) + if not job: + raise KeyError(job_id) return JobStatusResponse( job_id=job_id, - vmid=vmid, - status=status, - progress_percentage=progress, - message=msg, - error_details=None, + vmid=job["vmid"], + status=job["status"], + progress_percentage=job["progress"], + message=job.get("message", ""), + error_details=job.get("error"), ) def cleanup_staging(vmid: int, delete: bool) -> dict: - """Remove staging files for a VM ID. Returns action details.""" - in_dir = Path("/mnt/converter/in") + """Remove staging files for a VM ID.""" out_dir = STAGING_OUT / str(vmid) if not delete: @@ -99,18 +98,202 @@ def cleanup_staging(vmid: int, delete: bool) -> dict: "message": f"Staging files preserved for VM {vmid}", } - # Remove extracted source files in /mnt/converter/in for this vmid - # (In production, you'd track which files belong to which job) - cleaned_in = False - cleaned_out = False - if out_dir.exists(): - import shutil shutil.rmtree(out_dir, ignore_errors=True) - cleaned_out = True + logger.info("Cleaned up output dir: %s", out_dir) return { "job_id": f"job_{vmid}", "action_taken": "purged", "message": f"Staging files deleted for VM {vmid}", } + + +# --------------------------------------------------------------------------- +# Background job processor +# --------------------------------------------------------------------------- + +def _update_job(job_id: str, *, status: Optional[JobStatus] = None, + progress: Optional[int] = None, message: Optional[str] = None, + error: Optional[str] = None): + with _jobs_lock: + j = _jobs.get(job_id) + if not j: + return + if status is not None: + j["status"] = status + if progress is not None: + j["progress"] = progress + if message is not None: + j["message"] = message + if error is not None: + j["error"] = error + + +def _run(cmd: list[str], check: bool = True, timeout: int = 3600) -> subprocess.CompletedProcess: + logger.debug("CMD: %s", " ".join(cmd)) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + if result.stdout: + logger.debug(" stdout: %s", result.stdout.strip()[:500]) + if result.stderr: + logger.debug(" stderr: %s", result.stderr.strip()[:500]) + if check and result.returncode != 0: + raise RuntimeError( + f"Command failed (rc={result.returncode}): {' '.join(cmd)}\n" + f"stderr: {result.stderr.strip()[:500]}" + ) + return result + + +def _process_job(job_id: str, req: JobSubmissionRequest): + try: + vmid = req.vmid + logger.info("=== Job %s started for VM %d ===", job_id, vmid) + + _update_job(job_id, status=JobStatus.PROCESSING_CONVERSION, progress=10, + message="Converting disk image...") + + out_dir = STAGING_OUT / str(vmid) + out_dir.mkdir(parents=True, exist_ok=True) + + # Collect all disks (boot disk + additional) + disk_specs = [req.boot_disk] + (req.additional_disks or []) + disk_paths = [] + + for idx, spec in enumerate(disk_specs): + output_path = out_dir / f"vm-{vmid}-disk-{idx}.qcow2" + + if spec.disk_type == DiskType.IMAGE_FILE: + source_path = STAGING_IN / spec.source_filename + if not source_path.exists(): + raise FileNotFoundError(f"Source file not found: {source_path}") + + logger.info("Converting disk %d: %s → %s", idx, source_path, output_path) + _run(["qemu-img", "convert", + "-f", spec.format.value if spec.format else "vmdk", + "-O", "qcow2", + str(source_path), str(output_path)], + timeout=7200) + _update_job(job_id, progress=20 + (idx + 1) * 15, + message=f"Converted disk {idx + 1}/{len(disk_specs)}") + else: + size_gb = spec.size_gb or 32 + logger.info("Creating empty disk %d: %d GiB", idx, size_gb) + _run(["qemu-img", "create", "-f", "qcow2", + str(output_path), f"{size_gb}G"]) + _update_job(job_id, progress=20 + (idx + 1) * 15, + message=f"Created empty disk {idx + 1}") + + disk_paths.append(output_path) + + # ── Resize boot disk if needed ────────────────────────────── + boot_disk = disk_paths[0] if disk_paths else None + if boot_disk and boot_disk.exists(): + info_json = _run(["qemu-img", "info", "--output=json", str(boot_disk)]).stdout + info = json.loads(info_json) + current_bytes = info.get("virtual-size", 0) + current_gb = current_bytes / (1024 ** 3) + + target_gb = req.target_disk_size_gb + if target_gb is None and current_gb > 30: + target_gb = 30 + logger.info("Auto-shrink: %.1f GiB → %d GiB", current_gb, target_gb) + elif target_gb is None: + target_gb = int(round(current_gb)) + 1 + + if target_gb < current_gb: + _update_job(job_id, progress=65, + message=f"Shrinking disk to {target_gb} GiB...") + logger.info("Shrinking disk: %.1f GiB → %d GiB via virt-resize", current_gb, target_gb) + try: + temp = Path(str(boot_disk) + ".resized") + _run(["qemu-img", "create", "-f", "qcow2", str(temp), f"{target_gb}G"]) + _run(["virt-resize", "--shrink", "--resize-force", + str(boot_disk), str(temp)], timeout=7200) + os.remove(boot_disk) + os.rename(temp, boot_disk) + logger.info("Disk shrunk successfully via virt-resize") + except Exception: + logger.warning("virt-resize failed — falling back to qemu-img resize (may corrupt FS)") + _run(["qemu-img", "resize", "--shrink", str(boot_disk), f"{target_gb}G"]) + elif target_gb > current_gb: + logger.info("Expanding disk: %.1f GiB → %d GiB", current_gb, target_gb) + _run(["qemu-img", "resize", str(boot_disk), f"{target_gb}G"]) + + # ── Detach any existing VM with this ID ───────────────────── + existing = subprocess.run(["qm", "status", str(vmid)], + capture_output=True, text=True) + if "does not exist" not in existing.stderr and "does not exist" not in existing.stdout: + logger.warning("VM %d already exists — destroying it first", vmid) + subprocess.run(["qm", "stop", str(vmid)], capture_output=True) + subprocess.run(["qm", "destroy", str(vmid), "--purge"], capture_output=True) + time.sleep(2) + + # ── Create VM ────────────────────────────────────────────── + _update_job(job_id, status=JobStatus.IMPORTING_STORAGE, progress=70, + message="Creating VM...") + + bios = "ovmf" if req.boot_type.value == "uefi" else "seabios" + _run(["qm", "create", str(vmid), + "--name", req.vm_name, + "--bios", bios, + "--cores", str(req.cpu_cores), + "--memory", str(req.ram_mb), + "--net0", "virtio,bridge=vmbr0", + "--scsihw", "virtio-scsi-single"]) + + # ── Import and attach each disk ──────────────────────────── + for idx, disk_path in enumerate(disk_paths): + logger.info("Importing disk %d: %s", idx, disk_path) + result = _run(["qm", "importdisk", str(vmid), str(disk_path), + req.target_storage]) + output = result.stdout + result.stderr + + # Parse volume ID from qm importdisk output + match = re.search(r"imported disk '([^']+)'", output) + if not match: + match = re.search(r"Formatting '([^']+)'", output) + if not match: + match = re.search(r"Successfully imported disk as '([^']+)'", output) + if not match: + raise RuntimeError( + f"Could not parse volume ID from qm importdisk output:\n{output}" + ) + vol_id = match.group(1) + logger.info(" Volume ID: %s", vol_id) + + scsi_id = f"scsi{idx}" + _run(["qm", "set", str(vmid), f"--{scsi_id}", vol_id]) + _update_job(job_id, progress=75 + idx * 10, + message=f"Imported disk {idx + 1}/{len(disk_paths)}") + + # ── Add EFI disk if UEFI ────────────────────────────────── + if req.boot_type.value == "uefi": + logger.info("Adding EFI disk") + _run(["qm", "set", str(vmid), "--efidisk0", + f"{req.target_storage}:0,format=raw,size=4M"]) + + # ── Finalise ─────────────────────────────────────────────── + _run(["qm", "set", str(vmid), "--boot", "order=scsi0"]) + _run(["qm", "set", str(vmid), "--serial0", "socket"]) + + # Log final config for debugging + config = _run(["qm", "config", str(vmid)], check=False) + logger.info("Final VM config:\n%s", config.stdout.strip()) + + _update_job(job_id, status=JobStatus.COMPLETED, progress=100, + message=f"VM {vmid} created and ready.") + + with _jobs_lock: + j = _jobs.get(job_id) + if j: + j["disk_paths"] = disk_paths + j["source_files"] = [spec.source_filename for spec in disk_specs + if spec.source_filename] + + logger.info("=== Job %s completed: VM %d ===", job_id, vmid) + + except Exception as exc: + logger.exception("Job %s failed", job_id) + _update_job(job_id, status=JobStatus.FAILED, progress=0, + message="Job failed", error=str(exc))