#!/usr/bin/env python3 # /// script # requires-python = ">=3.14" # dependencies = [] # /// import argparse import base64 import json import os import time import shlex import subprocess import sys def run_cmd(cmd, **kwargs): print(f"+ {' '.join(cmd)}", flush=True) return subprocess.run(cmd, **kwargs) def run_cmd_check(cmd, **kwargs): result = run_cmd(cmd, **kwargs) if result.returncode != 0: if result.stdout: print(result.stdout, flush=True) if result.stderr: print(result.stderr, file=sys.stderr, flush=True) sys.exit(result.returncode) return result def decode_services(services_json_b64): return json.loads(base64.b64decode(services_json_b64)) def trigger_workflow_and_get_run_id(workflow, repo, ref, params_b64): """Trigger a workflow via gh api with return_run_details=true and return the run ID.""" cmd = [ "gh", "api", f"repos/{repo}/actions/workflows/{workflow}/dispatches", "--method", "POST", "-H", "Accept: application/vnd.github+json", "-f", f"ref={ref}", "-F", "return_run_details=true", ] if params_b64: params = base64.b64decode(params_b64).decode() for token in shlex.split(params): if token.startswith("-"): cmd.append(token) elif "=" in token: key, value = token.split("=", 1) cmd.append(f"inputs[{key}]={value}") else: cmd.append(token) result = run_cmd(cmd, capture_output=True, text=True) if result.returncode != 0: print(result.stderr, file=sys.stderr) sys.exit(result.returncode) data = json.loads(result.stdout) return str(data["workflow_run_id"]) def wait_workflow_success(run_id, repo, interval=10): time.sleep(interval) # sleep to wait workflow run ready run_cmd_check( ["gh", "run", "watch", run_id, "--repo", repo, "--interval", str(interval), "--exit-status"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def wait_for_services_ready(services_json_b64): timeout = os.environ.get("JOB_TIMEOUT", "1800") services = decode_services(services_json_b64) for svc in services: name = svc["name"] namespace = svc["namespace"] wait_for = svc.get("wait_for", "") if wait_for == "daemonset-ready": resource = name.split("/", 1)[1] if name.startswith("daemonset/") else name deadline = time.time() + int(timeout) while True: result = run_cmd( ["kubectl", "get", "daemonset", resource, "-n", namespace, "-ojson"], capture_output=True, text=True, ) if result.returncode != 0: if time.time() >= deadline: print(result.stderr, file=sys.stderr, flush=True) sys.exit(result.returncode) else: status = json.loads(result.stdout).get("status", {}) desired = status.get("desiredNumberScheduled", 0) if ( desired > 0 and status.get("updatedNumberScheduled") == desired and status.get("numberAvailable") == desired and status.get("numberUnavailable", 0) == 0 ): print(f"daemonset/{resource} is ready on {desired} nodes", flush=True) break if time.time() >= deadline: print(f"timed out waiting for daemonset/{resource} to be ready", file=sys.stderr, flush=True) sys.exit(1) time.sleep(10) continue print(f"wait service {name} to be created", flush=True) run_cmd_check( ["kubectl", "wait", "--for", "create", f"--timeout={timeout}s", name, "-n", namespace], ) print(f"service {name} has been created", flush=True) if wait_for: result = run_cmd( ["kubectl", "get", name, "-n", namespace, "-ojsonpath={.spec.replicas}"], capture_output=True, text=True, ) if result.returncode != 0: sys.exit(result.returncode) replicas = result.stdout.strip() # Strip single quotes: shell transparently removes them (e.g. jsonpath='{.foo}'=1 # becomes jsonpath={.foo}=1), but subprocess passes them literally to kubectl, # causing "unexpected path string" errors. wait_for = wait_for.replace("REPLICAS", replicas).replace("'", "") print(f"wait service {name} to meet condition: {wait_for}", flush=True) run_cmd_check( ["kubectl", "wait", "--for", wait_for, f"--timeout={timeout}s", name, "-n", namespace], ) def wait_for_services_gone(services_json_b64): timeout = os.environ.get("JOB_TIMEOUT", "1800") services = decode_services(services_json_b64) for svc in services: name = svc["name"] namespace = svc["namespace"] print(f"wait service {name} to be deleted", flush=True) run_cmd_check( ["kubectl", "wait", "--for", "delete", f"--timeout={timeout}s", name, "-n", namespace], ) print(f"service {name} has been deleted", flush=True) def cmd_deploy(args): print(f"run workflow {args.workflow} in {args.repo}({args.ref})", flush=True) run_id = trigger_workflow_and_get_run_id(args.workflow, args.repo, args.ref, args.params_b64) if args.wait_workflow_success == "true": wait_workflow_success(run_id, args.repo) wait_for_services_ready(args.services_json_b64) def cmd_check_destroy(args): if args.check_before_destroy == "false": print("skip checking residual k8s resources before destroy", flush=True) return services = decode_services(args.services_json_b64) for svc in services: name = svc["name"] namespace = svc["namespace"] result = run_cmd( ["kubectl", "get", name, "-n", namespace, "--ignore-not-found"], capture_output=True, text=True, ) if result.returncode != 0: # Treat "NotFound" errors (namespace or resource missing) as success: # ArgoCD cascade-deletes namespaces along with their resources, so # --ignore-not-found may not suppress the namespace-not-found error on # older kubectl versions. if "NotFound" in result.stderr or "not found" in result.stderr.lower(): print(f"service {name} namespace or resource gone, skipping", flush=True) continue if result.stderr: print(result.stderr, file=sys.stderr, flush=True) sys.exit(result.returncode) if result.stdout.strip(): print(f"service {name} is still in cluster", flush=True) sys.exit(1) def cmd_finalize(args): print(f"run workflow {args.workflow} in {args.repo}({args.ref})", flush=True) run_id = trigger_workflow_and_get_run_id(args.workflow, args.repo, args.ref, args.params_b64) if args.wait_workflow_success == "true": wait_workflow_success(run_id, args.repo) wait_for_services_gone(args.services_json_b64) def main(): parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) deploy_parser = subparsers.add_parser("deploy") deploy_parser.add_argument("--workflow", required=True) deploy_parser.add_argument("--repo", required=True) deploy_parser.add_argument("--ref", required=True) deploy_parser.add_argument("--params-b64", dest="params_b64", default="") deploy_parser.add_argument("--wait-workflow-success", dest="wait_workflow_success", default="false") deploy_parser.add_argument("--services-json-b64", dest="services_json_b64", required=True) check_parser = subparsers.add_parser("check-destroy") check_parser.add_argument("--check-before-destroy", dest="check_before_destroy", default="true") check_parser.add_argument("--services-json-b64", dest="services_json_b64", required=True) finalize_parser = subparsers.add_parser("finalize") finalize_parser.add_argument("--workflow", required=True) finalize_parser.add_argument("--repo", required=True) finalize_parser.add_argument("--ref", required=True) finalize_parser.add_argument("--params-b64", dest="params_b64", default="") finalize_parser.add_argument("--wait-workflow-success", dest="wait_workflow_success", default="false") finalize_parser.add_argument("--services-json-b64", dest="services_json_b64", required=True) args = parser.parse_args() if args.command == "deploy": cmd_deploy(args) elif args.command == "check-destroy": cmd_check_destroy(args) elif args.command == "finalize": cmd_finalize(args) if __name__ == "__main__": main()