mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-30 16:47:01 +09:00
Juan Denis
6296887bad
Comments, docstrings and a few strings cited plan numbers ("plan 29
#9", "plan 86 §B2") from a git-ignored folder, so they pointed at
nothing; one reached users in a bottleneck hint. Code and frontend
strings are unchanged apart from those citations (checked by AST and
token comparison). A vhost still carrying the old asset comment reads
as undrifted, and a guard test keeps new citations out.
53 lines
2.3 KiB
Python
53 lines
2.3 KiB
Python
"""One-shot appliance first-boot bootstrap.
|
|
|
|
Some appliances generate a config/certificate tree exactly once, on first run.
|
|
Abusing a compose init container leaves it behind; abusing an entrypoint reruns
|
|
it every deploy. Instead this runs the declared ``bootstrap.command`` ONCE via
|
|
``docker compose run --rm <service> <command>`` — after the volumes exist and
|
|
before the first ``up`` — and flips ``Application.bootstrap_done`` on success so
|
|
it never re-runs. Failure is a visible, retryable apply-step failure.
|
|
|
|
The runner is injectable (``set_runner``) so the apply engine can be proven
|
|
without Docker.
|
|
"""
|
|
|
|
import shlex
|
|
from typing import Any, Callable, Dict, Optional
|
|
|
|
|
|
class BootstrapService:
|
|
|
|
_runner: Optional[Callable] = None # test seam
|
|
|
|
@classmethod
|
|
def set_runner(cls, fn: Optional[Callable]) -> None:
|
|
"""Install/clear an injected runner ``fn(app, service, command, timeout)``
|
|
returning ``{'success': bool, 'output'|'error': str}``."""
|
|
cls._runner = fn
|
|
|
|
@classmethod
|
|
def run_once(cls, app, command: str, timeout_seconds: Optional[int] = None,
|
|
service: Optional[str] = None) -> Dict[str, Any]:
|
|
service = service or app.name
|
|
if cls._runner is not None:
|
|
return cls._runner(app, service, command, timeout_seconds)
|
|
return cls._docker_run(app, service, command, timeout_seconds)
|
|
|
|
@staticmethod
|
|
def _docker_run(app, service: str, command: str,
|
|
timeout_seconds: Optional[int]) -> Dict[str, Any]:
|
|
from app.utils.system import run_checked
|
|
root = getattr(app, 'root_path', None)
|
|
if not root:
|
|
return {'success': False,
|
|
'error': 'This service has no deployed Compose project to run bootstrap in. Deploy it first.'}
|
|
cmd = ['docker', 'compose', 'run', '--rm', service] + shlex.split(command)
|
|
proc = run_checked(cmd, cwd=root, timeout=timeout_seconds or 300)
|
|
if proc['returncode'] is None:
|
|
# timed out, or docker is absent
|
|
return {'success': False, 'error': proc['error']}
|
|
if not proc['success']:
|
|
return {'success': False,
|
|
'error': (proc['error'] or proc['output'] or 'bootstrap failed')[:500]}
|
|
return {'success': True, 'output': proc['output'].strip()[:500]}
|