94 lines
2.8 KiB
Python
Executable File
94 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Plan and validate canary promotion gates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
CHECKS_FILE = REPO_ROOT / "infra" / "promotion" / "checks.tsv"
|
|
|
|
|
|
def checks() -> list[dict[str, str]]:
|
|
with CHECKS_FILE.open(encoding="utf-8", newline="") as handle:
|
|
return list(csv.DictReader(handle, delimiter="\t"))
|
|
|
|
|
|
def plan() -> int:
|
|
print("Promotion Plan")
|
|
print("==============")
|
|
print("dev branch -> validation -> main -> Argo CD sync")
|
|
print()
|
|
for row in checks():
|
|
marker = "required" if row["required"] == "true" else "optional-live"
|
|
print(f"{marker:13} {row['stage']:18} {row['command']}")
|
|
print()
|
|
print("Prod promotion remains manual until dev namespace/app wiring is added.")
|
|
return 0
|
|
|
|
|
|
def validate(local_only: bool) -> int:
|
|
failures = 0
|
|
print("Promotion Validation")
|
|
print("====================")
|
|
for row in checks():
|
|
is_required = row["required"] == "true"
|
|
if local_only and not is_required:
|
|
print(f"skip {row['stage']} - live check gated")
|
|
continue
|
|
print(f"run {row['stage']} - {row['command']}")
|
|
process = subprocess.run(
|
|
row["command"].split(),
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
check=False,
|
|
)
|
|
if process.returncode == 0:
|
|
print("ok")
|
|
else:
|
|
failures += 1
|
|
print(process.stdout.rstrip())
|
|
print(f"failures={failures}")
|
|
return 1 if failures else 0
|
|
|
|
|
|
def rollback(app: str) -> int:
|
|
print("Rollback Plan")
|
|
print("=============")
|
|
print(f"app: {app}")
|
|
print("1. identify last known good Git commit or image digest")
|
|
print("2. git revert <bad-commit>")
|
|
print("3. ./jeannie safety-case --since HEAD~1")
|
|
print("4. ./jeannie validate")
|
|
print("5. push to main and let Argo CD reconcile")
|
|
print("6. ./jeannie synthetic-checks")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
subparsers.add_parser("plan")
|
|
validate_parser = subparsers.add_parser("validate")
|
|
validate_parser.add_argument("--all", action="store_true", help="Include optional live checks.")
|
|
rollback_parser = subparsers.add_parser("rollback")
|
|
rollback_parser.add_argument("app")
|
|
args = parser.parse_args()
|
|
if args.command == "plan":
|
|
return plan()
|
|
if args.command == "validate":
|
|
return validate(local_only=not args.all)
|
|
if args.command == "rollback":
|
|
return rollback(args.app)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|