268 lines
9.8 KiB
Python
268 lines
9.8 KiB
Python
"""
|
|
Backend API — Proxmox Image Conversion Engine
|
|
|
|
Runs on the Proxmox host (srv2) at http://10.2.0.2:9000
|
|
All API routes live under /api/v1/.
|
|
Matches open-api.yaml v1.1.0 spec exactly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from models import (
|
|
JobSubmissionRequest,
|
|
JobStatusResponse,
|
|
CleanupRequest,
|
|
CleanupResponse,
|
|
AnalyzeRequest,
|
|
AnalyzeResponse,
|
|
AnalyzeStatusResponse,
|
|
CloneRequest,
|
|
HealthResponse,
|
|
ErrorResponse,
|
|
)
|
|
from converter import extract_if_needed, discover_disk, detect_os, detect_efi, start_analysis, get_analysis_status
|
|
from provisioner import submit_job, get_job_status, cleanup_staging, clone_vm
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Logging
|
|
# ---------------------------------------------------------------------------
|
|
logger = logging.getLogger("backend")
|
|
logger.setLevel(logging.INFO)
|
|
|
|
# Console → systemd journal
|
|
_ch = logging.StreamHandler()
|
|
_ch.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
logger.addHandler(_ch)
|
|
|
|
# File handler → shared log directory
|
|
LOG_DIR = Path(__file__).resolve().parent.parent / "logs"
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
_fh = RotatingFileHandler(
|
|
LOG_DIR / "vm-bench-backend.log", maxBytes=10 * 1024 * 1024, backupCount=5,
|
|
)
|
|
_fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s"))
|
|
logger.addHandler(_fh)
|
|
|
|
logger.info("Backend starting — log file: %s", LOG_DIR / "vm-bench-backend.log")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
API_PREFIX = "/api/v1"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App
|
|
# ---------------------------------------------------------------------------
|
|
app = FastAPI(
|
|
title="Proxmox Image Conversion Engine",
|
|
version="1.1.0",
|
|
docs_url="/docs",
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _validate_filename(filename: str) -> None:
|
|
"""Reject filenames that attempt path traversal."""
|
|
if ".." in filename or filename.startswith("/"):
|
|
raise HTTPException(status_code=400, detail="Invalid filename: path traversal not allowed")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — all under /api/v1/
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.get(f"{API_PREFIX}/health", response_model=HealthResponse)
|
|
def health() -> HealthResponse:
|
|
"""Health check."""
|
|
return HealthResponse(status="ok")
|
|
|
|
|
|
@app.get(f"{API_PREFIX}/storage/pools")
|
|
def storage_pools():
|
|
"""Return available Proxmox storage pools that support VM disk images."""
|
|
pools = []
|
|
try:
|
|
cfg = Path("/etc/pve/storage.cfg")
|
|
if cfg.exists():
|
|
lines = cfg.read_text()
|
|
current = {}
|
|
for line in lines.split("\n"):
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if ":" in stripped and not line.startswith(" "):
|
|
if current and current.get("name"):
|
|
pools.append(current)
|
|
parts = stripped.split(":", 1)
|
|
stype = parts[0].strip()
|
|
sname = parts[1].strip() if len(parts) > 1 else ""
|
|
current = {"name": sname, "type": stype}
|
|
elif line.startswith(" ") and current:
|
|
if " " in stripped:
|
|
key, val = stripped.split(" ", 1)
|
|
current[key] = val
|
|
if current and current.get("name"):
|
|
pools.append(current)
|
|
except Exception:
|
|
pass
|
|
|
|
result = []
|
|
for p in pools:
|
|
if p.get("type") in ("lvmthin", "zfspool", "rbd", "dir"):
|
|
content = p.get("content", "")
|
|
if "images" in content:
|
|
result.append({
|
|
"name": p["name"],
|
|
"type": p["type"],
|
|
})
|
|
|
|
if not result:
|
|
result.append({"name": "local-lvm", "type": "lvmthin"})
|
|
|
|
return {"pools": result}
|
|
|
|
|
|
@app.get(f"{API_PREFIX}/sessions")
|
|
def list_sessions():
|
|
"""Return existing session directories with staged files that can be resumed."""
|
|
tmp = Path("/mnt/converter/tmp") if Path("/mnt/converter/tmp").exists() else Path(__file__).resolve().parent.parent / "tmp"
|
|
sessions = []
|
|
try:
|
|
for entry in sorted(tmp.iterdir(), reverse=True):
|
|
if not entry.is_dir():
|
|
continue
|
|
in_dir = entry / "in"
|
|
if not in_dir.is_dir():
|
|
continue
|
|
session_files = []
|
|
total_bytes = 0
|
|
has_disks = False
|
|
all_files = sorted([f for f in in_dir.rglob("*") if f.is_file()], key=lambda f: f.stat().st_size, reverse=True)
|
|
for f in all_files:
|
|
sz = f.stat().st_size
|
|
total_bytes += sz
|
|
rel = str(f.relative_to(in_dir))
|
|
session_files.append({"name": rel, "size_bytes": sz, "size_gb": round(sz / (1024**3), 1)})
|
|
if f.suffix.lower() in {".vmdk", ".qcow2", ".img", ".raw", ".vhd", ".vhdx"}:
|
|
has_disks = True
|
|
if not session_files:
|
|
continue
|
|
archives = {".7z", ".zip", ".rar", ".tar.gz", ".tgz", ".tar", ".gz", ".bz2", ".xz", ".ova"}
|
|
has_source = has_disks or any(f["name"].lower().endswith(tuple(archives)) for f in session_files)
|
|
sessions.append({
|
|
"session_id": entry.name,
|
|
"files": session_files[:10],
|
|
"total_size_gb": round(total_bytes / (1024**3), 1),
|
|
"has_disks": has_disks,
|
|
"resumable": has_source,
|
|
})
|
|
except Exception:
|
|
pass
|
|
return {"sessions": sessions}
|
|
|
|
|
|
@app.post(f"{API_PREFIX}/analyze", response_model=AnalyzeStatusResponse, status_code=202)
|
|
def analyze(req: AnalyzeRequest) -> AnalyzeStatusResponse:
|
|
"""Analyze a source file asynchronously: extraction + disk probing + OS detection."""
|
|
_validate_filename(req.source_filename)
|
|
logger.info("Analyze request: vmid=%d file=%s", req.vmid, req.source_filename)
|
|
try:
|
|
result = start_analysis(req.vmid, req.source_filename)
|
|
return AnalyzeStatusResponse(
|
|
analysis_id=result["analysis_id"],
|
|
vmid=req.vmid,
|
|
status=result["status"],
|
|
message=result["message"],
|
|
)
|
|
except Exception as exc:
|
|
logger.error("Analysis start failed: %s", exc)
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@app.get(f"{API_PREFIX}/analyze/{{analysis_id}}", response_model=AnalyzeStatusResponse)
|
|
def analyze_status(analysis_id: str) -> AnalyzeStatusResponse:
|
|
"""Poll async analysis status."""
|
|
try:
|
|
result = get_analysis_status(analysis_id)
|
|
return AnalyzeStatusResponse(**result)
|
|
except KeyError:
|
|
logger.warning("Analysis not found: %s", analysis_id)
|
|
raise HTTPException(status_code=404, detail=f"Analysis not found: {analysis_id}")
|
|
|
|
|
|
@app.post(f"{API_PREFIX}/jobs", response_model=JobStatusResponse, status_code=202)
|
|
def create_job(req: JobSubmissionRequest) -> JobStatusResponse:
|
|
"""Submit a conversion + provisioning job."""
|
|
logger.info("Job submit: vmid=%d name=%s disk=%s storage=%s",
|
|
req.vmid, req.vm_name,
|
|
req.boot_disk.source_filename, req.target_storage)
|
|
try:
|
|
result = submit_job(req)
|
|
logger.info("Job queued: %s", result.job_id)
|
|
return result
|
|
except Exception as exc:
|
|
logger.error("Job submission failed: %s", exc)
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@app.get(f"{API_PREFIX}/jobs/{{job_id}}", response_model=JobStatusResponse)
|
|
def job_status(job_id: str) -> JobStatusResponse:
|
|
"""Get current status of a conversion job."""
|
|
try:
|
|
result = get_job_status(job_id)
|
|
return result
|
|
except Exception:
|
|
logger.warning("Job not found: %s", job_id)
|
|
raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")
|
|
|
|
|
|
@app.post(f"{API_PREFIX}/jobs/{{job_id}}/cleanup", response_model=CleanupResponse)
|
|
def cleanup_job(job_id: str, req: CleanupRequest) -> CleanupResponse:
|
|
"""Delete or preserve staging files for a job."""
|
|
try:
|
|
vmid = int(job_id.split("_")[1])
|
|
except (IndexError, ValueError):
|
|
vmid = 0
|
|
logger.info("Cleanup: job=%s vmid=%d delete=%s", job_id, vmid, req.delete_staging_files)
|
|
result = cleanup_staging(vmid, req.delete_staging_files, req.session_id)
|
|
return CleanupResponse(
|
|
job_id=result["job_id"],
|
|
action_taken=result["action_taken"],
|
|
message=result["message"],
|
|
)
|
|
|
|
|
|
@app.post(f"{API_PREFIX}/clone", response_model=JobStatusResponse, status_code=202)
|
|
def clone_vm_endpoint(req: CloneRequest) -> JobStatusResponse:
|
|
"""Clone an existing VM with a new ID and name."""
|
|
logger.info("Clone request: %d → %d (%s)", req.source_vmid, req.target_vmid, req.target_name)
|
|
try:
|
|
return clone_vm(req)
|
|
except Exception as exc:
|
|
logger.error("Clone failed: %s", exc)
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run
|
|
# ---------------------------------------------------------------------------
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=9000)
|