mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 21:43:02 +00:00
853 lines
35 KiB
Python
853 lines
35 KiB
Python
"""Virtualenv installation and process supervision for plugins."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess # nosec B404 - argument arrays only, never shell=True
|
|
import sys
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from .manifest import PluginManifest, read_archive_member, ui_subtree, validate_archive_limits
|
|
from .storage import PluginPaths, PluginStorage, write_release_asset
|
|
|
|
logger = logging.getLogger("PluginRuntime")
|
|
|
|
STOP_TIMEOUT_SECONDS = 5.0
|
|
VENV_TIMEOUT_SECONDS = 120.0
|
|
PIP_TIMEOUT_SECONDS = 300.0
|
|
PROCESS_KILL_TIMEOUT_SECONDS = 2.0
|
|
INSTALL_OUTPUT_MAX_BYTES = 64 * 1024
|
|
CRASH_WINDOW_SECONDS = 60.0
|
|
CRASH_MAX_EXITS = 5
|
|
LOG_MAX_BYTES = 5 * 1024 * 1024 # rotate simply by truncating when oversized
|
|
|
|
|
|
class PluginState(str, Enum):
|
|
DISABLED = "DISABLED"
|
|
STOPPED = "STOPPED"
|
|
STARTING = "STARTING"
|
|
RUNNING = "RUNNING"
|
|
STOPPING = "STOPPING"
|
|
FAILED = "FAILED"
|
|
|
|
|
|
@dataclass
|
|
class ProcessHandle:
|
|
plugin_id: str
|
|
process: subprocess.Popen
|
|
log_fp: object
|
|
started_at: float = field(default_factory=time.monotonic)
|
|
process_group: Optional[int] = None
|
|
log_thread: Optional[threading.Thread] = None
|
|
|
|
|
|
class PluginRuntime:
|
|
"""Installs wheels into isolated venvs and supervises plugin processes."""
|
|
|
|
def __init__(
|
|
self,
|
|
storage: PluginStorage,
|
|
*,
|
|
python_executable: Optional[str] = None,
|
|
stop_timeout: float = STOP_TIMEOUT_SECONDS,
|
|
crash_window: float = CRASH_WINDOW_SECONDS,
|
|
crash_max_exits: int = CRASH_MAX_EXITS,
|
|
popen_factory: Optional[Callable[..., subprocess.Popen]] = None,
|
|
run_factory: Optional[Callable[..., subprocess.CompletedProcess]] = None,
|
|
):
|
|
self.storage = storage
|
|
self.python_executable = python_executable or sys.executable
|
|
self.stop_timeout = stop_timeout
|
|
self.crash_window = crash_window
|
|
self.crash_max_exits = crash_max_exits
|
|
self._popen = popen_factory or subprocess.Popen
|
|
self._run = run_factory or self._run_install_command
|
|
self.on_output: Optional[Callable[[str], None]] = None
|
|
self._lock = threading.RLock()
|
|
self._handles: dict[str, ProcessHandle] = {}
|
|
self._runtime_state: dict[str, PluginState] = {}
|
|
self._crash_times: dict[str, deque[float]] = {}
|
|
self._exit_codes: dict[str, Optional[int]] = {}
|
|
self._supervise_stop = threading.Event()
|
|
self._supervise_thread: Optional[threading.Thread] = None
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Installation
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def install_wheel(self, wheel_path: Path | str, manifest: PluginManifest) -> dict:
|
|
"""Create release dir + venv and pip-install the wheel. Does not enable/start."""
|
|
wheel_path = Path(wheel_path).resolve()
|
|
if not wheel_path.is_file():
|
|
raise FileNotFoundError(f"wheel not found: {wheel_path}")
|
|
if not wheel_path.name.endswith(".whl"):
|
|
raise ValueError("install source must be a .whl file")
|
|
|
|
plugin_id = manifest.id
|
|
version = manifest.version
|
|
paths = self.storage.ensure_plugin_layout(plugin_id, version)
|
|
self.storage.write_manifest(plugin_id, version, manifest)
|
|
|
|
# Keep the exact install artifact with the release. Container data
|
|
# volumes outlive the image, so this lets us rebuild the isolated venv
|
|
# after a future base-image Python minor-version change.
|
|
release_dir = paths.release_dir(version)
|
|
archived_wheel = release_dir / wheel_path.name
|
|
if archived_wheel.resolve() != wheel_path:
|
|
for previous_wheel in release_dir.glob("*.whl"):
|
|
previous_wheel.unlink(missing_ok=True)
|
|
shutil.copy2(wheel_path, archived_wheel)
|
|
else:
|
|
archived_wheel = wheel_path
|
|
|
|
# Extract optional UI assets from the wheel into the release dir
|
|
self._extract_ui_assets(archived_wheel, release_dir, manifest)
|
|
# Also extract optional config.default.json next to the manifest if present
|
|
self._extract_config_default_file(archived_wheel, release_dir)
|
|
|
|
venv_path = paths.venv_dir(version)
|
|
if manifest.runtime is not None:
|
|
self._create_venv(venv_path)
|
|
self._pip_install(venv_path, archived_wheel)
|
|
|
|
# Seed data/config.json from defaults when the user has no saved config yet
|
|
self._seed_config_if_missing(plugin_id, manifest, paths.release_dir(version))
|
|
|
|
# Preserve enabled flag across reinstall
|
|
prev = self.storage.read_state(plugin_id) or {}
|
|
enabled = bool(prev.get("enabled", False))
|
|
self.storage.set_current(plugin_id, version)
|
|
self.storage.write_state(plugin_id, {"version": version, "enabled": enabled})
|
|
|
|
with self._lock:
|
|
if plugin_id not in self._runtime_state:
|
|
self._runtime_state[plugin_id] = (
|
|
PluginState.DISABLED if not enabled else PluginState.STOPPED
|
|
)
|
|
|
|
return {
|
|
"id": plugin_id,
|
|
"version": version,
|
|
"enabled": enabled,
|
|
"release_dir": str(paths.release_dir(version)),
|
|
"venv_dir": str(venv_path) if manifest.runtime else None,
|
|
"data_dir": str(paths.data_dir),
|
|
}
|
|
|
|
def ensure_venv_compatible(self, plugin_id: str) -> None:
|
|
"""Rebuild a persisted plugin venv when Python's minor version changes."""
|
|
state = self.storage.read_state(plugin_id)
|
|
manifest = self.storage.load_current_manifest(plugin_id)
|
|
if state is None or manifest is None or manifest.runtime is None:
|
|
return
|
|
|
|
version = str(state.get("version") or manifest.version)
|
|
paths = self.storage.paths_for(plugin_id)
|
|
venv_path = paths.venv_dir(version)
|
|
# Reconcile before either missing-config or matching-version early return:
|
|
# a newly created pyvenv.cfg does not mean pip/entrypoint validation finished.
|
|
rebuild_marker = venv_path.with_name(".venv-rebuild-in-progress")
|
|
if rebuild_marker.exists():
|
|
backup_name = rebuild_marker.read_text(encoding="ascii")
|
|
suffix = backup_name.removeprefix(".venv-backup-")
|
|
if backup_name != f".venv-backup-{uuid.UUID(hex=suffix).hex}":
|
|
raise ValueError("invalid venv rebuild backup marker")
|
|
interrupted_backup = venv_path.with_name(backup_name)
|
|
if interrupted_backup.exists():
|
|
if venv_path.exists():
|
|
shutil.rmtree(venv_path)
|
|
interrupted_backup.rename(venv_path)
|
|
# No backup means interruption before rename or after rollback.
|
|
if not venv_path.exists():
|
|
raise RuntimeError("interrupted venv rebuild has no recoverable environment")
|
|
rebuild_marker.unlink()
|
|
|
|
config_path = venv_path / "pyvenv.cfg"
|
|
if not config_path.is_file():
|
|
# Legacy/test venvs have no reliable version marker. Let normal
|
|
# entrypoint validation report a useful error instead of deleting.
|
|
return
|
|
|
|
persisted_version = None
|
|
try:
|
|
for line in config_path.read_text(encoding="utf-8").splitlines():
|
|
key, separator, value = line.partition("=")
|
|
if separator and key.strip().lower() == "version":
|
|
parts = value.strip().split(".")
|
|
if len(parts) >= 2:
|
|
persisted_version = (int(parts[0]), int(parts[1]))
|
|
break
|
|
except (OSError, ValueError):
|
|
return
|
|
|
|
current_version = (sys.version_info.major, sys.version_info.minor)
|
|
if persisted_version is None or persisted_version == current_version:
|
|
return
|
|
|
|
wheels = sorted(paths.release_dir(version).glob("*.whl"))
|
|
if not wheels:
|
|
raise RuntimeError(
|
|
f"plugin venv uses Python {persisted_version[0]}.{persisted_version[1]}, "
|
|
f"but the container uses {current_version[0]}.{current_version[1]}; "
|
|
"reinstall the plugin to rebuild its environment"
|
|
)
|
|
|
|
logger.warning(
|
|
"Rebuilding %s venv for Python %s.%s (was %s.%s)",
|
|
plugin_id,
|
|
current_version[0],
|
|
current_version[1],
|
|
persisted_version[0],
|
|
persisted_version[1],
|
|
)
|
|
# Build at the final absolute path: moving a freshly built venv would
|
|
# invalidate pip-generated shebangs. Keep the old environment for rollback.
|
|
backup = venv_path.with_name(f".venv-backup-{uuid.uuid4().hex}")
|
|
# Publish intent atomically before moving the old environment. This is
|
|
# process-interruption recovery, not a filesystem power-loss guarantee.
|
|
write_release_asset(venv_path.parent, rebuild_marker.name, backup.name.encode("ascii"))
|
|
venv_path.rename(backup)
|
|
try:
|
|
self._create_venv(venv_path)
|
|
self._pip_install(venv_path, wheels[0])
|
|
self.resolve_entrypoint(paths, version, manifest.runtime.entrypoint)
|
|
except BaseException:
|
|
if venv_path.exists():
|
|
shutil.rmtree(venv_path)
|
|
backup.rename(venv_path)
|
|
rebuild_marker.unlink()
|
|
raise
|
|
else:
|
|
# Commit only after validation; an interrupted cleanup may leave an
|
|
# unused backup, but must never restore a partly deleted backup.
|
|
rebuild_marker.unlink()
|
|
shutil.rmtree(backup)
|
|
|
|
def _announce(self, line: str) -> None:
|
|
if self.on_output is not None:
|
|
try:
|
|
self.on_output(line)
|
|
except Exception:
|
|
logger.debug("install output listener failed", exc_info=True)
|
|
|
|
def _run_install_command(self, cmd, *, timeout, **kwargs):
|
|
"""Drain output into bounded memory and kill the entire installer group."""
|
|
# Callers construct Python/venv/pip argv; installing trusted wheels is intentional.
|
|
proc = subprocess.Popen( # nosec B603
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
start_new_session=(os.name == "posix"),
|
|
)
|
|
handle = ProcessHandle(
|
|
"installer", proc, None, process_group=proc.pid if os.name == "posix" else None
|
|
)
|
|
output = bytearray()
|
|
pending = bytearray()
|
|
|
|
def emit(line: bytes) -> None:
|
|
self._announce(line.decode("utf-8", errors="replace").rstrip("\r"))
|
|
|
|
def drain():
|
|
with proc.stdout as source:
|
|
while chunk := source.read1(64 * 1024):
|
|
output.extend(chunk)
|
|
del output[:-INSTALL_OUTPUT_MAX_BYTES]
|
|
if self.on_output is not None:
|
|
pending.extend(chunk)
|
|
*lines, rest = pending.split(b"\n")
|
|
pending[:] = rest
|
|
del pending[:-INSTALL_OUTPUT_MAX_BYTES]
|
|
for line in lines:
|
|
emit(line)
|
|
if pending:
|
|
emit(bytes(pending))
|
|
|
|
reader = threading.Thread(target=drain, name="plugin-install-output", daemon=True)
|
|
reader.start()
|
|
try:
|
|
proc.wait(timeout=timeout)
|
|
finally:
|
|
self._signal_handle(handle, signal.SIGKILL)
|
|
proc.wait(timeout=PROCESS_KILL_TIMEOUT_SECONDS)
|
|
reader.join(timeout=PROCESS_KILL_TIMEOUT_SECONDS)
|
|
return subprocess.CompletedProcess(
|
|
cmd, proc.returncode, output.decode("utf-8", errors="replace"), ""
|
|
)
|
|
|
|
def _create_venv(self, venv_path: Path) -> None:
|
|
if venv_path.exists():
|
|
# Reinstall into existing venv is ok; recreate if broken
|
|
py = self._venv_python(venv_path)
|
|
if py.is_file():
|
|
return
|
|
venv_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._announce(f"Creating environment: {venv_path.name}")
|
|
result = self._run(
|
|
[self.python_executable, "-m", "venv", str(venv_path)],
|
|
timeout=VENV_TIMEOUT_SECONDS,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"venv creation failed ({result.returncode}): {result.stderr or result.stdout}"
|
|
)
|
|
|
|
def _pip_install(self, venv_path: Path, wheel_path: Path) -> None:
|
|
py = self._venv_python(venv_path)
|
|
if not py.is_file():
|
|
raise RuntimeError(f"venv python missing: {py}")
|
|
install_path = self._ensure_pip_wheel_filename(wheel_path)
|
|
self._announce(f"Installing {wheel_path.name} with pip")
|
|
try:
|
|
result = self._run(
|
|
[str(py), "-m", "pip", "install", "--upgrade", str(install_path)],
|
|
timeout=PIP_TIMEOUT_SECONDS,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
finally:
|
|
if install_path != wheel_path and install_path.exists():
|
|
try:
|
|
install_path.unlink()
|
|
except OSError:
|
|
pass
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"pip install failed ({result.returncode}): {result.stderr or result.stdout}"
|
|
)
|
|
|
|
@staticmethod
|
|
def _looks_like_wheel_filename(name: str) -> bool:
|
|
# PEP 427: {distribution}-{version}(-{build})?-{py}-{abi}-{plat}.whl
|
|
if not name.endswith(".whl"):
|
|
return False
|
|
stem = name[: -len(".whl")]
|
|
return stem.count("-") >= 4
|
|
|
|
@classmethod
|
|
def _ensure_pip_wheel_filename(cls, wheel_path: Path) -> Path:
|
|
"""Return a path whose basename pip will accept as a wheel filename."""
|
|
if cls._looks_like_wheel_filename(wheel_path.name):
|
|
return wheel_path
|
|
|
|
import re
|
|
import shutil
|
|
import zipfile
|
|
|
|
name = None
|
|
version = None
|
|
try:
|
|
with zipfile.ZipFile(wheel_path) as zf:
|
|
validate_archive_limits(zf)
|
|
for member in zf.namelist():
|
|
if member.endswith(".dist-info/METADATA") and member.count("/") == 1:
|
|
text = read_archive_member(zf, member, metadata=True).decode(
|
|
"utf-8", errors="replace"
|
|
)
|
|
for line in text.splitlines():
|
|
if line.startswith("Name:"):
|
|
name = line.split(":", 1)[1].strip()
|
|
elif line.startswith("Version:"):
|
|
version = line.split(":", 1)[1].strip()
|
|
break
|
|
except zipfile.BadZipFile:
|
|
name = None
|
|
|
|
def _normalize(value: str) -> str:
|
|
value = value.strip().replace("-", "_")
|
|
value = re.sub(r"[^A-Za-z0-9._]", "_", value)
|
|
return value or "package"
|
|
|
|
dist = _normalize(name or "package")
|
|
ver = _normalize(version or "0.0.0")
|
|
# py3-none-any is valid for pure-Python wheels and accepted by pip.
|
|
fixed_name = f"{dist}-{ver}-py3-none-any.whl"
|
|
fixed_path = wheel_path.with_name(fixed_name)
|
|
if fixed_path.exists() and fixed_path != wheel_path:
|
|
fixed_path.unlink()
|
|
shutil.copy2(wheel_path, fixed_path)
|
|
logger.info(
|
|
"Rewrote wheel filename for pip: %s -> %s",
|
|
wheel_path.name,
|
|
fixed_path.name,
|
|
)
|
|
return fixed_path
|
|
|
|
@staticmethod
|
|
def _venv_python(venv_path: Path) -> Path:
|
|
if os.name == "nt":
|
|
return venv_path / "Scripts" / "python.exe"
|
|
return venv_path / "bin" / "python"
|
|
|
|
@staticmethod
|
|
def _venv_bin_dir(venv_path: Path) -> Path:
|
|
if os.name == "nt":
|
|
return venv_path / "Scripts"
|
|
return venv_path / "bin"
|
|
|
|
def resolve_entrypoint(self, paths: PluginPaths, version: str, entrypoint: str) -> Path:
|
|
bin_dir = self._venv_bin_dir(paths.venv_dir(version))
|
|
candidate = (bin_dir / entrypoint).resolve()
|
|
bin_root = bin_dir.resolve()
|
|
try:
|
|
candidate.relative_to(bin_root)
|
|
except ValueError as exc:
|
|
raise RuntimeError("entrypoint path escapes plugin venv") from exc
|
|
if not candidate.is_file():
|
|
# Windows may use .exe
|
|
exe = Path(str(candidate) + ".exe")
|
|
if exe.is_file():
|
|
return exe
|
|
raise FileNotFoundError(f"entrypoint not found in venv: {entrypoint}")
|
|
return candidate
|
|
|
|
def _extract_ui_assets(
|
|
self, wheel_path: Path, release_dir: Path, manifest: PluginManifest
|
|
) -> None:
|
|
if manifest.ui is None:
|
|
return
|
|
import zipfile
|
|
|
|
entry = manifest.ui.entry.replace("\\", "/")
|
|
top = ui_subtree(entry)
|
|
try:
|
|
with zipfile.ZipFile(wheel_path) as zf:
|
|
validate_archive_limits(zf)
|
|
for member in zf.infolist():
|
|
parts = member.filename.split("/")
|
|
if member.is_dir() or top not in parts[:-1]:
|
|
continue
|
|
if any(part in {"", ".", ".."} or "\\" in part for part in parts):
|
|
raise ValueError("unsafe UI archive path")
|
|
rel = "/".join(parts[parts.index(top) :])
|
|
write_release_asset(release_dir, rel, read_archive_member(zf, member))
|
|
except zipfile.BadZipFile as exc:
|
|
raise RuntimeError(f"invalid wheel while extracting UI: {exc}") from exc
|
|
|
|
# Ensure entry file exists if packaged at package root differently
|
|
entry_path = release_dir / entry
|
|
if not entry_path.is_file():
|
|
logger.warning(
|
|
"UI entry %s not found in release dir after wheel extract for %s",
|
|
entry,
|
|
manifest.id,
|
|
)
|
|
|
|
def _extract_config_default_file(self, wheel_path: Path, release_dir: Path) -> None:
|
|
"""Copy optional config.default.json from the wheel into the release dir."""
|
|
import zipfile
|
|
|
|
try:
|
|
with zipfile.ZipFile(wheel_path) as zf:
|
|
validate_archive_limits(zf)
|
|
candidates = [
|
|
name
|
|
for name in zf.namelist()
|
|
if name.endswith("/config.default.json") or name == "config.default.json"
|
|
]
|
|
if not candidates:
|
|
return
|
|
preferred = [c for c in candidates if "share/openhop/plugins/" in c]
|
|
chosen = preferred[0] if preferred else candidates[0]
|
|
if ".." in Path(chosen).parts:
|
|
return
|
|
write_release_asset(
|
|
release_dir,
|
|
"config.default.json",
|
|
read_archive_member(zf, chosen, metadata=True),
|
|
)
|
|
except zipfile.BadZipFile:
|
|
return
|
|
|
|
def resolve_config_defaults(
|
|
self, plugin_id: str, manifest: Optional[PluginManifest] = None
|
|
) -> dict:
|
|
"""Load defaults from manifest and/or release config.default.json."""
|
|
if manifest is None:
|
|
manifest = self.storage.load_current_manifest(plugin_id)
|
|
defaults: dict = {}
|
|
if manifest and manifest.config_defaults:
|
|
defaults.update(manifest.config_defaults)
|
|
|
|
paths = self.storage.paths_for(plugin_id)
|
|
candidates: list[Path] = []
|
|
if paths.current_link.exists() or paths.current_link.is_symlink():
|
|
candidates.append(paths.current_link / "config.default.json")
|
|
version = self.storage.current_version(plugin_id)
|
|
if version:
|
|
try:
|
|
candidates.append(paths.release_dir(version) / "config.default.json")
|
|
except ValueError:
|
|
pass
|
|
candidates.append(paths.manifest_path().parent / "config.default.json")
|
|
|
|
seen: set[str] = set()
|
|
for candidate in candidates:
|
|
key = str(candidate)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
if not candidate.is_file():
|
|
continue
|
|
try:
|
|
import json
|
|
|
|
data = json.loads(candidate.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
defaults.update(data)
|
|
break
|
|
except (OSError, json.JSONDecodeError, ValueError) as exc:
|
|
logger.warning("Ignoring invalid %s: %s", candidate, exc)
|
|
return defaults
|
|
|
|
def _seed_config_if_missing(
|
|
self, plugin_id: str, manifest: PluginManifest, release_dir: Path
|
|
) -> None:
|
|
"""Write data/config.json from defaults when the user has no saved file yet."""
|
|
config_path = self.storage.config_path(plugin_id)
|
|
if config_path.is_file():
|
|
return
|
|
defaults = {}
|
|
if manifest.config_defaults:
|
|
defaults.update(manifest.config_defaults)
|
|
default_file = release_dir / "config.default.json"
|
|
if default_file.is_file():
|
|
try:
|
|
import json
|
|
|
|
data = json.loads(default_file.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
defaults.update(data)
|
|
except (OSError, json.JSONDecodeError, ValueError) as exc:
|
|
logger.warning("Could not read config.default.json for %s: %s", plugin_id, exc)
|
|
if not defaults:
|
|
return
|
|
try:
|
|
self.storage.write_config(plugin_id, defaults)
|
|
logger.info("Seeded default config.json for plugin %s", plugin_id)
|
|
except ValueError as exc:
|
|
logger.warning("Could not seed config for %s: %s", plugin_id, exc)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Process lifecycle
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def get_state(self, plugin_id: str) -> PluginState:
|
|
with self._lock:
|
|
persisted = self.storage.read_state(plugin_id)
|
|
if persisted is not None and not persisted.get("enabled", False):
|
|
# Keep FAILED visible even when disabled? Spec: disable → DISABLED
|
|
current = self._runtime_state.get(plugin_id)
|
|
if current in (PluginState.STOPPING, PluginState.STARTING):
|
|
return current
|
|
return PluginState.DISABLED
|
|
if plugin_id in self._handles:
|
|
proc = self._handles[plugin_id].process
|
|
if proc.poll() is None:
|
|
return PluginState.RUNNING
|
|
return self._runtime_state.get(plugin_id, PluginState.STOPPED)
|
|
|
|
def start(self, plugin_id: str) -> PluginState:
|
|
with self._lock:
|
|
state = self.storage.read_state(plugin_id)
|
|
if state is None:
|
|
raise KeyError(f"plugin not installed: {plugin_id}")
|
|
if not state.get("enabled", False):
|
|
# Allow explicit start only when enabled — callers enable first
|
|
raise RuntimeError(f"plugin is disabled: {plugin_id}")
|
|
|
|
# Persisted Docker volumes can outlive the image's Python minor
|
|
# version. Rebuild before resolving the installed entrypoint.
|
|
self.ensure_venv_compatible(plugin_id)
|
|
|
|
existing = self._handles.get(plugin_id)
|
|
if existing and existing.process.poll() is None:
|
|
self._runtime_state[plugin_id] = PluginState.RUNNING
|
|
return PluginState.RUNNING
|
|
|
|
manifest = self.storage.load_current_manifest(plugin_id)
|
|
if manifest is None:
|
|
raise RuntimeError(f"manifest missing for {plugin_id}")
|
|
if manifest.runtime is None:
|
|
# UI-only: nothing to start
|
|
self._runtime_state[plugin_id] = PluginState.STOPPED
|
|
return PluginState.STOPPED
|
|
|
|
version = str(state.get("version") or manifest.version)
|
|
paths = self.storage.paths_for(plugin_id)
|
|
exe = self.resolve_entrypoint(paths, version, manifest.runtime.entrypoint)
|
|
|
|
self._runtime_state[plugin_id] = PluginState.STARTING
|
|
paths.logs_dir.mkdir(parents=True, exist_ok=True)
|
|
self._rotate_log_if_needed(paths.log_file)
|
|
log_fp = open(paths.log_file, "ab", buffering=0)
|
|
|
|
env = os.environ.copy()
|
|
# Never expose a stale manager-only credential left behind by an
|
|
# upgraded deployment to third-party plugin processes.
|
|
env.pop("OPENHOP_PLUGIN_GITHUB_TOKEN", None)
|
|
env["OPENHOP_PLUGIN_ID"] = plugin_id
|
|
env["OPENHOP_PLUGIN_DATA"] = str(paths.data_dir)
|
|
|
|
try:
|
|
proc = self._popen(
|
|
[str(exe)],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
env=env,
|
|
cwd=str(paths.data_dir),
|
|
shell=False,
|
|
start_new_session=(os.name == "posix"),
|
|
)
|
|
except Exception:
|
|
log_fp.close()
|
|
self._runtime_state[plugin_id] = PluginState.FAILED
|
|
raise
|
|
|
|
self._handles[plugin_id] = ProcessHandle(
|
|
plugin_id=plugin_id,
|
|
process=proc,
|
|
log_fp=log_fp,
|
|
process_group=proc.pid if os.name == "posix" else None,
|
|
)
|
|
handle = self._handles[plugin_id]
|
|
handle.log_thread = threading.Thread(
|
|
target=self._drain_plugin_log,
|
|
args=(handle,),
|
|
name=f"plugin-log-{plugin_id}",
|
|
daemon=True,
|
|
)
|
|
handle.log_thread.start()
|
|
self._runtime_state[plugin_id] = PluginState.RUNNING
|
|
self._exit_codes[plugin_id] = None
|
|
logger.info("Started plugin %s pid=%s", plugin_id, proc.pid)
|
|
return PluginState.RUNNING
|
|
|
|
@staticmethod
|
|
def _drain_plugin_log(handle: ProcessHandle) -> None:
|
|
"""Continuously cap captured stdout/stderr, including newline-free output."""
|
|
source = getattr(handle.process, "stdout", None)
|
|
if source is None:
|
|
return
|
|
try:
|
|
with source:
|
|
while chunk := source.read1(64 * 1024):
|
|
chunk = chunk[-LOG_MAX_BYTES:]
|
|
if handle.log_fp.tell() + len(chunk) > LOG_MAX_BYTES:
|
|
handle.log_fp.seek(0)
|
|
handle.log_fp.truncate()
|
|
handle.log_fp.write(chunk)
|
|
except (OSError, ValueError):
|
|
logger.debug("Plugin log stream closed for %s", handle.plugin_id)
|
|
|
|
def stop(self, plugin_id: str, *, mark_disabled: bool = False) -> PluginState:
|
|
with self._lock:
|
|
handle = self._handles.get(plugin_id)
|
|
self._runtime_state[plugin_id] = (
|
|
PluginState.DISABLED if mark_disabled else PluginState.STOPPING
|
|
)
|
|
if handle is None:
|
|
self._runtime_state[plugin_id] = (
|
|
PluginState.DISABLED if mark_disabled else PluginState.STOPPED
|
|
)
|
|
return self._runtime_state[plugin_id]
|
|
|
|
# Keep the group identity captured at spawn even if the leader exited.
|
|
self._terminate_handle(handle)
|
|
|
|
with self._lock:
|
|
self._cleanup_handle(plugin_id)
|
|
self._runtime_state[plugin_id] = (
|
|
PluginState.DISABLED if mark_disabled else PluginState.STOPPED
|
|
)
|
|
logger.info("Stopped plugin %s", plugin_id)
|
|
return self._runtime_state[plugin_id]
|
|
|
|
@staticmethod
|
|
def _signal_handle(handle: ProcessHandle, sig: int) -> None:
|
|
group = handle.process_group
|
|
if group is not None:
|
|
# Never signal the manager's own group or a non-positive group ID.
|
|
if group <= 1 or group == os.getpgrp():
|
|
raise RuntimeError("refusing to signal unsafe plugin process group")
|
|
try:
|
|
os.killpg(group, sig)
|
|
except (PermissionError, ProcessLookupError):
|
|
pass
|
|
elif handle.process.poll() is None:
|
|
if sig == signal.SIGKILL:
|
|
handle.process.kill()
|
|
else:
|
|
handle.process.send_signal(sig)
|
|
|
|
def _terminate_handle(self, handle: ProcessHandle) -> None:
|
|
self._signal_handle(handle, signal.SIGTERM)
|
|
deadline = time.monotonic() + self.stop_timeout
|
|
while time.monotonic() < deadline:
|
|
handle.process.poll() # reap the leader, but do not equate it to the group
|
|
if handle.process_group is None:
|
|
if handle.process.poll() is not None:
|
|
break
|
|
else:
|
|
try:
|
|
os.killpg(handle.process_group, 0)
|
|
except (PermissionError, ProcessLookupError):
|
|
break
|
|
time.sleep(0.05)
|
|
self._signal_handle(handle, signal.SIGKILL)
|
|
try:
|
|
handle.process.wait(timeout=PROCESS_KILL_TIMEOUT_SECONDS)
|
|
except subprocess.TimeoutExpired:
|
|
logger.warning("Plugin %s did not reap after SIGKILL", handle.plugin_id)
|
|
|
|
def restart(self, plugin_id: str) -> PluginState:
|
|
self.stop(plugin_id)
|
|
return self.start(plugin_id)
|
|
|
|
def _cleanup_handle(self, plugin_id: str) -> None:
|
|
handle = self._handles.pop(plugin_id, None)
|
|
if handle is None:
|
|
return
|
|
# Crash supervision also owns descendants left behind by an exited leader.
|
|
self._signal_handle(handle, signal.SIGKILL)
|
|
try:
|
|
if handle.process.poll() is not None:
|
|
self._exit_codes[plugin_id] = handle.process.returncode
|
|
except Exception as exc:
|
|
logger.debug("Could not record exit status for %s: %s", plugin_id, exc)
|
|
if handle.log_thread is not None:
|
|
handle.log_thread.join(timeout=PROCESS_KILL_TIMEOUT_SECONDS)
|
|
try:
|
|
handle.log_fp.close()
|
|
except Exception as exc:
|
|
logger.debug("Could not close plugin log for %s: %s", plugin_id, exc)
|
|
|
|
@staticmethod
|
|
def _rotate_log_if_needed(log_file: Path) -> None:
|
|
try:
|
|
if log_file.is_file() and log_file.stat().st_size > LOG_MAX_BYTES:
|
|
rotated = log_file.with_suffix(".log.1")
|
|
if rotated.exists():
|
|
rotated.unlink()
|
|
log_file.rename(rotated)
|
|
except OSError:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Supervision
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def start_supervisor(self) -> None:
|
|
if self._supervise_thread and self._supervise_thread.is_alive():
|
|
return
|
|
self._supervise_stop.clear()
|
|
self._supervise_thread = threading.Thread(
|
|
target=self._supervise_loop, name="plugin-supervisor", daemon=True
|
|
)
|
|
self._supervise_thread.start()
|
|
|
|
def stop_supervisor(self) -> None:
|
|
self._supervise_stop.set()
|
|
thread = self._supervise_thread
|
|
if thread and thread.is_alive():
|
|
thread.join(timeout=2)
|
|
self._supervise_thread = None
|
|
|
|
def _supervise_loop(self) -> None:
|
|
while not self._supervise_stop.wait(0.5):
|
|
try:
|
|
self._check_crashes()
|
|
except Exception as exc:
|
|
logger.debug("supervisor tick error: %s", exc)
|
|
|
|
def _check_crashes(self) -> None:
|
|
with self._lock:
|
|
finished: list[tuple[str, int]] = []
|
|
for plugin_id, handle in list(self._handles.items()):
|
|
code = handle.process.poll()
|
|
if code is None:
|
|
continue
|
|
finished.append((plugin_id, code))
|
|
|
|
for plugin_id, code in finished:
|
|
self._cleanup_handle(plugin_id)
|
|
state = self.storage.read_state(plugin_id) or {}
|
|
if not state.get("enabled", False):
|
|
self._runtime_state[plugin_id] = PluginState.DISABLED
|
|
continue
|
|
# Expected stop paths set STOPPING/DISABLED already
|
|
current = self._runtime_state.get(plugin_id)
|
|
if current in (PluginState.STOPPING, PluginState.DISABLED, PluginState.STOPPED):
|
|
if current != PluginState.DISABLED:
|
|
self._runtime_state[plugin_id] = PluginState.STOPPED
|
|
continue
|
|
|
|
logger.warning("Plugin %s exited unexpectedly code=%s", plugin_id, code)
|
|
self._exit_codes[plugin_id] = code
|
|
times = self._crash_times.setdefault(plugin_id, deque())
|
|
now = time.monotonic()
|
|
times.append(now)
|
|
while times and (now - times[0]) > self.crash_window:
|
|
times.popleft()
|
|
if len(times) >= self.crash_max_exits:
|
|
logger.error(
|
|
"Plugin %s crash-looped (%d exits in %ss) → FAILED",
|
|
plugin_id,
|
|
len(times),
|
|
self.crash_window,
|
|
)
|
|
self._runtime_state[plugin_id] = PluginState.FAILED
|
|
continue
|
|
|
|
# Restart
|
|
try:
|
|
self._runtime_state[plugin_id] = PluginState.STOPPED
|
|
self.start(plugin_id)
|
|
except Exception as exc:
|
|
logger.error("Failed to restart plugin %s: %s", plugin_id, exc)
|
|
self._runtime_state[plugin_id] = PluginState.FAILED
|
|
|
|
def status_dict(self, plugin_id: str) -> dict:
|
|
state = self.storage.read_state(plugin_id)
|
|
if state is None:
|
|
raise KeyError(plugin_id)
|
|
manifest = self.storage.load_current_manifest(plugin_id)
|
|
paths = self.storage.paths_for(plugin_id)
|
|
runtime_state = self.get_state(plugin_id)
|
|
pid = None
|
|
with self._lock:
|
|
handle = self._handles.get(plugin_id)
|
|
if handle and handle.process.poll() is None:
|
|
pid = handle.process.pid
|
|
last_exit = self._exit_codes.get(plugin_id)
|
|
return {
|
|
"id": plugin_id,
|
|
"name": manifest.name if manifest else plugin_id,
|
|
"version": state.get("version"),
|
|
"enabled": bool(state.get("enabled", False)),
|
|
"state": runtime_state.value,
|
|
"pid": pid,
|
|
"last_exit_code": last_exit,
|
|
"has_runtime": bool(manifest and manifest.runtime),
|
|
"has_ui": bool(manifest and manifest.ui),
|
|
"ui_entry": manifest.ui.entry if manifest and manifest.ui else None,
|
|
"data_dir": str(paths.data_dir),
|
|
"description": manifest.description if manifest else "",
|
|
"source": state.get("source") or "local",
|
|
"repository": state.get("repository") or None,
|
|
}
|