sandcastle-ng

sandcastle-wait (5389B)

---

 #!/usr/bin/env python3
 
 # This file is in the public domain.
 
 from __future__ import annotations
 
 import argparse
 import signal
 import subprocess
 import sys
 import time
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
 
 CONTAINER = "taler-sandcastle"
 UNIT = "setup-sandcastle.service"
 STARTUP_ATTEMPTS = 20
 STARTUP_RETRY_SECONDS = 0.1
 POLL_SECONDS = 1
 JOURNAL_STOP_SECONDS = 2
 
 
 class TerminationRequested(Exception):
     def __init__(self, signum: int):
         super().__init__(signum)
         self.signum = signum
 
 
 def container_is_running() -> bool:
     try:
         result = subprocess.run(
             ["podman", "inspect", "--format", "{{.State.Running}}", CONTAINER],
             check=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.DEVNULL,
             text=True,
         )
     except subprocess.CalledProcessError:
         return False
     return result.stdout.strip() == "true"
 
 
 def unit_state(*, quiet: bool = False) -> str:
     result = subprocess.run(
         [
             "podman",
             "exec",
             CONTAINER,
             "systemctl",
             "show",
             "--property=ActiveState",
             "--value",
             UNIT,
         ],
         check=True,
         stdout=subprocess.PIPE,
         stderr=subprocess.DEVNULL if quiet else None,
         text=True,
     )
     return result.stdout.strip()
 
 
 def initial_unit_state() -> str | None:
     # A detached podman run can report success just before exec is ready. Retry
     # that narrow startup race without imposing a timeout on provisioning.
     for attempt in range(STARTUP_ATTEMPTS):
         try:
             return unit_state(quiet=True)
         except subprocess.CalledProcessError:
             if attempt + 1 < STARTUP_ATTEMPTS:
                 time.sleep(STARTUP_RETRY_SECONDS)
 
     print(f"Cannot query {UNIT} in container {CONTAINER}.", file=sys.stderr)
     return None
 
 
 def stop_process(process: subprocess.Popen[bytes]) -> None:
     if process.poll() is not None:
         return
     try:
         process.terminate()
     except ProcessLookupError:
         process.wait()
         return
     try:
         process.wait(timeout=JOURNAL_STOP_SECONDS)
     except subprocess.TimeoutExpired:
         process.kill()
         process.wait()
 
 
 @contextmanager
 def journal_output() -> Iterator[subprocess.Popen[bytes]]:
     process = subprocess.Popen(
         [
             "podman",
             "exec",
             CONTAINER,
             "journalctl",
             "--unit",
             UNIT,
             "--follow",
             "--no-pager",
         ]
     )
     try:
         yield process
     finally:
         stop_process(process)
 
 
 def report_failure(message: str) -> None:
     print(message, file=sys.stderr)
     subprocess.run(
         [
             "podman",
             "exec",
             CONTAINER,
             "systemctl",
             "status",
             "--no-pager",
             "--full",
             UNIT,
         ],
         check=False,
         stdout=sys.stderr,
         stderr=sys.stderr,
     )
 
 
 def terminal_status(state: str) -> int:
     if state == "active":
         print("Sandcastle provisioning succeeded.")
         return 0
     if state == "failed":
         report_failure("Sandcastle provisioning failed.")
         return 1
 
     report_failure(f"Sandcastle provisioning entered unexpected state {state!r}.")
     return 1
 
 
 def monitor_provisioning() -> int:
     if not container_is_running():
         print(
             f"Container {CONTAINER} does not exist or is not running.",
             file=sys.stderr,
         )
         return 1
 
     state = initial_unit_state()
     if state is None:
         return 1
     if state != "activating":
         return terminal_status(state)
 
     print(f"Sandcastle provisioning is running; following {UNIT} output.", flush=True)
     with journal_output() as journal:
         while True:
             time.sleep(POLL_SECONDS)
             try:
                 state = unit_state()
             except subprocess.CalledProcessError:
                 print(
                     f"Lost contact with {UNIT} in container {CONTAINER}.",
                     file=sys.stderr,
                 )
                 return 1
 
             if state != "activating":
                 break
 
             journal_status = journal.poll()
             if journal_status is not None:
                 print(
                     f"Stopped receiving output from {UNIT} "
                     f"(podman exec exited with status {journal_status}).",
                     file=sys.stderr,
                 )
                 return 1
 
     return terminal_status(state)
 
 
 def request_termination(signum: int, _frame: object) -> None:
     raise TerminationRequested(signum)
 
 
 def main(argv: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         description="Wait for sandcastle provisioning to finish."
     )
     parser.parse_args(argv)
     signal.signal(signal.SIGTERM, request_termination)
 
     try:
         return monitor_provisioning()
     except KeyboardInterrupt:
         return 128 + signal.SIGINT
     except TerminationRequested as request:
         return 128 + request.signum
     except FileNotFoundError as error:
         print(f"Cannot execute {error.filename}: {error.strerror}.", file=sys.stderr)
         return 1
 
 
 if __name__ == "__main__":
     raise SystemExit(main())