diff --git a/backend/app.py b/backend/app.py index 1d06ab1..c24771d 100644 --- a/backend/app.py +++ b/backend/app.py @@ -22,11 +22,12 @@ from models import ( CleanupResponse, AnalyzeRequest, AnalyzeResponse, + AnalyzeStatusResponse, CloneRequest, HealthResponse, ErrorResponse, ) -from converter import extract_if_needed, discover_disk, detect_os, detect_efi +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 # --------------------------------------------------------------------------- @@ -93,39 +94,33 @@ def health() -> HealthResponse: return HealthResponse(status="ok") -@app.post(f"{API_PREFIX}/analyze", response_model=AnalyzeResponse) -def analyze(req: AnalyzeRequest) -> AnalyzeResponse: - """Analyze a source file: extract if archive, find disk, detect OS + EFI.""" +@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: - source = extract_if_needed(req.source_filename) - logger.info("Source ready: %s", source) + 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("Extraction failed: %s", 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: - disk = discover_disk(source) - logger.info("Disk found: %s (format=%s, size=%.1f GiB)", disk.path, disk.format, disk.size_gb) - except Exception as exc: - logger.error("Disk discovery failed: %s", exc) - raise HTTPException(status_code=400, detail=f"Disk discovery failed: {exc}") - - extract_dir = source if source.is_dir() else source.parent - os_type = detect_os(disk.path, extract_dir) - efi = detect_efi(disk.path) - logger.info("Analysis result: os=%s efi=%s", os_type, efi) - - return AnalyzeResponse( - vmid=req.vmid, - filename=disk.path.name, - disk_format=disk.format, - disk_size_gb=disk.size_gb, - os_type=os_type, - efi_detectable=efi, - ) + 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) diff --git a/backend/converter.py b/backend/converter.py index ed2ce28..296577d 100644 --- a/backend/converter.py +++ b/backend/converter.py @@ -11,6 +11,8 @@ import os import re import shutil import subprocess +import threading +import time from pathlib import Path from dataclasses import dataclass from typing import Optional @@ -320,9 +322,166 @@ def detect_efi(disk_path: Path) -> Optional[bool]: ) if result.returncode != 0: return None - # EFI System Partition is typically VFAT; look for it if "vfat" in result.stdout.lower(): return True return False except Exception: return None + + +# --------------------------------------------------------------------------- +# Async analysis — background extraction + probing +# --------------------------------------------------------------------------- + +_analyses: dict[str, dict] = {} +_analyses_lock = threading.Lock() + + +def start_analysis(vmid: int, source_filename: str) -> dict: + analysis_id = f"analysis_{vmid}_{int(time.time())}" + with _analyses_lock: + _analyses[analysis_id] = { + "vmid": vmid, + "status": "queued", + "message": "Queued", + "result": None, + "error": None, + } + thread = threading.Thread(target=_process_analysis, args=(analysis_id, vmid, source_filename), daemon=True) + thread.start() + return {"analysis_id": analysis_id, "status": "queued", "message": "Analysis queued"} + + +def get_analysis_status(analysis_id: str) -> dict: + with _analyses_lock: + a = _analyses.get(analysis_id) + if not a: + raise KeyError(analysis_id) + return { + "analysis_id": analysis_id, + "vmid": a["vmid"], + "status": a["status"], + "message": a.get("message", ""), + "result": a.get("result"), + "error_details": a.get("error"), + } + + +def _update_analysis(analysis_id: str, status: str, message: str = "", result=None, error: str = None): + with _analyses_lock: + a = _analyses.get(analysis_id) + if not a: + return + a["status"] = status + if message: + a["message"] = message + if result is not None: + a["result"] = result + if error is not None: + a["error"] = error + + +def _process_analysis(analysis_id: str, vmid: int, source_filename: str): + try: + _update_analysis(analysis_id, "processing", "Extracting archive...") + source = extract_if_needed(source_filename) + logger.info("Extraction done: %s", source) + + _update_analysis(analysis_id, "processing", "Extracting nested archives...") + source = _extract_nested(source) + + _update_analysis(analysis_id, "processing", "Discovering disk image...") + disk = discover_disk(source) + logger.info("Disk found: %s (format=%s, size=%.1f GiB)", disk.path, disk.format, disk.size_gb) + + _update_analysis(analysis_id, "processing", "Detecting OS and boot type...") + extract_dir = source if source.is_dir() else source.parent + os_type = detect_os(disk.path, extract_dir) + efi = detect_efi(disk.path) + logger.info("Analysis done: os=%s efi=%s", os_type, efi) + + from models import AnalyzeResponse + result = AnalyzeResponse( + vmid=vmid, + filename=disk.path.name, + disk_format=disk.format, + disk_size_gb=disk.size_gb, + os_type=os_type, + efi_detectable=efi, + ) + _update_analysis(analysis_id, "completed", "Analysis complete", result=result.model_dump()) + + except Exception as exc: + logger.exception("Analysis %s failed", analysis_id) + _update_analysis(analysis_id, "failed", "Analysis failed", error=str(exc)) + + +# --------------------------------------------------------------------------- +# Nested archive extraction — handles split zips, chained archives +# --------------------------------------------------------------------------- + +def _extract_nested(base: Path) -> Path: + """Recursively extract nested archives until disk images are found + or no more archives remain. Handles split zips (.z01-.zNN + .zip).""" + for _ in range(5): + if not base.is_dir(): + return base + + cand = _find_nested_archive(base) + if cand is None: + break + + logger.info("Extracting nested archive: %s", cand) + _extract_one_archive(cand, base) + try: + cand.unlink() + except Exception: + pass + + base = _flatten_nested(base) + return base + + +def _find_nested_archive(base: Path) -> Optional[Path]: + """Find the next archive to extract inside a directory. Prioritises + split zip masters if .z01 segments are present.""" + files = sorted([f for f in base.iterdir() if f.is_file()], key=lambda f: f.name) + + split_master = None + has_segments = False + for f in files: + name = f.name.lower() + if re.search(r'\.z\d{2,3}$', name): + has_segments = True + if name.endswith(".zip") and not re.search(r'\.z\d{2,3}$', name): + split_master = f + + if has_segments and split_master: + return split_master + + for f in files: + if _detect_archive_ext(f.name.lower()) is not None: + return f + return None + + +def _extract_one_archive(archive_path: Path, target_dir: Path): + name = archive_path.name.lower() + ext = _detect_archive_ext(name) + tool_map = { + ".7z": ("7z", ["7z", "x", "-y", str(archive_path)]), + ".zip": ("7z", ["7z", "x", "-y", str(archive_path)]), + ".rar": ("unrar", ["unrar", "x", "-y", str(archive_path)]), + ".tar.gz": ("tar", ["tar", "-xzf", str(archive_path)]), + ".tgz": ("tar", ["tar", "-xzf", str(archive_path)]), + ".tar": ("tar", ["tar", "-xf", str(archive_path)]), + ".gz": ("gunzip", ["gunzip", "-f", str(archive_path)]), + ".bz2": ("bunzip2", ["bunzip2", "-f", str(archive_path)]), + ".xz": ("xz", ["xz", "-d", str(archive_path)]), + } + tool, cmd = tool_map[ext] + if not shutil.which(tool): + raise RuntimeError(f"Extraction tool '{tool}' not found on host") + result = subprocess.run(cmd, cwd=str(target_dir), capture_output=True, text=True, timeout=900) + if result.returncode != 0: + logger.warning("Nested extraction %s failed: %s", archive_path.name, result.stderr.strip()[:200]) diff --git a/backend/models.py b/backend/models.py index 9aa697a..603ee13 100644 --- a/backend/models.py +++ b/backend/models.py @@ -35,6 +35,13 @@ class JobStatus(str, Enum): FAILED = "failed" +class AnalyzeStatus(str, Enum): + QUEUED = "queued" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + + # --------------------------------------------------------------------------- # Request schemas # --------------------------------------------------------------------------- @@ -117,3 +124,12 @@ class CloneRequest(BaseModel): source_vmid: int target_vmid: int target_name: str + + +class AnalyzeStatusResponse(BaseModel): + analysis_id: str + vmid: int + status: AnalyzeStatus + message: str = "" + result: Optional[AnalyzeResponse] = None + error_details: Optional[str] = None diff --git a/frontend/api_client.py b/frontend/api_client.py index f743a4a..612379f 100644 --- a/frontend/api_client.py +++ b/frontend/api_client.py @@ -15,7 +15,6 @@ from typing import Optional BACKEND_URL = os.getenv("BACKEND_URL", "http://10.2.0.2:9000") TIMEOUT = 30 -ANALYZE_TIMEOUT = 300 # guestfish needs time to probe large disk images class ApiError(Exception): @@ -44,12 +43,16 @@ class ApiClient: return self._get("/api/v1/health") def analyze(self, vmid: int, filename: str, source_type: str = "upload") -> dict: - """POST /api/v1/analyze — probe source file for OS, size, format.""" + """POST /api/v1/analyze — start async analysis, returns analysis_id.""" return self._post("/api/v1/analyze", { "vmid": vmid, "source_filename": filename, "source_type": source_type, - }, timeout=ANALYZE_TIMEOUT) + }) + + def get_analysis(self, analysis_id: str) -> dict: + """GET /api/v1/analyze/{analysis_id} — poll analysis status.""" + return self._get(f"/api/v1/analyze/{analysis_id}") def create_job(self, payload: dict) -> dict: """POST /api/v1/jobs — submit conversion job.""" diff --git a/frontend/app.py b/frontend/app.py index 4ab0501..9c2c9ab 100644 --- a/frontend/app.py +++ b/frontend/app.py @@ -299,7 +299,7 @@ async def session_progress(filename: str): }) -@app.post("/session/analyze", response_class=HTMLResponse) +@app.post("/session/analyze") async def session_analyze( request: Request, vmid: int = Form(...), @@ -307,28 +307,63 @@ async def session_analyze( vm_name: str = Form(""), session_id: str = Form(""), ): - """Phase 2 — call the backend /analyze endpoint and render the result.""" + """Phase 2 — start async backend analysis, return analysis_id for polling.""" err = _validate_vmid(vmid) if err: - return render("_analysis.html", request=request, error=err) + return JSONResponse({"phase": "error", "error": err}, status_code=400) - # Prepend session subdirectory to the filename for the backend - # Files are in tmp/{guid}/in/, so the backend needs {guid}/in/filename backend_filename = f"{session_id}/in/{filename}" if session_id else filename try: analysis = api.analyze(vmid=vmid, filename=backend_filename) except ApiError as exc: - return render("_analysis.html", request=request, - error=f"Backend analysis failed: {exc.detail}") + return JSONResponse({"phase": "error", "error": f"Backend analysis failed: {exc.detail}"}, status_code=502) - vm_name = vm_name.strip() - if not vm_name: - vm_name = (analysis.get("os_type") or "vm") + f"-{vmid}" + return JSONResponse({ + "phase": "analyzing", + "analysis_id": analysis.get("analysis_id"), + "vmid": vmid, + "filename": filename, + "vm_name": vm_name.strip(), + }) + + +@app.get("/session/analyze/status/{analysis_id}") +async def session_analyze_status(analysis_id: str): + """Poll backend for async analysis status.""" + try: + return api.get_analysis(analysis_id) + except ApiError as exc: + return JSONResponse({"error": exc.detail}, status_code=502) + + +@app.get("/session/analyze/result/{analysis_id}", response_class=HTMLResponse) +async def session_analyze_result( + request: Request, + analysis_id: str, + vmid: str = "", + source_filename: str = "", + vm_name: str = "", +): + """Render the analysis result after polling completes.""" + try: + status = api.get_analysis(analysis_id) + result = status.get("result") + if not result: + return render("_analysis.html", request=request, + error="Analysis result not ready.") + except ApiError as exc: + return render("_analysis.html", request=request, + error=f"Failed to fetch analysis: {exc.detail}") + + analysis = result + name = vm_name.strip() + if not name: + name = (analysis.get("os_type") or "vm") + f"-{vmid}" return render("_analysis.html", request=request, - vmid=vmid, source_filename=filename, - vm_name=vm_name, analysis=analysis) + vmid=vmid, source_filename=source_filename, + vm_name=name, analysis=analysis) @app.post("/session/confirm", response_class=HTMLResponse) diff --git a/frontend/templates/index.html b/frontend/templates/index.html index c1fcd80..63be7ac 100644 --- a/frontend/templates/index.html +++ b/frontend/templates/index.html @@ -147,10 +147,41 @@ async function runAnalysis(vmid, filename, vmName) { fd.append('vmid', vmid); fd.append('filename', filename); fd.append('vm_name', vmName); fd.append('session_id', window.VM_BENCH_SID || ''); const resp = await fetch('/session/analyze', { method: 'POST', body: fd }); - const html = await resp.text(); - document.getElementById('analysis-section').innerHTML = html; - document.getElementById('session-status').classList.add('pve-hidden'); - initConfirmForm(); + const data = await resp.json(); + + if (data.phase === 'error') { + setPhase('Analysis failed', 0, false); + showError(data.error); + return; + } + + var analysisId = data.analysis_id; + var startTime = Date.now(); + + for (;;) { + await sleep(2000); + try { + var pr = await fetch('/session/analyze/status/' + analysisId); + var sdata = await pr.json(); + var elapsed = Math.round((Date.now() - startTime) / 1000); + setPhase('Analysing... ' + (sdata.message || '') + ' (' + elapsed + 's)', sdata.status === 'completed' ? 100 : 50, sdata.status !== 'completed'); + + if (sdata.status === 'completed') { + await sleep(500); + var rr = await fetch('/session/analyze/result/' + analysisId + '?source_filename=' + encodeURIComponent(filename) + '&vm_name=' + encodeURIComponent(vmName) + '&vmid=' + vmid); + var html = await rr.text(); + document.getElementById('analysis-section').innerHTML = html; + document.getElementById('session-status').classList.add('pve-hidden'); + initConfirmForm(); + return; + } + if (sdata.status === 'failed') { + setPhase('Analysis failed', 0, false); + showError(sdata.error_details || 'Analysis failed'); + return; + } + } catch (_) {} + } } function initConfirmForm() { diff --git a/frontend/templates/scp.html b/frontend/templates/scp.html index 741db34..dcf7b9e 100644 --- a/frontend/templates/scp.html +++ b/frontend/templates/scp.html @@ -106,15 +106,42 @@ async function startScpPull(e) { setScpPhase('Pull complete (' + (pdata.file_size_gb || 0) + ' GiB)', 100, false); await new Promise(r => setTimeout(r, 500)); setScpPhase('Step 2/2: Analysing source image...', 0, true); - const fd = new FormData(); + var fd = new FormData(); fd.append('vmid', document.getElementById('scp-vmid').value); fd.append('filename', filename); fd.append('vm_name', document.getElementById('scp-vm-name').value.trim()); fd.append('session_id', SESSION_ID); - const ar = await fetch('/session/analyze', { method: 'POST', body: fd }); - document.getElementById('scp-analysis-section').innerHTML = await ar.text(); - document.getElementById('scp-status').classList.add('pve-hidden'); - btn.disabled = false; return; + var ar = await fetch('/session/analyze', { method: 'POST', body: fd }); + var adata = await ar.json(); + if (adata.phase === 'error') { + setScpPhase('Analysis failed', 0, false); + showScpError(adata.error); + btn.disabled = false; return; + } + var analysisId = adata.analysis_id; + var vmid = document.getElementById('scp-vmid').value; + var vmName = document.getElementById('scp-vm-name').value.trim(); + var astart = Date.now(); + for (;;) { + await new Promise(r => setTimeout(r, 2000)); + var sr = await fetch('/session/analyze/status/' + analysisId); + var sdata = await sr.json(); + var elapsed = Math.round((Date.now() - astart) / 1000); + setScpPhase('Analysing... ' + (sdata.message || '') + ' (' + elapsed + 's)', sdata.status === 'completed' ? 100 : 50, sdata.status !== 'completed'); + if (sdata.status === 'completed') { + await new Promise(r => setTimeout(r, 500)); + var rr = await fetch('/session/analyze/result/' + analysisId + '?source_filename=' + encodeURIComponent(filename) + '&vm_name=' + encodeURIComponent(vmName) + '&vmid=' + vmid); + document.getElementById('scp-analysis-section').innerHTML = await rr.text(); + document.getElementById('scp-status').classList.add('pve-hidden'); + scpInitConfirmForm(); + btn.disabled = false; return; + } + if (sdata.status === 'failed') { + setScpPhase('Analysis failed', 0, false); + showScpError(sdata.error_details || 'Analysis failed'); + btn.disabled = false; return; + } + } } if (pdata.phase === 'error') { setScpPhase('SCP pull failed', 0, false); @@ -147,6 +174,68 @@ function showScpError(msg) { const el = document.getElementById('scp-error'); if (el) el.innerHTML += '
' + msg + '
'; } + +function scpInitConfirmForm() { + const form = document.getElementById('confirm-form'); + if (!form) return; + form.addEventListener('submit', scpSubmitJob); + const bootSel = form.querySelector('select[name="auto_detect_boot"]'); + if (bootSel) { + bootSel.addEventListener('change', function() { + document.getElementById('boot-type-group').style.display = + this.value === 'false' ? 'block' : 'none'; + }); + } +} + +async function scpSubmitJob(e) { + e.preventDefault(); + const form = e.target; + const btn = form.querySelector('#submit-btn'); + btn.disabled = true; + const formData = new FormData(form); + formData.append('session_id', SESSION_ID); + try { + const resp = await fetch('/session/confirm', { method: 'POST', body: formData }); + const html = await resp.text(); + document.getElementById('scp-form-panel').style.display = 'none'; + document.getElementById('scp-analysis-section').innerHTML = ''; + document.getElementById('scp-analysis-section').innerHTML = html; + scpInitPolling(); + } catch (err) { + btn.disabled = false; + } +} + +var _scpPollTimer = null; +function scpInitPolling() { + const container = document.getElementById('polling-container'); + if (!container) return; + const jobId = container.dataset.jobId; + const startTime = Date.now(); + var completed = false; + + async function poll() { + try { + const resp = await fetch('/session/status/' + jobId); + const data = await resp.json(); + const pct = data.progress_percentage || 0; + var fill = document.getElementById('pve-progress-bar'); + if (fill) { fill.style.width = pct + '%'; fill.textContent = pct + '%'; } + var elMsg = document.getElementById('status-message'); + if (elMsg) elMsg.textContent = data.message || ''; + var badge = document.getElementById('status-badge'); + if (badge) { + if (data.status === 'completed') { badge.className = 'pve-badge pve-badge-completed'; badge.textContent = 'Completed'; } + else if (data.status === 'failed') { badge.className = 'pve-badge pve-badge-failed'; badge.textContent = 'Failed'; } + else { badge.className = 'pve-badge pve-badge-running'; badge.innerHTML = ' Processing'; } + } + if (data.status === 'completed' || data.status === 'failed') { completed = true; return; } + } catch (err) {} + if (!completed) _scpPollTimer = setTimeout(poll, 2000); + } + _scpPollTimer = setTimeout(poll, 3000); +} {% endblock %}