Claus Lohmar
affcde7afb
The job submission receives the original filename (e.g. Debian_13_VMG.7z) but qemu-img convert needs the actual VMDK path inside the archive. Now reuse extract_if_needed() + discover_disk() from converter module to locate the real disk image before conversion.
301 lines
12 KiB
Python
301 lines
12 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,
|
|
)
|
|
from converter import extract_if_needed, discover_disk
|
|
|
|
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 _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)
|
|
|
|
_run(["qemu-img", "convert",
|
|
"-f", disk.format,
|
|
"-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))
|