"""Swarm variant-matrix round orchestrator.""" from __future__ import annotations import argparse import json import os import subprocess import sys from plugins.filter.merge.with_defaults import merge_with_defaults from utils import PROJECT_ROOT from utils.storage.constrained import host_storage_constrained from utils.tests.swarm.derive_includes import derive_includes, variant_scope from utils.tests.swarm.extend_inventory import mesh_enabled from utils.tests.swarm.mesh import ( bootstrap_mesh, converge_mesh, lab_sudo_without_prompt, layout_applications, mesh_controller, switch_to_mesh_transport, write_mesh, ) from utils.tests.swarm.run import DISK_FLOOR_MB, run_step from utils.tests.swarm.write.extras import ensure_swarm_keypairs _SWARM_DIR = PROJECT_ROOT / "scripts" / "tests" / "deploy" / "swarm" _SWARM_SCRIPTS = _SWARM_DIR / "routine" _ROLES_DIR = str(PROJECT_ROOT / "roles") _SWARM_EXTRAS_VARS = "inventories/development/swarm.yml" _DEFAULT_INVENTORY_DIR = "/tmp/inv" # noqa: S108 - ephemeral swarm-test inventory base in CI def _provision( *, app_id: str, inv_dir: str, round_variants: dict[str, int], vars_payload: dict ) -> int: env = os.environ.copy() env["APP_ID"] = app_id env["INFINITO_INVENTORY_DIR"] = inv_dir env["INFINITO_APP_VARIANTS"] = json.dumps(round_variants, sort_keys=True) env["INFINITO_VARS_PAYLOAD"] = json.dumps(vars_payload, sort_keys=True) return run_step( ["bash", str(_SWARM_SCRIPTS / "02_provision_inventory.sh")], env=env, label=f"provision inventory ({inv_dir})", ) def _extend_inventory( *, app_id: str, inv_dir: str, round_variants: dict[str, int] ) -> int: env = os.environ.copy() env["APP_ID"] = app_id env["INV_PATH"] = f"{inv_dir}/devices.yml" env["INFINITO_APP_VARIANTS"] = json.dumps(round_variants, sort_keys=True) return run_step( ["python3", "-m", "utils.tests.swarm.extend_inventory"], env=env, label="extend inventory (workers + group memberships)", ) def _force_shared_db(*, inv_dir: str) -> int: env = os.environ.copy() env["INV_DIR"] = inv_dir return run_step( ["python3", "-m", "utils.tests.swarm.force_shared_db"], env=env, label="force shared DB (swarm: embedded DB is compose-only)", ) def _write_extras(*, extras_path: str) -> int: env = os.environ.copy() env["OUT_PATH"] = extras_path return run_step( ["python3", "-m", "utils.tests.swarm.write.extras"], env=env, label=f"write runtime extras ({extras_path})", ) def _reset_credentials( *, app_id: str, inv_dir: str, round_variants: dict[str, int] ) -> int: """Regenerate the round's credentials so the update pass has to carry them.""" return run_step( [ "python3", "-m", "cli.administration.inventory.credentials.reset", "--inventory-dir", inv_dir, "--host", os.environ["MGR"], "--schema", "--users", "--include", *derive_includes(app_id, variants=round_variants), "--app-variants", json.dumps(round_variants, sort_keys=True), "--mirror", "--mirror-keep", "applications.svc-net-wireguard", "--backup", "--except", "administrator", ], env=os.environ.copy(), label="reset credentials (rotation gate before the async pass)", ) def _deploy( *, app_id: str, inv_dir: str, extras_path: str, round_index: int, total: int, update_pass: bool = False, ) -> int: """Run one deploy of the round. Args: update_pass: run it as the async update pass. """ env = os.environ.copy() env["APP_ID"] = app_id cmd = [ "python3", "-m", "cli.administration.deploy.swarm", f"{inv_dir}/devices.yml", "-p", f"{inv_dir}/.password", "--skip-build", "--skip-cleanup", "--skip-backup", "-e", f"@{_SWARM_EXTRAS_VARS}", "-e", f"@{extras_path}", "-e", f"VARIANT_INDEX={json.dumps(round_index)}", "-e", f"PRIMARY_APPS={json.dumps([app_id])}", ] pass_label = ( f"matrix-deploy: round {round_index + 1}/{total} " f"variants=[{round_index}] apps=['{app_id}']" ) if update_pass: cmd += ["-e", "ASYNC_ENABLED=true"] label = f"update pass (round {round_index + 1}/{total})" print(f"=== {pass_label} PASS 2 (async) ===", flush=True) else: label = f"deploy round {round_index + 1}/{total}" print(f"=== {pass_label} PASS 1 (sync) ===", flush=True) return run_step(env=env, cmd=cmd, label=label) def _deploy_backup_host(*, app_id: str, inv_dir: str, extras_path: str) -> int: env = os.environ.copy() env["APP_ID"] = app_id cmd = [ "python3", "-m", "cli.administration.deploy.dedicated", f"{inv_dir}/backup.yml", "-p", f"{inv_dir}/.password", "--skip-build", "--skip-cleanup", "--skip-backup", "-e", f"@{_SWARM_EXTRAS_VARS}", "-e", f"@{extras_path}", ] return run_step(env=env, cmd=cmd, label="deploy backup host (backup.yml)") def _converge_and_verify(*, app_id: str) -> int: env = os.environ.copy() env["APP_ID"] = app_id rc = run_step( ["bash", str(_SWARM_SCRIPTS / "03_wait_converge.sh")], env=env, label="wait for stack convergence", ) if rc != 0: return rc return run_step( ["bash", str(_SWARM_SCRIPTS / "04_verify_reachable.sh")], env=env, label="verify reachability", ) def _drill_env(*, app_id: str, inv_dir: str, extras_path: str) -> dict[str, str]: env = os.environ.copy() env["APP_ID"] = app_id env["INFINITO_INVENTORY_DIR"] = inv_dir env["DRILL_EXTRAS"] = extras_path env["DISK_FLOOR_MB"] = str(DISK_FLOOR_MB) return env def _drill_is_coming(*, app_id: str, inv_dir: str, extras_path: str) -> bool: env = _drill_env(app_id=app_id, inv_dir=inv_dir, extras_path=extras_path) env["DRILL_PROBE"] = "true" probe = subprocess.run( ["bash", str(_SWARM_SCRIPTS / "backup" / "base.sh")], env=env, check=False, capture_output=True, text=True, ) if "DRILL=no" in probe.stdout: return False if "DRILL=yes" not in probe.stdout: print(f"=== drill probe gave no verdict for {app_id}; assuming it drills ===") return True def _backup_restore_drill(*, app_id: str, inv_dir: str, extras_path: str) -> int: return run_step( ["bash", str(_SWARM_SCRIPTS / "backup" / "base.sh")], env=_drill_env(app_id=app_id, inv_dir=inv_dir, extras_path=extras_path), label="backup + restore DR drill", ) def _backup_phase( *, app_id: str, inv_dir: str, deploy_extras: str, drill_extras: str ) -> int: rc = _deploy_backup_host(app_id=app_id, inv_dir=inv_dir, extras_path=deploy_extras) if rc == 0: rc = _backup_restore_drill( app_id=app_id, inv_dir=inv_dir, extras_path=drill_extras ) return rc def _verify_recovered_marker(*, app_id: str) -> int: env = os.environ.copy() env["APP_ID"] = app_id return run_step( ["bash", str(_SWARM_SCRIPTS / "backup" / "verify_recovered_marker.sh")], env=env, label="verify recovered marker (post update pass)", ) def _mesh_prologue(*, app_id: str, inv_dir: str, round_variants: dict[str, int]) -> int: """Put every node and the controller on the mesh, then move the transport. Args: app_id: primary application of the round. inv_dir: inventory directory of the round. round_variants: the round's ``{app_id: variant_index}`` map. """ rc = bootstrap_mesh(inv_dir=inv_dir) if rc == 0: rc = lab_sudo_without_prompt(inv_dir=inv_dir) if rc == 0: rc = _reset_credentials( app_id=app_id, inv_dir=inv_dir, round_variants=round_variants ) if rc == 0: rc = write_mesh(inv_dir=inv_dir, rotate=True) if rc == 0: rc = converge_mesh(inv_dir=inv_dir) if rc == 0: rc = mesh_controller(inv_dir=inv_dir) if rc == 0: rc = switch_to_mesh_transport(inv_dir=inv_dir) return rc def _purge(*, purge_set: tuple[str, ...]) -> int: if not purge_set: return 0 env = os.environ.copy() env["apps"] = ",".join(purge_set) return run_step( ["bash", str(_SWARM_DIR / "utils" / "clean" / "purge_stacks.sh")], env=env, label=f"purge prior-round stacks ({', '.join(purge_set)})", ) def _parse_args(argv: list[str] | None) -> argparse.Namespace: from cli.administration.deploy.development.variant_select import add_variant_args p = argparse.ArgumentParser( prog="utils.tests.swarm.matrix", description=( "Iterate the variant-matrix rounds of one application against the " "live swarm test cluster." ), ) p.add_argument( "--id", "--app", dest="app", default=os.environ.get("APP_ID"), help="Primary application id (default: $APP_ID).", ) p.add_argument( "--inventory-dir", default=os.environ.get( "INFINITO_INVENTORY_DIR", _DEFAULT_INVENTORY_DIR ), # nocheck: swarm-test base; matrix sets it per round, compose resolves the key via its own handler help=( "Base inventory dir; the planner derives per-round folders " "