mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 21:43:02 +00:00
213 lines
7.3 KiB
Python
213 lines
7.3 KiB
Python
"""Container supervision for Repeater and its plugin manager."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import signal
|
|
|
|
# Subprocess calls below use fixed argument arrays and never shell=True.
|
|
import subprocess # nosec B404
|
|
import sys
|
|
import time
|
|
from collections.abc import Callable, Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
logger = logging.getLogger("ContainerSupervisor")
|
|
_FALSE_VALUES = {"0", "false", "no", "off"}
|
|
|
|
|
|
def plugin_manager_enabled(
|
|
config_path: Path | str,
|
|
environ: Mapping[str, str] | None = None,
|
|
) -> bool:
|
|
"""Return whether the container should run the plugin manager."""
|
|
env = os.environ if environ is None else environ
|
|
if str(env.get("OPENHOP_PLUGIN_MANAGER", "1")).strip().lower() in _FALSE_VALUES:
|
|
return False
|
|
|
|
try:
|
|
with open(config_path, encoding="utf-8") as config_file:
|
|
config = yaml.safe_load(config_file) or {}
|
|
except (OSError, yaml.YAMLError):
|
|
return True
|
|
|
|
if not isinstance(config, dict):
|
|
return True
|
|
plugins = config.get("plugins")
|
|
return not (isinstance(plugins, dict) and plugins.get("enabled") is False)
|
|
|
|
|
|
def _signal_process_group(process: subprocess.Popen, signum: int) -> None:
|
|
"""Signal a child session, falling back to the direct process."""
|
|
try:
|
|
os.killpg(process.pid, signum)
|
|
except ProcessLookupError:
|
|
return
|
|
except OSError:
|
|
try:
|
|
process.send_signal(signum)
|
|
except ProcessLookupError:
|
|
return
|
|
|
|
|
|
def _process_group_exists(process: subprocess.Popen) -> bool:
|
|
"""Return whether any process remains in a child session."""
|
|
try:
|
|
os.killpg(process.pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
return True
|
|
return True
|
|
|
|
|
|
class ContainerSupervisor:
|
|
"""Run, restart, and gracefully stop container child processes."""
|
|
|
|
def __init__(
|
|
self,
|
|
config_path: Path | str,
|
|
*,
|
|
manager_enabled: bool | None = None,
|
|
python_executable: str = "python3",
|
|
popen_factory: Callable[..., subprocess.Popen] = subprocess.Popen,
|
|
signal_process_group: Callable[[subprocess.Popen, int], None] = _signal_process_group,
|
|
process_group_exists: Callable[[subprocess.Popen], bool] = _process_group_exists,
|
|
sleep: Callable[[float], None] = time.sleep,
|
|
manager_restart_delay: float = 2.0,
|
|
shutdown_timeout: float = 8.0,
|
|
):
|
|
self.config_path = str(config_path)
|
|
self.manager_enabled = (
|
|
plugin_manager_enabled(config_path) if manager_enabled is None else manager_enabled
|
|
)
|
|
self.python_executable = python_executable
|
|
self._popen = popen_factory
|
|
self._signal_process_group = signal_process_group
|
|
self._process_group_exists = process_group_exists
|
|
self._sleep = sleep
|
|
self.manager_restart_delay = manager_restart_delay
|
|
self.shutdown_timeout = shutdown_timeout
|
|
self.manager_process: subprocess.Popen | None = None
|
|
self.repeater_process: subprocess.Popen | None = None
|
|
self._stopping = False
|
|
self._shutdown_signal = signal.SIGTERM
|
|
|
|
def _start(self, module: str) -> subprocess.Popen:
|
|
command = [
|
|
self.python_executable,
|
|
"-m",
|
|
module,
|
|
"--config",
|
|
self.config_path,
|
|
]
|
|
return self._popen(command, start_new_session=True)
|
|
|
|
def _start_manager(self) -> None:
|
|
logger.info("Starting plugin manager")
|
|
self.manager_process = self._start("repeater.plugins")
|
|
|
|
def request_shutdown(self, signum: int = signal.SIGTERM) -> None:
|
|
if self._stopping:
|
|
return
|
|
self._stopping = True
|
|
self._shutdown_signal = signum
|
|
logger.info("Signal %s received; stopping container services", signum)
|
|
for process in (self.repeater_process, self.manager_process):
|
|
if process is not None:
|
|
self._signal_process_group(process, signum)
|
|
|
|
def _wait_or_kill(self, process: subprocess.Popen | None) -> None:
|
|
if process is None:
|
|
return
|
|
try:
|
|
process.wait(timeout=self.shutdown_timeout)
|
|
return
|
|
except subprocess.TimeoutExpired:
|
|
logger.warning("Process group %s did not stop in time; killing it", process.pid)
|
|
self._signal_process_group(process, signal.SIGKILL)
|
|
try:
|
|
process.wait(timeout=2)
|
|
except (subprocess.TimeoutExpired, ChildProcessError):
|
|
pass
|
|
|
|
def _finish(self) -> None:
|
|
if not self._stopping:
|
|
self.request_shutdown(signal.SIGTERM)
|
|
self._wait_or_kill(self.repeater_process)
|
|
self._wait_or_kill(self.manager_process)
|
|
|
|
def run(self) -> int:
|
|
if self.manager_enabled:
|
|
self._start_manager()
|
|
else:
|
|
logger.info("Plugin manager disabled")
|
|
|
|
self.repeater_process = self._start("repeater.main")
|
|
|
|
try:
|
|
while not self._stopping:
|
|
repeater_code = self.repeater_process.poll()
|
|
if repeater_code is not None:
|
|
return int(repeater_code)
|
|
|
|
if self.manager_enabled and self.manager_process is not None:
|
|
manager_code = self.manager_process.poll()
|
|
if manager_code is not None:
|
|
logger.error(
|
|
"Plugin manager exited unexpectedly with status %s; restarting",
|
|
manager_code,
|
|
)
|
|
# Terminate any plugin descendants left in the old process group.
|
|
self._signal_process_group(self.manager_process, signal.SIGTERM)
|
|
self._sleep(self.manager_restart_delay)
|
|
if self._process_group_exists(self.manager_process):
|
|
logger.warning(
|
|
"Plugin descendants did not stop with manager %s; killing group",
|
|
self.manager_process.pid,
|
|
)
|
|
self._signal_process_group(self.manager_process, signal.SIGKILL)
|
|
if not self._stopping and self.repeater_process.poll() is None:
|
|
self._start_manager()
|
|
|
|
self._sleep(0.2)
|
|
finally:
|
|
self._finish()
|
|
|
|
code = self.repeater_process.poll()
|
|
if code is None:
|
|
return 128 + int(self._shutdown_signal)
|
|
return int(code)
|
|
|
|
|
|
def build_arg_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="openHop Docker process supervisor")
|
|
parser.add_argument("--config", required=True, help="Path to Repeater config.yaml")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_arg_parser().parse_args(argv)
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
supervisor = ContainerSupervisor(args.config)
|
|
|
|
def handle_signal(signum: int, _frame: Any) -> None:
|
|
supervisor.request_shutdown(signum)
|
|
|
|
for signum in (signal.SIGINT, signal.SIGTERM):
|
|
signal.signal(signum, handle_signal)
|
|
|
|
return supervisor.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|