Claus Lohmar
7e1303a217
1. BIOS: provisioner now calls detect_efi() on the first disk when auto_detect_boot=true. Detects EFI partition via guestfish. Falls back to user's boot_type if detection inconclusive. 2. Reuse: 'Create Another VM' now shows an inline form asking only for VM Name and ID, reusing the same disk image, format, and boot settings from the completed job. No redirect needed.
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""
|
|
VM provisioning on the Proxmox host — real qm integration.
|
|
|
|
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 threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from models import (
|
|
JobSubmissionRequest,
|
|
JobStatusResponse,
|
|
JobStatus,
|
|
DiskType,
|
|
BootType,
|
|
)
|
|
from converter import extract_if_needed, discover_disk, detect_efi
|
|
|
|
logger = logging.getLogger("backend.provisioner")
|
|
|
|
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:
|
|
"""Queue a conversion job and start it in a background thread."""
|
|
job_id = f"job_{req.vmid}_{int(time.time())}"
|
|
|
|
out_dir = STAGING_OUT / str(req.vmid)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
with _jobs_lock:
|
|
_jobs[job_id] = {
|
|
"vmid": req.vmid,
|
|
"status": JobStatus.QUEUED,
|
|
"progress": 0,
|
|
"message": "Queued",
|
|
"error": None,
|
|
"request": req,
|
|
"source_files": [],
|
|
"disk_paths": [],
|
|
}
|
|
|
|
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 accepted and queued.",
|
|
)
|
|
|
|
|
|
def get_job_status(job_id: str) -> JobStatusResponse:
|
|
"""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=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."""
|
|
out_dir = STAGING_OUT / str(vmid)
|
|
|
|
if not delete:
|
|
return {
|
|
"job_id": f"job_{vmid}",
|
|
"action_taken": "retained",
|
|
"message": f"Staging files preserved for VM {vmid}",
|
|
}
|
|
|
|
if out_dir.exists():
|
|
shutil.rmtree(out_dir, ignore_errors=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 _convert_with_progress(
|
|
job_id: str, source_path: Path, output_path: Path, format: str,
|
|
start_pct: int, end_pct: int,
|
|
):
|
|
"""Run qemu-img convert in background and update job progress
|
|
by polling the output file size against the source virtual size."""
|
|
# Get source virtual size for percentage estimation
|
|
info = _run(["qemu-img", "info", "--output=json", str(source_path)])
|
|
source_size = json.loads(info.stdout).get("virtual-size", 0)
|
|
if source_size == 0:
|
|
source_size = source_path.stat().st_size
|
|
|
|
proc = subprocess.Popen(
|
|
["qemu-img", "convert", "-f", format, "-O", "qcow2",
|
|
str(source_path), str(output_path)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
# Poll output file size and update progress
|
|
while proc.poll() is None:
|
|
time.sleep(2)
|
|
if output_path.exists():
|
|
current = output_path.stat().st_size
|
|
if source_size > 0:
|
|
ratio = min(current / source_size, 1.0)
|
|
pct = start_pct + int(ratio * (end_pct - start_pct))
|
|
gb_done = current / (1024**3)
|
|
gb_total = source_size / (1024**3)
|
|
_update_job(job_id, progress=pct,
|
|
message=f"Converting disk... {gb_done:.1f} / {gb_total:.1f} GiB")
|
|
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(f"qemu-img convert failed (rc={proc.returncode})")
|
|
|
|
_update_job(job_id, progress=end_pct, message="Disk converted.")
|
|
|
|
|
|
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:
|
|
# Extract archive if needed, then locate the actual disk image
|
|
source = extract_if_needed(spec.source_filename)
|
|
disk = discover_disk(source)
|
|
source_path = disk.path
|
|
logger.info("Disk %d: source=%s format=%s size=%.1f GiB",
|
|
idx, source_path, disk.format, disk.size_gb)
|
|
|
|
# Convert with real-time progress
|
|
_convert_with_progress(
|
|
job_id, source_path, output_path, disk.format,
|
|
start_pct=10 + idx * 10, end_pct=55 + idx * 10,
|
|
)
|
|
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=60,
|
|
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...")
|
|
|
|
# Sanitize VM name: Proxmox requires DNS-safe names (a-z, 0-9, hyphens)
|
|
safe_name = re.sub(r'[^a-zA-Z0-9-]', '-', req.vm_name).strip('-').lower()
|
|
if not safe_name:
|
|
safe_name = f"vm-{vmid}"
|
|
|
|
# ── Auto-detect boot type if requested ─────────────────────
|
|
final_boot = req.boot_type
|
|
if req.auto_detect_boot and disk_paths:
|
|
logger.info("Auto-detecting boot type...")
|
|
efi = detect_efi(disk_paths[0])
|
|
if efi is True:
|
|
final_boot = BootType.UEFI
|
|
logger.info(" EFI partition detected → UEFI (OVMF)")
|
|
elif efi is False:
|
|
final_boot = BootType.LEGACY
|
|
logger.info(" No EFI partition → Legacy BIOS (SeaBIOS)")
|
|
else:
|
|
logger.info(" EFI detection inconclusive — using %s", final_boot.value)
|
|
|
|
bios = "ovmf" if final_boot.value == "uefi" else "seabios"
|
|
_run(["qm", "create", str(vmid),
|
|
"--name", safe_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", "disk", "import", 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 final_boot.value == "uefi":
|
|
logger.info("Adding EFI disk")
|
|
_run(["qm", "set", str(vmid), "--efidisk0",
|
|
f"{req.target_storage}:0,format=raw,size=4M"])
|
|
|
|
# ── Finalise ───────────────────────────────────────────────
|
|
_update_job(job_id, progress=95, message="Configuring VM...")
|
|
_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))
|