#!/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 try: import yaml except ModuleNotFoundError: raise SystemExit("PyYAML 필요 — 'pip install pyyaml'(또는 'pip3 install --break-system-packages pyyaml') 후 다시 실행하세요.") URL = os.environ["YAKCLOUD_URL"].rstrip("/") TOKEN = os.environ["YAKCLOUD_TOKEN"] CLUSTER_REF = os.environ.get("YAKCLOUD_CLUSTER") CLUSTER = "" # main 에서 이름→id 로 해석해 채운다 CLUSTER_HOST = "" # 클러스터 기본 도메인 — 항상 노출(운영은 여기 + 운영 도메인) ENV = os.environ.get("YAKCLOUD_ENV", "dev") # 대상 환경 environments. (기본 dev). promote=prod ENV_DOMAINS: list = [] # main 에서 environments..domains 로 채움(운영 도메인) 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) hint = { 401: " — 배포 토큰(YAKCLOUD_TOKEN)이 없거나 만료/오류", 403: " — 이 토큰 권한으로는 불가(엔드포인트 권한 확인)", 404: " — 경로의 id/이름 확인", 405: " — 이 배포 토큰/엔드포인트로 허용되지 않는 작업(콘솔에서 처리 필요)", }.get(e.code, "") raise SystemExit(f"[api] {method} {path} -> {e.code}{hint}: {payload[:300]}") 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 _ready_sources() -> str: return ", ".join(s.get("name", "?") for s in cluster_services() if s.get("status") == "READY") 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"] mode = req.get("mode", "shared") # shared(기본)=공유 외부 서버에 격리 DB 민팅 · local=인클러스터 프로비저닝(옵션, size 사용) if DRY: if match: log(f"source '{name}' 상태={match.get('status')} — READY 대기 필요") else: plan_or_mode = "shared" if mode == "shared" else plan log(f"source '{name}' ({stype}, {plan_or_mode}) 없음 → 프로비저닝 예정(POST /services, mode={mode}). 현재 READY: {_ready_sources()}") return None if not match: log(f"source '{name}' ({stype}, {'shared' if mode=='shared' else plan}) 프로비저닝 시도… (mode={mode})") try: # 소스 생성 = POST /services (clusterId 는 body). /clusters/{id}/services 는 GET 전용. # local=인클러스터 프로비저닝(size 사용) · shared=공유 외부(size 무관, 백엔드가 격리 DB/계정 민팅). body = {"clusterId": CLUSTER, "type": stype, "name": name, "mode": mode} if mode == "local": body["size"] = plan api("POST", "/services", body) except SystemExit as e: raise SystemExit( f"소스 '{name}' 생성 실패 — 콘솔에서 소스를 만든 뒤 requires[].name·binds[].source 를 그 소스명으로 " f"지정하거나 다시 시도하세요.\n 현재 READY 소스: {_ready_sources()}\n (원인: {e})") 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 reconcile_domain(fqdn: str) -> None: """워크로드 expose host 를 도메인 레지스트리에 리컨실 — 없으면 등록. 관리형 도메인(yakenator.io/openrepublic.club/sapiens.inc 등)은 즉시 ACTIVE, 외부는 TXT 검증 안내.""" if not fqdn or fqdn == CLUSTER_HOST: return # 클러스터 기본 도메인은 이미 등록·라우팅됨 doms = unwrap(api("GET", f"/clusters/{CLUSTER}/domains")) or [] if any(d.get("fqdn") == fqdn for d in doms): log(f"domain '{fqdn}' 이미 등록됨") return if DRY: log(f"domain '{fqdn}' 없음 → 등록 예정(관리형=즉시 ACTIVE, 외부=TXT 검증)") return try: d = unwrap(api("POST", f"/clusters/{CLUSTER}/domains", {"fqdn": fqdn})) except SystemExit as e: log(f"⚠ domain '{fqdn}' 등록 실패 — {e}") return log(f"domain '{fqdn}' 등록: status={d.get('status')} cert={d.get('certStatus')}") v = d.get("verify") if v: log(f' 외부 도메인 — DNS 에 TXT 추가 후 검증: {v["host"]} TXT "{v["value"]}"') def _repo(img: str) -> str: """이미지에서 :tag 제거한 repo(레지스트리 포트 host:5000/… 의 콜론은 보존).""" seg = img.rsplit("/", 1)[-1] return img.rsplit(":", 1)[0] if ":" in seg else img 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)), } # 노출 도메인 = 워크로드 지정 host(s) + 환경 도메인(운영). 커스텀이 있으면 기본 도메인도 함께 노출. wl_hosts = ex.get("hosts") or ([ex["host"]] if ex.get("host") else []) custom = list(dict.fromkeys(wl_hosts + ENV_DOMAINS)) if custom: body["exposeHosts"] = list(dict.fromkeys(custom + ([CLUSTER_HOST] if CLUSTER_HOST else []))) else: body["exposeHost"] = CLUSTER_HOST or None # 환경변수(선언적): 리스트[{key,value,secret?}] 또는 맵{KEY: VALUE}. env = w.get("env") or [] if isinstance(env, dict): env = [{"key": k, "value": v} for k, v in env.items()] body["env"] = [{"key": e["key"], "value": str(e.get("value", "")), "secret": bool(e.get("secret", False))} for e in env] # 충돌 가드 — 같은 이름의 기존 배포가 '다른 앱'(다른 이미지 repo)이면 덮어쓰기 거부(무조건 배포 금지). # 같은 repo(태그만 다름)=정상 재배포는 PATCH 로 진행. (백엔드 create_deployment 도 동일하게 409 로 강제.) existing = next((d for d in cluster_deployments() if d.get("name") == w["name"]), None) if existing: ex_repo = _repo(existing.get("image") or "") if ex_repo and ex_repo != _repo(image): raise SystemExit( f"배포명 '{w['name']}' 충돌 — 이 클러스터에 이미 다른 앱이 그 이름을 쓰고 있습니다.\n" f" 기존 image: {existing.get('image')}\n 내 image: {image}\n" f" → 워크로드 이름을 바꾸거나(yakcloud set 로 name 변경) 기존 배포를 먼저 정리하세요.") # 도메인 리컨실 — 워크로드 host(들) + 환경(운영) 도메인 등록(없으면). 충돌 통과 뒤에만. for h in custom: reconcile_domain(h) if DRY: note = "기존 동일 앱 → PATCH 롤링" if existing else "신규 생성" log(f"deploy '{w['name']}' 예정({note}): image={image} port={body['port']} path={body['path']} " f"replicas={body['replicasDesired']} hosts={body.get('exposeHosts') or body.get('exposeHost')}") return None, False 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, ENV_DOMAINS path = next((a for a in sys.argv[1:] if not a.startswith("--")), "yakcloud.yaml") m = yaml.safe_load(open(path)) # 환경 해석: environments..{cluster,domains}. 하위호환: 없으면 top-level cluster:. envs = m.get("environments") or {} env_cfg = envs.get(ENV) or {} # environments 의 env 클러스터가 최우선(저장/전역 YAKCLOUD_CLUSTER 는 하위호환 폴백만). ref = env_cfg.get("cluster") or CLUSTER_REF or m.get("cluster") if not ref: raise SystemExit( f"환경 '{ENV}' 대상 클러스터 미지정 — 매니페스트 environments.{ENV}.cluster (또는 YAKCLOUD_CLUSTER). " f"등록된 환경: {', '.join(envs) or '(없음)'}") ENV_DOMAINS = env_cfg.get("domains") or [] CLUSTER, cname = resolve_cluster(ref) CLUSTER_HOST = (unwrap(api("GET", f"/clusters/{CLUSTER}")) or {}).get("defaultHostname") or "" dom = f" +도메인 {ENV_DOMAINS}" if ENV_DOMAINS else "" log(f"[{ENV}] project '{m.get('project')}' → cluster '{cname}' ({CLUSTER}){dom} (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()