autonomic-bot
f283a371bb
Sanitized single-commit public mirror of recipe-maintainer. - Removed test-ssh/.testenv (live creds); added test-ssh/.testenv.example placeholders. - Removed plans/ and planned-updates/ (deployment-planning docs) so no client/ deployment domains appear in the public repo. - All other secret stores were already gitignored. - docs.coopcloud.tech retained as a submodule (public upstream).
292 lines
9.2 KiB
Python
292 lines
9.2 KiB
Python
"""Wrapper for the abra CLI.
|
|
|
|
Encapsulates TTY wrapping, --chaos flag, --no-input, timeout handling,
|
|
and the Linux `timeout` command guard.
|
|
|
|
All abra commands go through this module so that learnings (TTY requirements,
|
|
--chaos caveats, timeout behaviors) are encoded in one place.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Result type
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class AbraResult:
|
|
"""Result of a shell/abra command."""
|
|
returncode: int
|
|
stdout: str
|
|
stderr: str
|
|
timed_out: bool
|
|
command: str
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.returncode == 0 and not self.timed_out
|
|
|
|
def json(self) -> list | dict | None:
|
|
"""Try to parse stdout as JSON."""
|
|
try:
|
|
return json.loads(self.stdout)
|
|
except (json.JSONDecodeError, ValueError):
|
|
return None
|
|
|
|
def jsonl(self) -> list[list | dict]:
|
|
"""Parse stdout as JSON Lines (one JSON document per line)."""
|
|
results = []
|
|
for line in self.stdout.strip().split("\n"):
|
|
if line.strip():
|
|
try:
|
|
results.append(json.loads(line))
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
return results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TTY requirements from learnings.md
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# These abra subcommands require a TTY wrapper (script -qefc).
|
|
# Without it they fail with "not a TTY" or hang indefinitely.
|
|
_TTY_COMMANDS = {
|
|
"secret insert", "secret remove", "secret generate",
|
|
"volume remove",
|
|
"cmd",
|
|
"backup create", "backup snapshots", "backup list",
|
|
"restore",
|
|
"recipe lint",
|
|
}
|
|
|
|
|
|
def _needs_tty(args: str) -> bool:
|
|
"""Check if an abra command needs the TTY wrapper."""
|
|
# Normalize: "app secret insert ..." → check "secret insert"
|
|
parts = args.strip()
|
|
# Strip leading "app " if present
|
|
if parts.startswith("app "):
|
|
parts = parts[4:]
|
|
for cmd in _TTY_COMMANDS:
|
|
if parts.startswith(cmd):
|
|
return True
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core execution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run(cmd: str, *, check: bool = True, timeout: int = 120) -> AbraResult:
|
|
"""Run a shell command with a hard timeout.
|
|
|
|
Uses the Linux `timeout` command to guarantee the process tree is
|
|
killed after `timeout` seconds, even if the process ignores signals.
|
|
subprocess.run gets a slightly longer timeout as a fallback.
|
|
"""
|
|
wrapped = f"timeout --kill-after=5 {timeout} {cmd}"
|
|
print(f" $ {cmd}", flush=True)
|
|
try:
|
|
result = subprocess.run(
|
|
wrapped, shell=True, capture_output=True, text=True,
|
|
timeout=timeout + 15,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
print(f" TIMEOUT after {timeout}s (subprocess fallback)", flush=True)
|
|
if check:
|
|
raise RuntimeError(f"Command timed out after {timeout}s: {cmd}")
|
|
return AbraResult(
|
|
returncode=124, stdout="", stderr="",
|
|
timed_out=True, command=cmd,
|
|
)
|
|
|
|
timed_out = result.returncode == 124
|
|
if result.stdout.strip():
|
|
for line in result.stdout.strip().split("\n"):
|
|
print(f" {line}", flush=True)
|
|
if result.returncode != 0:
|
|
if result.stderr.strip():
|
|
for line in result.stderr.strip().split("\n"):
|
|
print(f" stderr: {line}", flush=True)
|
|
if timed_out:
|
|
print(f" TIMEOUT after {timeout}s", flush=True)
|
|
if check:
|
|
raise RuntimeError(f"Command timed out after {timeout}s: {cmd}")
|
|
elif check:
|
|
raise RuntimeError(
|
|
f"Command failed (exit {result.returncode}): {cmd}"
|
|
)
|
|
|
|
return AbraResult(
|
|
returncode=result.returncode,
|
|
stdout=result.stdout,
|
|
stderr=result.stderr,
|
|
timed_out=timed_out,
|
|
command=cmd,
|
|
)
|
|
|
|
|
|
def abra(args: str, *, tty_wrap: bool | None = None,
|
|
check: bool = True, timeout: int = 120,
|
|
chaos: bool = True) -> AbraResult:
|
|
"""Run an abra command.
|
|
|
|
- tty_wrap: wrap with script -qefc for commands that need TTY.
|
|
If None, auto-detects from the command.
|
|
- chaos: append --chaos (default True for local dev)
|
|
- Always appends --no-input
|
|
"""
|
|
flags = "--no-input"
|
|
if chaos and "--chaos" not in args:
|
|
flags += " --chaos"
|
|
|
|
cmd = f"abra {args} {flags}".strip()
|
|
|
|
if tty_wrap is None:
|
|
tty_wrap = _needs_tty(args)
|
|
|
|
if tty_wrap:
|
|
cmd = f'script -qefc "{cmd}" /dev/null 2>&1'
|
|
|
|
return run(cmd, check=check, timeout=timeout)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# High-level abra operations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def app_deploy(domain: str, *, force: bool = True,
|
|
chaos: bool = True, timeout: int = 60) -> AbraResult:
|
|
"""Deploy an app."""
|
|
flags = f"app deploy {domain}"
|
|
if force:
|
|
flags += " --force"
|
|
return abra(flags, chaos=chaos, check=False, timeout=timeout)
|
|
|
|
|
|
def app_undeploy(domain: str, *, timeout: int = 60) -> AbraResult:
|
|
"""Undeploy an app."""
|
|
return abra(f"app undeploy {domain}", chaos=False, check=False, timeout=timeout)
|
|
|
|
|
|
def app_new(recipe: str, server: str, domain: str, *,
|
|
chaos: bool = True, timeout: int = 60) -> AbraResult:
|
|
"""Create a new app instance."""
|
|
return abra(
|
|
f"app new {recipe} --server {server} --domain {domain}",
|
|
chaos=chaos, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_ps(domain: str, *, chaos: bool = True) -> AbraResult:
|
|
"""Get app process status (machine-readable)."""
|
|
return abra(f"app ps {domain} -m", chaos=chaos, check=False, timeout=30)
|
|
|
|
|
|
def app_secret_generate(domain: str, *,
|
|
chaos: bool = True, timeout: int = 60) -> AbraResult:
|
|
"""Generate all secrets for an app."""
|
|
return abra(
|
|
f"app secret generate {domain} --all",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_secret_insert(domain: str, name: str, version: str,
|
|
value: str, *, chaos: bool = True,
|
|
timeout: int = 60) -> AbraResult:
|
|
"""Insert a specific secret."""
|
|
return abra(
|
|
f"app secret insert {domain} {name} {version} {value}",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_secret_remove_all(domain: str, *,
|
|
chaos: bool = True,
|
|
timeout: int = 60) -> AbraResult:
|
|
"""Remove all secrets for an app."""
|
|
return abra(
|
|
f"app secret remove {domain} --all",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_volume_remove(domain: str, *, timeout: int = 60) -> AbraResult:
|
|
"""Remove all volumes for an app."""
|
|
return abra(
|
|
f"app volume remove {domain} --force",
|
|
chaos=False, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_cmd(domain: str, service: str, cmd_name: str, *,
|
|
chaos: bool = True, timeout: int = 120) -> AbraResult:
|
|
"""Run an abra app cmd."""
|
|
return abra(
|
|
f"app cmd {domain} {service} {cmd_name}",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_ls(server: str, *, timeout: int = 30) -> AbraResult:
|
|
"""List all apps on a server (machine-readable)."""
|
|
return abra(f"app ls -s {server} -S -m", chaos=False, check=False,
|
|
timeout=timeout)
|
|
|
|
|
|
def app_backup_create(domain: str, *,
|
|
chaos: bool = True, timeout: int = 120) -> AbraResult:
|
|
"""Create a backup of an app."""
|
|
return abra(
|
|
f"app backup create {domain}",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def app_backup_restore(domain: str, *,
|
|
chaos: bool = True, timeout: int = 120) -> AbraResult:
|
|
"""Restore an app from backup."""
|
|
return abra(
|
|
f"app restore {domain}",
|
|
chaos=chaos, check=False, timeout=timeout,
|
|
)
|
|
|
|
|
|
def recipe_fetch(recipe: str, *, force: bool = True,
|
|
timeout: int = 60) -> AbraResult:
|
|
"""Fetch a recipe from upstream."""
|
|
flags = f"recipe fetch {recipe}"
|
|
if force:
|
|
flags += " --force"
|
|
return abra(flags, chaos=False, timeout=timeout)
|
|
|
|
|
|
def recipe_versions(recipe: str) -> AbraResult:
|
|
"""List recipe versions (machine-readable)."""
|
|
return abra(f"recipe versions {recipe} -m", chaos=False, check=False,
|
|
timeout=30)
|
|
|
|
|
|
def recipe_upgrade(recipe: str, *, dry_run: bool = True) -> AbraResult:
|
|
"""Check for recipe upgrades (machine-readable)."""
|
|
flags = f"recipe upgrade {recipe} -m"
|
|
if dry_run:
|
|
flags += " -n"
|
|
return abra(flags, chaos=False, check=False, timeout=30)
|
|
|
|
|
|
def recipe_lint(recipe: str) -> AbraResult:
|
|
"""Lint a recipe."""
|
|
return abra(f"recipe lint {recipe} -C", chaos=False, check=False,
|
|
timeout=30)
|
|
|
|
|
|
def recipe_diff(recipe: str) -> AbraResult:
|
|
"""Show local changes in a recipe checkout."""
|
|
return abra(f"recipe diff {recipe}", chaos=False, check=False, timeout=15)
|