#!/usr/bin/env python3 """yakcloud deploy — 매니페스트(yakcloud.yaml) 기반 리컨실 배포 CLI (콘솔 API). 흐름(선언적 idempotent): 1) requires 리컨실 — 논리 이름의 소스가 READY 면 스킵, 없으면 type/plan 으로 프로비저닝 → READY 대기. 2) workloads 배포 — 각 워크로드 배포(POST .../deployments). 기존이면 PATCH(무중단 롤링). 3) 바인딩 — 각 bind(alias→source) 를 POST /services/{serviceId}/bindings {deploymentId, alias}. 인증: 콘솔 API 에 개인 배포 토큰(PAT) Bearer. (환경변수) YAKCLOUD_URL 예) https://console.yakenator.io YAKCLOUD_TOKEN 배포 토큰(PAT) — 콘솔 설정에서 발급 YAKCLOUD_CLUSTER 대상 클러스터 이름 또는 콘솔 id (없으면 매니페스트 cluster:) TAG 이미지 태그 치환용(${TAG}); 없으면 latest 사용: yakcloud_deploy.py [manifest.yaml] [--dry-run] """ from __future__ import annotations import json import os import sys import time import urllib.error import urllib.request import yaml URL = os.environ["YAKCLOUD_URL"].rstrip("/") TOKEN = os.environ["YAKCLOUD_TOKEN"] CLUSTER_REF = os.environ.get("YAKCLOUD_CLUSTER") CLUSTER = "" # main 에서 이름→id 로 해석해 채운다 CLUSTER_HOST = "" # 클러스터 기본 도메인 — expose.host(s) 미지정 시 여기로 노출 TAG = os.environ.get("TAG", "latest") DRY = "--dry-run" in sys.argv[1:] def api(method: str, path: str, body: dict | None = None, _retry: int = 5) -> dict: data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(f"{URL}/api/v1{path}", data=data, method=method, headers={"Authorization": f"Bearer {TOKEN}", "content-type": "application/json"}) try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read().decode() return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: payload = e.read().decode() if e.code == 429 and _retry > 0: # 쓰기 레이트리밋 — retryAfterSec 백오프 후 재시도 try: wait = json.loads(payload).get("error", {}).get("detail", {}).get("retryAfterSec", 2) except Exception: # noqa: BLE001 wait = 2 time.sleep(max(1, int(wait)) + 1) return api(method, path, body, _retry - 1) raise SystemExit(f"[api] {method} {path} -> {e.code}: {payload[:400]}") def unwrap(r): return r.get("data", r) if isinstance(r, dict) else r def log(m: str) -> None: print(("\033[33m[dry]\033[0m " if DRY else "\033[36m▸\033[0m ") + m, flush=True) def resolve_cluster(ref: str) -> tuple[str, str]: clusters = unwrap(api("GET", "/clusters")) or [] for c in clusters: if ref in (c.get("id"), c.get("name")): return c["id"], c.get("name") or c["id"] names = ", ".join(c.get("name", "?") for c in clusters) or "(없음)" raise SystemExit(f"클러스터 '{ref}' 를 찾을 수 없습니다. 계정 클러스터: {names}") def cluster_services() -> list[dict]: return unwrap(api("GET", f"/clusters/{CLUSTER}/services")) or [] def cluster_deployments() -> list[dict]: return unwrap(api("GET", f"/clusters/{CLUSTER}/deployments")) or [] def reconcile_source(req: dict) -> str | None: name, stype, plan = req["name"], req["type"].upper(), req.get("plan", "small") match = next((s for s in cluster_services() if s.get("name") == name), None) if match and match.get("status") == "READY": log(f"source '{name}' ({stype}) 이미 READY → 스킵 (id={match['id']})") return match["id"] if DRY: log(f"source '{name}' ({stype}, {plan}) 없음 → 프로비저닝 예정") return None if not match: log(f"source '{name}' ({stype}, {plan}) 프로비저닝…") api("POST", f"/clusters/{CLUSTER}/services", {"clusterId": CLUSTER, "type": stype, "name": name, "mode": "local", "size": plan}) for _ in range(120): # ~10분 s = next((s for s in cluster_services() if s.get("name") == name), None) if s and s.get("status") == "READY": log(f"source '{name}' READY (id={s['id']})") return s["id"] if s and s.get("status") == "ERROR": raise SystemExit(f"source '{name}' 프로비저닝 ERROR") time.sleep(5) raise SystemExit(f"source '{name}' READY 대기 초과") def deploy_workload(w: dict) -> tuple[str | None, bool]: image = w["image"].replace("${TAG}", TAG) ex = w.get("expose", {}) or {} res = w.get("resources", {}) or {} body = { "name": w["name"], "image": image, "port": w.get("port"), "replicasDesired": w.get("replicas", 1), "cpuRequest": res.get("cpu", "25m"), "memRequest": res.get("mem", "96Mi"), "healthPath": w.get("health"), "path": ex.get("path", "/"), "pathType": "Prefix", "rewritePrefix": bool(ex.get("rewrite", False)), } # 노출 도메인: 여러 개(hosts) > 단일(host) > 클러스터 기본 도메인. hosts = ex.get("hosts") if hosts: body["exposeHosts"] = hosts else: body["exposeHost"] = ex.get("host") or (CLUSTER_HOST or None) if DRY: log(f"deploy '{w['name']}' 예정: image={image} port={body['port']} path={body['path']} " f"replicas={body['replicasDesired']} hosts={hosts or body.get('exposeHost')}") return None, False # 멱등: 기존 배포면 PATCH(이미지 갱신 → kubectl apply = 무중단 롤링). 없으면 신규 생성. existing = next((d for d in cluster_deployments() if d.get("name") == w["name"]), None) if existing: log(f"deploy '{w['name']}' 기존 존재 → PATCH 롤링 업데이트 image={image}") api("PATCH", f"/deployments/{existing['id']}", body) return existing["id"], False log(f"deploy '{w['name']}' 신규 생성 image={image}") dep = unwrap(api("POST", f"/clusters/{CLUSTER}/deployments", body)) return dep.get("id"), True def bind(dep_id: str | None, alias: str, service_id: str | None, source: str) -> None: if DRY or not dep_id or not service_id: log(f"bind '{alias}' → source '{source}' (serviceId={service_id}) 예정") return api("POST", f"/services/{service_id}/bindings", {"deploymentId": dep_id, "alias": alias}) log(f"bind '{alias}' → '{source}' 완료") def main() -> None: global CLUSTER, CLUSTER_HOST path = next((a for a in sys.argv[1:] if not a.startswith("--")), "yakcloud.yaml") m = yaml.safe_load(open(path)) ref = CLUSTER_REF or m.get("cluster") if not ref: raise SystemExit("대상 클러스터 미지정 — YAKCLOUD_CLUSTER 환경변수 또는 매니페스트 cluster: 필드") CLUSTER, cname = resolve_cluster(ref) CLUSTER_HOST = (unwrap(api("GET", f"/clusters/{CLUSTER}")) or {}).get("defaultHostname") or "" log(f"project '{m.get('project')}' → cluster '{cname}' ({CLUSTER}) (tag {TAG}){' [DRY-RUN]' if DRY else ''}") src_ids: dict[str, str | None] = {} for req in m.get("requires", []): src_ids[req["name"]] = reconcile_source(req) for w in m.get("workloads", []): dep_id, created = deploy_workload(w) if DRY or created: # 기존 배포는 PATCH(롤링)로 바인딩 유지 → 신규일 때만 바인딩 for b in w.get("binds", []): bind(dep_id, b["alias"], src_ids.get(b["source"]), b["source"]) log("완료 — 콘솔 앱 탭에서 배포/바인딩 확인" if not DRY else "완료(dry-run) — 실제 변경 없음") if __name__ == "__main__": main()