"""Incus VM lifecycle management.
These operations manage the VM itself (snapshot, restore, list, status)
and stay as `incus` CLI commands — they're inherently incus-specific.
Command execution and file operations have moved to transport.py (SSH).
"""
from __future__ import annotations
import asyncio
import json
import re
from dataclasses import dataclass, field
# ---------------------------------------------------------------------------
# Result types
# ---------------------------------------------------------------------------
@dataclass
class ExecResult:
"""Result of running a CLI command."""
stdout: str
stderr: str
exit_code: int
@dataclass
class VMInfo:
"""Status information for a single VM/container."""
name: str
status: str # Running, Stopped, etc.
type: str # container or virtual-machine
ipv4: str = ""
ipv6: str = ""
architecture: str = ""
pid: int = 0
processes: int = 0
memory_usage: int = 0
cpu_usage: int = 0
snapshots: list[str] = field(default_factory=list)
# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------
_VALID_NAME = re.compile(r"^[a-zA-Z][a-zA-Z0-9\-]*$")
def _validate_vm_name(name: str) -> None:
"""Validate that a VM/container name is safe."""
if not name or not name.strip():
raise ValueError("VM name cannot be empty")
if not _VALID_NAME.match(name):
raise ValueError(
f"Invalid VM name '{name}': must start with a letter, "
"contain only alphanumeric characters and hyphens"
)
def _validate_snapshot_name(name: str) -> None:
"""Validate a snapshot name."""
if not name or not name.strip():
raise ValueError("Snapshot name cannot be empty")
if not re.match(r"^[a-zA-Z0-9_\-]+$", name):
raise ValueError(
f"Invalid snapshot name '{name}': "
"only alphanumeric, hyphens, and underscores allowed"
)
# ---------------------------------------------------------------------------
# Core CLI runner
# ---------------------------------------------------------------------------
async def _run_incus(
*args: str,
timeout: int = 120,
) -> ExecResult:
"""Run an incus CLI command and capture output.
Args:
*args: Arguments to pass to `incus`.
timeout: Maximum seconds to wait.
Returns:
ExecResult with stdout, stderr, and exit code.
"""
proc = await asyncio.create_subprocess_exec(
"incus", *args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
stdin=asyncio.subprocess.DEVNULL,
)
try:
stdout_bytes, stderr_bytes = await asyncio.wait_for(
proc.communicate(),
timeout=timeout,
)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise TimeoutError(
f"incus command timed out after {timeout}s: incus {' '.join(args)}"
)
return ExecResult(
stdout=stdout_bytes.decode("utf-8", errors="replace"),
stderr=stderr_bytes.decode("utf-8", errors="replace"),
exit_code=proc.returncode or 0,
)
# ---------------------------------------------------------------------------
# VM lifecycle operations (incus-specific)
# ---------------------------------------------------------------------------
async def vm_snapshot(vm: str, name: str) -> ExecResult:
"""Create a snapshot of a VM/container.
Uses `incus snapshot create <vm> <name>`.
"""
_validate_vm_name(vm)
_validate_snapshot_name(name)
return await _run_incus("snapshot", "create", vm, name)
async def vm_restore(vm: str, snapshot: str) -> ExecResult:
"""Restore a VM/container to a named snapshot.
Uses `incus snapshot restore <vm> <snapshot>`.
"""
_validate_vm_name(vm)
_validate_snapshot_name(snapshot)
return await _run_incus("snapshot", "restore", vm, snapshot)
async def vm_status(vm: str) -> VMInfo:
"""Get detailed status of a VM/container.
Uses `incus list <vm> --format json` for structured data.
"""
_validate_vm_name(vm)
result = await _run_incus("list", vm, "--format", "json")
if result.exit_code != 0:
raise RuntimeError(
f"Failed to get status for {vm}: {result.stderr.strip()}"
)
try:
instances = json.loads(result.stdout)
except json.JSONDecodeError as e:
raise RuntimeError(f"Failed to parse incus output: {e}")
if not instances:
raise RuntimeError(f"VM/container '{vm}' not found")
# Find exact match (incus list does prefix matching)
instance = None
for inst in instances:
if inst.get("name") == vm:
instance = inst
break
if instance is None:
raise RuntimeError(f"VM/container '{vm}' not found")
# Extract network info
ipv4 = ""
ipv6 = ""
state = instance.get("state", {})
networks = state.get("network", {})
for iface_name, iface in networks.items():
if iface_name == "lo":
continue
for addr in iface.get("addresses", []):
if addr.get("family") == "inet" and addr.get("scope") == "global":
ipv4 = ipv4 or addr.get("address", "")
elif addr.get("family") == "inet6" and addr.get("scope") == "global":
ipv6 = ipv6 or addr.get("address", "")
# Extract snapshots
snapshot_names = []
for snap in instance.get("snapshots", []) or []:
if isinstance(snap, dict):
snapshot_names.append(snap.get("name", ""))
elif isinstance(snap, str):
snapshot_names.append(snap)
return VMInfo(
name=instance.get("name", vm),
status=instance.get("status", "Unknown"),
type=instance.get("type", "unknown"),
architecture=instance.get("architecture", ""),
ipv4=ipv4,
ipv6=ipv6,
pid=state.get("pid", 0),
processes=state.get("processes", 0),
memory_usage=state.get("memory", {}).get("usage", 0),
cpu_usage=state.get("cpu", {}).get("usage", 0),
snapshots=snapshot_names,
)
async def vm_list() -> list[VMInfo]:
"""List all VMs/containers with status.
Uses `incus list --format json`.
"""
result = await _run_incus("list", "--format", "json")
if result.exit_code != 0:
raise RuntimeError(
f"Failed to list VMs: {result.stderr.strip()}"
)
try:
instances = json.loads(result.stdout)
except json.JSONDecodeError as e:
raise RuntimeError(f"Failed to parse incus output: {e}")
vms: list[VMInfo] = []
for inst in instances:
ipv4 = ""
state = inst.get("state", {})
networks = state.get("network", {})
for iface_name, iface in networks.items():
if iface_name == "lo":
continue
for addr in iface.get("addresses", []):
if addr.get("family") == "inet" and addr.get("scope") == "global":
ipv4 = ipv4 or addr.get("address", "")
vms.append(VMInfo(
name=inst.get("name", ""),
status=inst.get("status", "Unknown"),
type=inst.get("type", "unknown"),
ipv4=ipv4,
))
return vms
