yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan_cli: Added lazy qmp

Browse Source
This commit is contained in:
Qubasa
2024-02-13 16:43:40 +07:00
parent e17fa83dad
commit 6d06742c49
3 changed files with 36 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
pkgs/clan-cli/clan_cli/machines/machines.py
View File

@@ -2,10 +2,10 @@ import json
import logging
from collections.abc import Generator
from contextlib import contextmanager
from os import path
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
from tempfile import NamedTemporaryFile
from clan_cli.dirs import vm_state_dir
from qemu.qmp import QEMUMonitorProtocol
from ..cmd import run
@@ -17,37 +17,25 @@ log = logging.getLogger(__name__)
class VMAttr:
def __init__(self, machine_name: str) -> None:
self.temp_dir = TemporaryDirectory(prefix="clan_vm-", suffix=f"-{machine_name}")
self._qmp_socket: Path = Path(self.temp_dir.name) / "qmp.sock"
self._qga_socket: Path = Path(self.temp_dir.name) / "qga.sock"
def __init__(self, state_dir: Path) -> None:
self._qmp_socket: Path = state_dir / "qmp.sock"
self._qga_socket: Path = state_dir / "qga.sock"
self._qmp: QEMUMonitorProtocol | None = None
@contextmanager
def qmp(self) -> Generator[QEMUMonitorProtocol, None, None]:
if self._qmp is None:
log.debug(f"qmp_socket: {self._qmp_socket}")
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
rpath = self._qmp_socket.resolve()
if not rpath.exists():
raise ClanError(f"qmp socket {rpath} does not exist")
self._qmp = QEMUMonitorProtocol(str(rpath))
self._qmp.connect()
try:
yield self._qmp
finally:
self._qmp.close()
@property
def qmp_socket(self) -> Path:
if self._qmp is None:
log.debug(f"qmp_socket: {self._qmp_socket}")
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
return self._qmp_socket
@property
def qga_socket(self) -> Path:
if self._qmp is None:
log.debug(f"qmp_socket: {self.qga_socket}")
self._qmp = QEMUMonitorProtocol(path.realpath(self._qmp_socket))
return self._qga_socket
class Machine:
def __init__(
@@ -70,7 +58,9 @@ class Machine:
self._deployment_info: None | dict[str, str] = deployment_info
self.vm: VMAttr = VMAttr(name)
state_dir = vm_state_dir(flake_url=str(self.flake), vm_name=self.name)
self.vm: VMAttr = VMAttr(state_dir)
def __str__(self) -> str:
return f"Machine(name={self.name}, flake={self.flake})"