kstyagi
3e17e3e72f
- Added new API endpoints for tool catalog and installation management. - Updated usage documentation to include new tool installation features. - Implemented asynchronous tool installation with job tracking. - Introduced integration tests for tool installation in disposable VMs. - Enhanced dashboard UI to reflect tool installation status and usage instructions. - Improved error handling and logging for tool installation processes. - Added support for multiple Python versions and Conda environments. - Refactored tool management logic to ensure better state handling and job persistence.
198 lines
9.0 KiB
Python
198 lines
9.0 KiB
Python
"""Atomic per-VM tool job records, serialized with lifecycle filesystem changes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
|
|
from .config import Settings
|
|
from .errors import ConflictError, NotFoundError, StateError
|
|
from .state import StateStore
|
|
from .tool_catalog import normalize_tools
|
|
|
|
|
|
ACTIVE_JOB_STATUSES = frozenset({"queued", "starting", "waiting", "installing"})
|
|
MAX_RETAINED_JOBS = 10
|
|
MAX_DOCUMENT_BYTES = 16 * 1024 * 1024
|
|
|
|
|
|
class ToolJobStore:
|
|
"""Only short transactions hold operations.lock; guest execution never does."""
|
|
|
|
def __init__(self, settings: Settings, state_store: StateStore, *, clock=time.time) -> None:
|
|
self.settings = settings
|
|
self.state = state_store
|
|
self.clock = clock
|
|
|
|
def _path(self, vm_id):
|
|
vm = self.state.get(vm_id) # Exact ID only, including while creating a VM.
|
|
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm.id):
|
|
raise StateError("invalid VM ID in tool history")
|
|
directory = self.settings.vm_dir(vm.id)
|
|
try:
|
|
metadata = directory.lstat()
|
|
except FileNotFoundError as error:
|
|
raise NotFoundError("VM runtime directory is no longer available") from error
|
|
if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.geteuid()
|
|
or metadata.st_mode & 0o077 or not vm.runtime_owned):
|
|
raise StateError("tool history requires a private, owned VM runtime directory")
|
|
return directory / "tools.json"
|
|
|
|
@staticmethod
|
|
def _empty():
|
|
return {"schema_version": 1, "jobs": [], "inventory": [], "checked_at": None,
|
|
"agent_ready": False, "guest_busy": False, "supported": None, "error": None}
|
|
|
|
def _read(self, vm_id):
|
|
path = self._path(vm_id)
|
|
try:
|
|
descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
except FileNotFoundError:
|
|
return self._empty()
|
|
except OSError as error:
|
|
raise StateError("could not open private tool installation history") from error
|
|
try:
|
|
with os.fdopen(descriptor, "rb") as stream:
|
|
metadata = os.fstat(stream.fileno())
|
|
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
|
|
or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o077):
|
|
raise StateError("unsafe tool history file")
|
|
raw = stream.read(MAX_DOCUMENT_BYTES + 1)
|
|
if len(raw) > MAX_DOCUMENT_BYTES:
|
|
raise StateError("tool history exceeds its size limit")
|
|
data = json.loads(raw)
|
|
if (not isinstance(data, dict) or data.get("schema_version") != 1
|
|
or not isinstance(data.get("jobs"), list) or not isinstance(data.get("inventory"), list)
|
|
or len(data["jobs"]) > MAX_RETAINED_JOBS):
|
|
raise StateError("invalid tool history document")
|
|
for job in data["jobs"]:
|
|
if (not isinstance(job, dict) or job.get("vm_id") != vm_id
|
|
or not re.fullmatch(r"tools-[0-9a-f]{32}", str(job.get("id", "")))
|
|
or not isinstance(job.get("items"), list) or len(job["items"]) > 12):
|
|
raise StateError("invalid tool installation job")
|
|
return data
|
|
except (OSError, ValueError) as error:
|
|
raise StateError("could not read tool installation history") from error
|
|
|
|
def _write(self, vm_id, data):
|
|
path = self._path(vm_id) # Never recreate a deleted VM directory.
|
|
descriptor, name = tempfile.mkstemp(prefix=".tools-", suffix=".json", dir=path.parent)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
|
os.fchmod(stream.fileno(), 0o600)
|
|
json.dump(data, stream, ensure_ascii=True, allow_nan=False)
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(name, path)
|
|
self.state._fsync_parent(path)
|
|
except OSError as error:
|
|
raise StateError("could not save tool installation history") from error
|
|
finally:
|
|
try:
|
|
os.unlink(name)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
@contextmanager
|
|
def transaction(self, vm_id):
|
|
with self.state.operation_lock():
|
|
data = self._read(vm_id)
|
|
yield data
|
|
self._write(vm_id, data)
|
|
|
|
def read(self, vm_id):
|
|
with self.state.operation_lock():
|
|
return self._read(vm_id)
|
|
|
|
def enqueue_locked(self, vm_id, selections, *, start_if_stopped=False, restart=False,
|
|
request_id=None, source="user"):
|
|
"""Called by lifecycle create/submit with operations.lock already held."""
|
|
selections = normalize_tools(selections)
|
|
data = self._read(vm_id)
|
|
wanted = [item.to_dict() for item in selections]
|
|
for previous in reversed(data["jobs"]):
|
|
same = (previous["selections"] == wanted and previous["start_if_stopped"] == start_if_stopped
|
|
and previous.get("restart", False) == restart)
|
|
if request_id is not None and previous.get("request_id") == request_id:
|
|
if not same:
|
|
raise ConflictError("this installation request ID was already used for different tools")
|
|
return copy.deepcopy(previous)
|
|
if previous["status"] in ACTIVE_JOB_STATUSES:
|
|
if same:
|
|
return copy.deepcopy(previous)
|
|
raise ConflictError("a tool installation is already queued or running for this VM")
|
|
now = self.clock()
|
|
job = {"id": "tools-" + uuid.uuid4().hex, "vm_id": vm_id, "status": "queued", "source": source,
|
|
"selections": wanted, "start_if_stopped": start_if_stopped, "restart": restart,
|
|
"request_id": request_id, "created_at": now, "updated_at": now, "completed_at": None,
|
|
"error": None, "outcome_unknown": False,
|
|
"items": [{**item.to_dict(), "key": item.key, "status": "queued", "detected_version": None,
|
|
"error": None, "output": "", "truncated": False} for item in selections]}
|
|
data["jobs"] = [*data["jobs"], job][-MAX_RETAINED_JOBS:]
|
|
self._write(vm_id, data)
|
|
return copy.deepcopy(job)
|
|
|
|
def get_job(self, vm_id, job_id):
|
|
for job in self.read(vm_id)["jobs"]:
|
|
if job["id"] == job_id:
|
|
return job
|
|
raise NotFoundError("tool installation job not found for this VM")
|
|
|
|
def update_job(self, vm_id, job_id, update):
|
|
with self.transaction(vm_id) as data:
|
|
for job in data["jobs"]:
|
|
if job["id"] == job_id:
|
|
update(job, data)
|
|
job["updated_at"] = self.clock()
|
|
return copy.deepcopy(job)
|
|
raise NotFoundError("tool installation job not found for this VM")
|
|
|
|
def claim(self):
|
|
with self.state.operation_lock():
|
|
for vm in self.state.load().vms.values():
|
|
# A failed/interrupted create may not own a runtime directory.
|
|
if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir():
|
|
continue
|
|
data = self._read(vm.id)
|
|
for job in data["jobs"]:
|
|
if job["status"] == "queued":
|
|
job["status"] = "starting" if vm.status == "stopped" or job.get("restart") else "waiting"
|
|
job["updated_at"] = self.clock()
|
|
self._write(vm.id, data)
|
|
return copy.deepcopy(job)
|
|
return None
|
|
|
|
@staticmethod
|
|
def interrupt(job, message, now, *, outcome_unknown=True):
|
|
job.update(status="interrupted", error=message, outcome_unknown=outcome_unknown,
|
|
updated_at=now, completed_at=now)
|
|
for item in job["items"]:
|
|
if item["status"] in {"installing", "queued"}:
|
|
item["status"] = "interrupted" if item["status"] == "installing" else "skipped"
|
|
item["error"] = message
|
|
|
|
def recover(self):
|
|
"""The worker owner lock proves that previous in-flight host workers exited."""
|
|
with self.state.operation_lock():
|
|
for vm in self.state.load().vms.values():
|
|
if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir():
|
|
continue
|
|
data = self._read(vm.id)
|
|
changed = False
|
|
for job in data["jobs"]:
|
|
if job["status"] in ACTIVE_JOB_STATUSES - {"queued"}:
|
|
self.interrupt(job, "Management server restarted. Refresh tool status before retrying; "
|
|
"the last installation may have completed inside the guest.", self.clock())
|
|
changed = True
|
|
if changed:
|
|
data.update(agent_ready=False, guest_busy=False)
|
|
self._write(vm.id, data)
|