Codex
218 lines
6.6 KiB
Python
218 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass
|
|
class CommandResult:
|
|
stdout: str
|
|
stderr: str
|
|
|
|
|
|
class CommandError(RuntimeError):
|
|
def __init__(self, command: list[str], result: subprocess.CompletedProcess[str]):
|
|
super().__init__(
|
|
f"Command failed ({result.returncode}): {' '.join(command)}\n{result.stderr.strip()}"
|
|
)
|
|
self.command = command
|
|
self.returncode = result.returncode
|
|
self.stdout = result.stdout
|
|
self.stderr = result.stderr
|
|
|
|
|
|
def run(command: list[str], timeout: int | None = None) -> CommandResult:
|
|
result = subprocess.run(
|
|
command,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise CommandError(command, result)
|
|
return CommandResult(stdout=result.stdout, stderr=result.stderr)
|
|
|
|
|
|
def find_executable(name: str) -> str:
|
|
return shutil.which(name) or ""
|
|
|
|
|
|
def pvesh_json(args: list[str], timeout: int | None = 120):
|
|
result = run(["pvesh", *args, "--output-format", "json"], timeout=timeout)
|
|
text = result.stdout.strip()
|
|
return json.loads(text) if text else None
|
|
|
|
|
|
def _extract_upid(text: str):
|
|
text = text.strip()
|
|
if not text:
|
|
return None
|
|
if text.startswith("UPID:"):
|
|
return text
|
|
match = re.search(r"UPID:[^\s\"']+", text)
|
|
if match:
|
|
return match.group(0)
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
if isinstance(data, str) and data.startswith("UPID:"):
|
|
return data
|
|
if isinstance(data, dict):
|
|
for key in ("upid", "data"):
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value.startswith("UPID:"):
|
|
return value
|
|
return None
|
|
|
|
|
|
def list_guests() -> list[dict]:
|
|
resources = pvesh_json(["get", "/cluster/resources", "--type", "vm"]) or []
|
|
guests: list[dict] = []
|
|
for item in resources:
|
|
raw_type = item.get("type")
|
|
guest_type = "lxc" if raw_type == "lxc" else "vm"
|
|
guests.append(
|
|
{
|
|
"vmid": int(item["vmid"]),
|
|
"name": item.get("name") or str(item["vmid"]),
|
|
"type": guest_type,
|
|
"node": item.get("node") or "",
|
|
"status": item.get("status") or "unknown",
|
|
}
|
|
)
|
|
return sorted(guests, key=lambda row: (row["node"], row["vmid"]))
|
|
|
|
|
|
def list_storages(node: str | None = None) -> list[str]:
|
|
path = f"/nodes/{node}/storage" if node else "/storage"
|
|
data = pvesh_json(["get", path]) or []
|
|
return sorted({item.get("storage") for item in data if item.get("storage")})
|
|
|
|
|
|
def start_proxmox_backup(
|
|
*,
|
|
node: str,
|
|
vmid: int,
|
|
storage: str,
|
|
mode: str,
|
|
compression: str,
|
|
) -> str:
|
|
command = [
|
|
"pvesh",
|
|
"create",
|
|
f"/nodes/{node}/vzdump",
|
|
"--vmid",
|
|
str(vmid),
|
|
"--storage",
|
|
storage,
|
|
"--mode",
|
|
mode,
|
|
"--compress",
|
|
compression,
|
|
"--output-format",
|
|
"json",
|
|
]
|
|
# pvesh create for vzdump blocks and streams the task log until the backup
|
|
# finishes rather than returning a UPID immediately, so this must not have
|
|
# a fixed timeout: it can legitimately run as long as the guest's backup
|
|
# takes (a prior 300s timeout was killing real, in-progress vzdump jobs).
|
|
result = run(command, timeout=None)
|
|
upid = _extract_upid(result.stdout)
|
|
if not upid:
|
|
upid = _extract_upid(result.stderr)
|
|
if upid:
|
|
return upid
|
|
|
|
# Some pvesh versions ignore --output-format for create actions and return
|
|
# plain task text. Include both streams so failures are diagnosable without
|
|
# assuming JSON.
|
|
raise RuntimeError(
|
|
"Could not read UPID from Proxmox response: "
|
|
f"stdout={result.stdout.strip()!r} stderr={result.stderr.strip()!r}"
|
|
)
|
|
|
|
|
|
def get_task_status(node: str, upid: str) -> dict:
|
|
return pvesh_json(["get", f"/nodes/{node}/tasks/{upid}/status"]) or {}
|
|
|
|
|
|
def _archive_name_timestamp(path: Path) -> datetime:
|
|
match = re.search(r"-(\d{4}_\d{2}_\d{2}-\d{2}_\d{2}_\d{2})", path.name)
|
|
if not match:
|
|
return datetime.min
|
|
try:
|
|
return datetime.strptime(match.group(1), "%Y_%m_%d-%H_%M_%S")
|
|
except ValueError:
|
|
return datetime.min
|
|
|
|
|
|
def discover_archive(local_backup_dir: str, vmid: int, guest_type: str, started_at: str) -> Path:
|
|
base = Path(local_backup_dir)
|
|
prefix = "vzdump-lxc" if guest_type == "lxc" else "vzdump-qemu"
|
|
archive_suffixes = (
|
|
".tar",
|
|
".tar.gz",
|
|
".tar.lzo",
|
|
".tar.zst",
|
|
".vma",
|
|
".vma.gz",
|
|
".vma.lzo",
|
|
".vma.zst",
|
|
)
|
|
started = datetime.fromisoformat(started_at)
|
|
candidates = []
|
|
for path in base.glob(f"{prefix}-{vmid}-*"):
|
|
if path.is_file() and path.name.endswith(archive_suffixes):
|
|
mtime = datetime.fromtimestamp(path.stat().st_mtime, tz=started.tzinfo)
|
|
if mtime >= started:
|
|
candidates.append(path)
|
|
if not candidates:
|
|
raise FileNotFoundError(f"No backup archive found for VMID {vmid} in {base}")
|
|
return max(candidates, key=lambda p: (p.stat().st_mtime_ns, _archive_name_timestamp(p), p.name))
|
|
|
|
|
|
def rclone_list_remotes(rclone_path: str) -> list[str]:
|
|
result = run([rclone_path, "listremotes"], timeout=60)
|
|
return [line.rstrip(":") for line in result.stdout.splitlines() if line.strip()]
|
|
|
|
|
|
def rclone_remote_type(rclone_path: str, remote_name: str) -> str:
|
|
remote = remote_name.rstrip(":")
|
|
result = run([rclone_path, "config", "show", remote])
|
|
for line in result.stdout.splitlines():
|
|
if line.strip().startswith("type"):
|
|
_, value = line.split("=", 1)
|
|
return value.strip()
|
|
return ""
|
|
|
|
|
|
def rclone_copyto(rclone_path: str, local_path: str, remote_object: str) -> None:
|
|
run([rclone_path, "copyto", local_path, remote_object], timeout=None)
|
|
|
|
|
|
def rclone_download(rclone_path: str, remote_object: str, local_path: str) -> None:
|
|
run([rclone_path, "copyto", remote_object, local_path], timeout=None)
|
|
|
|
|
|
def rclone_object_exists(rclone_path: str, remote_object: str) -> bool:
|
|
try:
|
|
run([rclone_path, "lsjson", remote_object], timeout=120)
|
|
return True
|
|
except CommandError:
|
|
return False
|
|
|
|
|
|
def rclone_delete_file(rclone_path: str, remote_object: str) -> None:
|
|
run([rclone_path, "deletefile", remote_object], timeout=300)
|
|
|
|
|
|
def rclone_rmdir(rclone_path: str, remote_dir: str) -> None:
|
|
run([rclone_path, "rmdir", remote_dir], timeout=120)
|