285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""Ordered VM lifecycle operations and compensation for partial failures."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import time
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .config import Settings
|
|
from .domain import VmRecord, VmSpec, new_vm_id
|
|
from .errors import TapCreationError, UvmError
|
|
from .firecracker.api import FirecrackerClient
|
|
from .firecracker.config import build_config, write_config
|
|
from .firecracker.process import FirecrackerProcessManager
|
|
from .images import ImageStore
|
|
from .integrity import load_manifest, verify_file
|
|
from .network import NetworkManager
|
|
from .state import StateStore
|
|
from .system import check_kvm, ensure_data_directories, require_root
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ListedVm:
|
|
vm: VmRecord
|
|
observed_status: str
|
|
|
|
|
|
class LifecycleService:
|
|
"""The only module allowed to coordinate multiple VM host resources."""
|
|
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
state_store: StateStore,
|
|
images: ImageStore,
|
|
network: NetworkManager,
|
|
process: FirecrackerProcessManager,
|
|
*,
|
|
client_factory: Callable[[Path, float], FirecrackerClient] = FirecrackerClient,
|
|
clock: Callable[[], float] = time.time,
|
|
) -> None:
|
|
self._settings = settings
|
|
self._state_store = state_store
|
|
self._images = images
|
|
self._network = network
|
|
self._process = process
|
|
self._client_factory = client_factory
|
|
self._clock = clock
|
|
|
|
def create(self, spec: VmSpec) -> VmRecord:
|
|
"""Reserve identity first, then create all VM resources in dependency order."""
|
|
|
|
require_root()
|
|
ensure_data_directories(self._settings)
|
|
check_kvm()
|
|
if not self._settings.firecracker_binary.exists():
|
|
raise UvmError("Firecracker is not installed. Run: sudo uvm install")
|
|
with self._state_store.operation_lock():
|
|
manifest = load_manifest(self._settings.integrity_manifest_path)
|
|
firecracker_checksum = self._settings.firecracker_binary_sha256
|
|
if manifest.verified or self._settings.allow_unverified_downloads:
|
|
firecracker_checksum = firecracker_checksum or manifest.checksums.get("firecracker")
|
|
verify_file(
|
|
self._settings.firecracker_binary,
|
|
firecracker_checksum,
|
|
"Firecracker binary",
|
|
"UVM_FIRECRACKER_BINARY_SHA256",
|
|
allow_unverified=self._settings.allow_unverified_downloads,
|
|
)
|
|
return self._create_locked(spec)
|
|
|
|
def _create_locked(self, spec: VmSpec) -> VmRecord:
|
|
assets = self._images.installed_assets()
|
|
vm = self._reserve_vm(spec)
|
|
|
|
tap_created = False
|
|
try:
|
|
runtime_dir = self._settings.vm_dir(vm.id)
|
|
runtime_dir.mkdir(parents=True, exist_ok=False)
|
|
runtime_dir.chmod(0o700)
|
|
disk = self._images.create_vm_disk(assets.rootfs, Path(vm.disk))
|
|
self._images.provision_credentials(disk, vm.username, vm.password)
|
|
self._network.ensure_bridge()
|
|
try:
|
|
self._network.create_tap(vm.tap)
|
|
except TapCreationError as error:
|
|
tap_created = error.tap_created
|
|
raise
|
|
else:
|
|
tap_created = True
|
|
|
|
config = build_config(self._settings, vm, assets.kernel, disk)
|
|
write_config(Path(vm.config), config)
|
|
|
|
process_info = self._process.start(vm)
|
|
vm.pid = process_info.pid
|
|
vm.process_start_time = process_info.start_time
|
|
vm.updated_at = self._now()
|
|
self._replace_vm(vm)
|
|
|
|
client = self._client_factory(Path(vm.socket), self._settings.api_timeout_s)
|
|
client.configure_and_start(config)
|
|
except BaseException as error:
|
|
cleanup_errors = self._rollback_create(vm, tap_created)
|
|
message = f"failed to create {vm.id}: {error}"
|
|
if cleanup_errors:
|
|
message = f"{message}; cleanup failed: {'; '.join(cleanup_errors)}"
|
|
vm.status = "failed"
|
|
vm.last_error = message
|
|
vm.updated_at = self._now()
|
|
self._replace_vm_if_present(vm)
|
|
else:
|
|
self._remove_vm_if_present(vm.id)
|
|
if isinstance(error, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
if isinstance(error, UvmError):
|
|
raise error
|
|
raise UvmError(message) from error
|
|
|
|
vm.status = "running"
|
|
vm.last_error = None
|
|
vm.updated_at = self._now()
|
|
self._replace_vm(vm)
|
|
return vm
|
|
|
|
def list_vms(self) -> list[ListedVm]:
|
|
state = self._state_store.load()
|
|
listed: list[ListedVm] = []
|
|
for vm in state.vms.values():
|
|
observed_status = vm.status
|
|
if vm.status in {"starting", "running", "stopping", "terminating"}:
|
|
if not self._process.is_alive(vm):
|
|
observed_status = "dead"
|
|
listed.append(ListedVm(vm=vm, observed_status=observed_status))
|
|
return listed
|
|
|
|
def find_for_ssh(self, identifier: str) -> VmRecord:
|
|
return self._state_store.find_by_id_or_ip(identifier)
|
|
|
|
def stop(self, vm_id: str) -> VmRecord:
|
|
"""Stop one VM while retaining its private writable disk for future use."""
|
|
|
|
require_root()
|
|
with self._state_store.operation_lock():
|
|
return self._stop_locked(vm_id)
|
|
|
|
def _stop_locked(self, vm_id: str) -> VmRecord:
|
|
vm = self._begin_transition(vm_id, "stopping")
|
|
try:
|
|
self._process.terminate(vm)
|
|
self._network.delete_tap(vm.tap)
|
|
except BaseException as error:
|
|
vm.status = "failed"
|
|
vm.last_error = f"failed to stop VM: {error}"
|
|
vm.updated_at = self._now()
|
|
self._replace_vm_if_present(vm)
|
|
if isinstance(error, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
if isinstance(error, UvmError):
|
|
raise error
|
|
raise UvmError(vm.last_error) from error
|
|
|
|
vm.status = "stopped"
|
|
vm.pid = None
|
|
vm.process_start_time = None
|
|
vm.last_error = None
|
|
vm.updated_at = self._now()
|
|
self._replace_vm(vm)
|
|
return vm
|
|
|
|
def destroy(self, vm_id: str) -> VmRecord:
|
|
"""Stop a VM, remove its private resources, then release its allocation."""
|
|
|
|
require_root()
|
|
with self._state_store.operation_lock():
|
|
return self._destroy_locked(vm_id)
|
|
|
|
def _destroy_locked(self, vm_id: str) -> VmRecord:
|
|
vm = self._begin_transition(vm_id, "terminating")
|
|
try:
|
|
self._process.terminate(vm)
|
|
self._network.delete_tap(vm.tap)
|
|
try:
|
|
shutil.rmtree(self._settings.vm_dir(vm.id))
|
|
except FileNotFoundError:
|
|
pass
|
|
except BaseException as error:
|
|
vm.status = "failed"
|
|
vm.last_error = f"failed to destroy VM: {error}"
|
|
vm.updated_at = self._now()
|
|
self._replace_vm_if_present(vm)
|
|
if isinstance(error, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
if isinstance(error, UvmError):
|
|
raise error
|
|
raise UvmError(vm.last_error) from error
|
|
|
|
self._remove_vm_if_present(vm.id)
|
|
return vm
|
|
|
|
def _reserve_vm(self, spec: VmSpec) -> VmRecord:
|
|
with self._state_store.transaction() as state:
|
|
vm_id = new_vm_id()
|
|
used_taps = {existing_vm.tap for existing_vm in state.vms.values()}
|
|
while vm_id in state.vms or self._network.tap_name(vm_id) in used_taps:
|
|
vm_id = new_vm_id()
|
|
guest_ip = self._network.allocate_ip(state.vms.values(), spec.guest_ip)
|
|
mac = self._network.mac_for(state.next_mac_index)
|
|
state.next_mac_index += 1
|
|
now = self._now()
|
|
runtime_dir = self._settings.vm_dir(vm_id)
|
|
vm = VmRecord(
|
|
id=vm_id,
|
|
cpu=spec.cpu,
|
|
ram_mib=spec.ram_mib,
|
|
guest_ip=str(guest_ip),
|
|
gateway=str(self._settings.gateway),
|
|
tap=self._network.tap_name(vm_id),
|
|
mac=mac,
|
|
socket=str(runtime_dir / "firecracker.sock"),
|
|
config=str(runtime_dir / "config.json"),
|
|
log=str(runtime_dir / "firecracker.log"),
|
|
disk=str(runtime_dir / "rootfs.ext4"),
|
|
username=spec.username,
|
|
password=spec.password,
|
|
status="starting",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
state.vms[vm.id] = vm
|
|
return vm
|
|
|
|
def _begin_transition(self, vm_id: str, target_status: str) -> VmRecord:
|
|
with self._state_store.transaction() as state:
|
|
try:
|
|
vm = state.vms[vm_id]
|
|
except KeyError as error:
|
|
raise UvmError(f"VM not found: {vm_id}") from error
|
|
if vm.status in {"stopping", "terminating"}:
|
|
raise UvmError(f"VM operation is already in progress: {vm_id}")
|
|
vm.status = target_status
|
|
vm.updated_at = self._now()
|
|
state.vms[vm_id] = vm
|
|
return vm
|
|
|
|
def _replace_vm(self, vm: VmRecord) -> None:
|
|
with self._state_store.transaction() as state:
|
|
if vm.id not in state.vms:
|
|
raise UvmError(f"VM not found: {vm.id}")
|
|
state.vms[vm.id] = vm
|
|
|
|
def _replace_vm_if_present(self, vm: VmRecord) -> None:
|
|
with self._state_store.transaction() as state:
|
|
if vm.id in state.vms:
|
|
state.vms[vm.id] = vm
|
|
|
|
def _remove_vm_if_present(self, vm_id: str) -> None:
|
|
with self._state_store.transaction() as state:
|
|
state.vms.pop(vm_id, None)
|
|
|
|
def _rollback_create(self, vm: VmRecord, tap_created: bool) -> list[str]:
|
|
errors: list[str] = []
|
|
if vm.pid is not None:
|
|
try:
|
|
self._process.terminate(vm)
|
|
except Exception as error:
|
|
errors.append(f"process: {error}")
|
|
if tap_created:
|
|
try:
|
|
self._network.delete_tap(vm.tap)
|
|
except Exception as error:
|
|
errors.append(f"network: {error}")
|
|
try:
|
|
shutil.rmtree(self._settings.vm_dir(vm.id))
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError as error:
|
|
errors.append(f"files: {error}")
|
|
return errors
|
|
|
|
def _now(self) -> int:
|
|
return int(self._clock())
|