yakenator/yakcloud-starter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cli): clusterName 환경별 소스명 매핑 + 운영 소스 3-모드 결정 조회(datasource options/ls) + 검증·문서 · v0.32.0

Browse Source
This commit is contained in:
jungwoo choi
2026-09-06 03:41:17 +09:00
parent 789aaee5b4
commit 075d7bc3d7
9 changed files with 356 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
.yakcloud/ctl.py
View File

@ -78,6 +78,28 @@ def resolve_cluster(ref: str) -> dict:
sys.exit(" ✗ 클러스터 '%s' 없음" % ref)
def resolve_requires(m: dict, env: str | None = None) -> list:
"""base + environments.<env> 오버레이(이름 매칭) 병합된 requires — deploy/dev 의 resolve_env 와 규약 일치.
실제 소스 이름(clusterName)·type/mode/plan 이 이 병합으로 환경별 결정된다(binds 는 논리 이름 고정)."""
env = env or os.environ.get("YAKCLOUD_ENV", "dev")
ec = (m.get("environments") or {}).get(env) or {}
by: dict = {}
order: list = []
for r in (m.get("requires") or []):
by[r["name"]] = dict(r); order.append(r["name"])
for r in (ec.get("requires") or []):
if r["name"] in by:
by[r["name"]].update(r)
else:
by[r["name"]] = dict(r); order.append(r["name"])
return [by[n] for n in order]
def actual_source(r: dict) -> str:
"""이 소스가 대상 클러스터에서 갖는 실제 이름(clusterName 우선, 없으면 논리 name)."""
return r.get("clusterName") or r["name"]
def find_workload(m: dict, name: str | None) -> dict:
ws = m.get("workloads", []) or []
if name:
@ -111,19 +133,23 @@ def patch_live(cid: str, name: str, patch: dict) -> bool:
def cmd_info(_a) -> None:
m = load_manifest()
print("project : %s" % m.get("project", "-"))
ref = m.get("cluster") or os.environ.get("YAKCLOUD_CLUSTER") or "-"
# 대상 클러스터 = environments.<env>.cluster 우선(cluster_ref 규약과 동일) — 미지정이면 None 폴백.
# 레거시 top-level cluster: 만 보던 게이트는 환경-온리 템플릿에서 라이브 상태가 항상 죽어 clusterName 진단이 안 뜸.
env = os.environ.get("YAKCLOUD_ENV", "dev")
ref = (((m.get("environments") or {}).get(env) or {}).get("cluster")
or os.environ.get("YAKCLOUD_CLUSTER") or m.get("cluster"))
deps: dict[str, dict] = {}
svcs: dict[str, dict] = {}
doms: list[dict] = []
cinfo = None
if URL and TOK and (m.get("cluster") or os.environ.get("YAKCLOUD_CLUSTER")):
cinfo = resolve_cluster(cluster_ref(m))
if URL and TOK and ref:
cinfo = resolve_cluster(ref)
deps = live_deps(cinfo["id"])
svcs = live_services(cinfo["id"])
doms = unwrap(api("GET", "/clusters/%s/domains" % cinfo["id"])) or []
print("cluster : %s (id %s) 도메인 %s" % (cinfo.get("name"), cinfo["id"], cinfo.get("defaultHostname", "-")))
else:
print("cluster : %s (라이브 상태는 YAKCLOUD_URL/TOKEN 설정 시 표시)" % ref)
print("cluster : %s (라이브 상태는 YAKCLOUD_URL/TOKEN + 대상 클러스터 설정 시 표시)" % (ref or "-"))
print("\nworkloads:")
for w in m.get("workloads", []) or []:
@ -144,13 +170,15 @@ def cmd_info(_a) -> None:
line += "\n binds: %s" % ", ".join("%s→%s" % (b["alias"], b["source"]) for b in binds)
print(line)
reqs = m.get("requires", []) or []
reqs = resolve_requires(m) # 현재 환경(YAKCLOUD_ENV, 기본 dev) 오버레이 반영 — clusterName·mode 등
if reqs:
print("\ndata sources (requires):")
for r in reqs:
s = svcs.get(r["name"])
actual = actual_source(r)
s = svcs.get(actual) # 라이브 상태는 '실제 소스 이름'으로 조회
st = (" [%s]" % s.get("status")) if s else ""
print(" • %-14s %s/%s%s" % (r["name"], r.get("type"), r.get("plan", "small"), st))
mp = (" → %s" % actual) if actual != r["name"] else "" # 논리→실제 매핑 표시
print(" • %-14s%s %s/%s%s" % (r["name"], mp, r.get("type"), r.get("plan", "small"), st))
if doms:
print("\ndomains (cluster):")
for d in doms:
@ -262,17 +290,42 @@ def cmd_source(a) -> None:
print(" ✓ requires 에서 제거: %s (관련 binds 정리). 'yakcloud deploy' 로 반영" % a.name)
def _ls_ref(m: dict, env_opt: str | None, cluster_opt: str | None) -> str:
"""소스 조회 대상 클러스터 ref — --cluster(명시) > --env(environments.<env>.cluster) > 현재 환경(cluster_ref)."""
if cluster_opt:
return cluster_opt
if env_opt:
ref = ((m.get("environments") or {}).get(env_opt) or {}).get("cluster")
if not ref:
sys.exit(" ✗ environments.%s.cluster 미지정 — --cluster <이름/id> 로 직접 지정하거나 매니페스트에 추가하세요." % env_opt)
return ref
return cluster_ref(m)
def cmd_source_ls(a) -> None:
"""클러스터에서 이용 가능한 데이터 소스 목록 — 이름·타입·내부/외부·상태·플랜."""
"""대상 클러스터에서 이용 가능한 데이터 소스 목록 — 이름·타입·내부/외부·상태·플랜.
운영 소스를 clusterName 으로 고를 때: 'yakcloud datasource ls --env prod' 로 실제 이름을 확인한다.
(로컬 dev 는 도커로 논리 이름 그대로 뜨므로 조회 불필요 — 이 명령은 주로 운영/검증 소스 선택용.)"""
need_api()
m = (yaml.safe_load(open(MANIFEST)) if os.path.exists(MANIFEST) else {}) or {}
c = resolve_cluster(cluster_ref(m))
env_opt = getattr(a, "env", None)
c = resolve_cluster(_ls_ref(m, env_opt, getattr(a, "cluster", None)))
svcs = unwrap(api("GET", "/clusters/%s/services" % c["id"])) or []
# ✓ 표시·매핑 = 그 환경의 병합 requires 기준(clusterName 우선) — 클러스터 서비스 이름과 일치.
used = {actual_source(r) for r in resolve_requires(m, env_opt)}
if getattr(a, "json", False): # 에이전트용 기계판독 출력
out = [{"name": s.get("name"), "type": s.get("type"), "mode": s.get("mode"),
"status": s.get("status"), "plan": s.get("size") or s.get("plan"),
"inRequires": s.get("name") in used} for s in svcs]
print(json.dumps({"cluster": c.get("name") or c["id"], "clusterId": c["id"],
"env": env_opt, "sources": out}, ensure_ascii=False, indent=2))
return
if not svcs:
print("클러스터 '%s' — 데이터 소스 없음. 콘솔 '데이터 소스' 또는 'yakcloud source add' 로 추가."
% (c.get("name") or c["id"]))
return
used = {r.get("name") for r in (m.get("requires", []) or [])}
rows = []
for s in sorted(svcs, key=lambda x: (x.get("type", ""), x.get("name", ""))):
mode = s.get("mode")
@ -282,16 +335,74 @@ def cmd_source_ls(a) -> None:
s.get("status", "?"), s.get("size") or s.get("plan") or "-"))
wn = max([len(r[1]) for r in rows] + [4])
wt = max([len(r[2]) for r in rows] + [4])
print("클러스터 '%s' 데이터 소스 (%d)" % (c.get("name") or c["id"], len(rows)))
tag = (" [env=%s]" % env_opt) if env_opt else ""
print("클러스터 '%s'%s 데이터 소스 (%d)" % (c.get("name") or c["id"], tag, len(rows)))
print(" %-*s %-*s %-4s %-8s 플랜" % (wn, "이름", wt, "타입", "위치", "상태"))
for mark, name, typ, loc, st, plan in rows:
print(" %s %-*s %-*s %-4s %-8s %s" % (mark, wn, name, wt, typ, loc, st, plan))
print("\n ✓=매니페스트 requires 에 이미 있음 · 바인딩: yakcloud bind <workload> <이름> <alias>")
print("\n ✓=이 환경 requires 에 이미 있음(clusterName 우선) · 바인딩: yakcloud bind <workload> <이름> <alias>")
print(" 이 소스를 논리 이름에 매핑: environments.%s.requires 에 { name: <논리>, clusterName: <위 이름> }" % (env_opt or "prod"))
def cmd_source_options(a) -> None:
"""운영 소스 '모드 결정'용 정보 — 타입별 지원 모드(local/remote/shared) + 이 배포의 공유 가용성 + 기존 소스.
예) 'yakcloud datasource options postgresql --env prod' → shared 가능 여부·기존 소스로 모드를 고른다.
(로컬 dev 는 도커로 뜨므로 이 결정은 주로 운영/검증용.)"""
need_api()
m = (yaml.safe_load(open(MANIFEST)) if os.path.exists(MANIFEST) else {}) or {}
stype = a.type.lower()
types = (unwrap(api("GET", "/source-options")) or {}).get("types") or {}
info = types.get(stype)
if not info:
sys.exit(" ✗ 알 수 없는 타입 '%s' — %s 중 하나" % (stype, ", ".join(sorted(types)) or "?"))
modes = info.get("modes") or []
shared_ok = bool(info.get("sharedAvailable"))
# 대상 클러스터의 같은 타입 기존 소스(있으면 clusterName 으로 붙일 후보)
env_opt = getattr(a, "env", None)
existing = []
existing_error = None
# 기존 소스 조회는 best-effort(모드 정보는 항상 출력). 단 '빈 목록'과 '조회 실패'를 반드시 구분해 노출한다
# — 삼키면 에이전트가 '소스 없음 → 신규 생성'으로 오판(있는데 중복 생성)할 수 있다.
try:
c = resolve_cluster(_ls_ref(m, env_opt, getattr(a, "cluster", None)))
svcs = unwrap(api("GET", "/clusters/%s/services" % c["id"])) or []
existing = [{"name": s.get("name"), "mode": s.get("mode"), "status": s.get("status")}
for s in svcs if str(s.get("type", "")).lower() == stype]
except SystemExit as e:
existing_error = (str(e).strip().lstrip("✗").strip() or "클러스터/소스 조회 실패")
except Exception as e: # noqa: BLE001 — 네트워크 오류 등도 조회만 실패시키고 모드 정보는 출력
existing_error = "조회 오류: %s" % e
if getattr(a, "json", False): # 에이전트용
out = {"type": stype, "modes": modes, "sharedAvailable": shared_ok, "existing": existing}
if existing_error:
out["existingError"] = existing_error # '빈 목록' vs '조회 실패' 구분 — 오판 방지
print(json.dumps(out, ensure_ascii=False, indent=2))
return
print("데이터 소스 '%s' — 운영 소스 모드 선택지" % stype)
print(" 지원 모드: %s" % ", ".join(modes))
if "shared" in modes:
print(" · shared (공유) : %s" % ("가능 — 이 배포에 공유 서버 있음 → requires 에 { mode: shared } (격리 DB 민팅)"
if shared_ok else "이 배포에 공유 서버 없음 → 사용 불가(local/remote 로)"))
if "remote" in modes:
print(" · remote (외부) : 콘솔/API로 외부 접속정보 등록(자격은 매니페스트/git 금지) → clusterName 으로 참조")
if "local" in modes:
print(" · local (인클러스터): 전용 인스턴스 신규 생성 → requires 에 { mode: local, plan: small }")
if existing_error:
print(" ⚠ 기존 소스 조회 실패(%s) — 대상 클러스터/토큰 확인. 위 모드 정보는 유효(기존 소스 유무는 미확인)." % existing_error)
elif existing:
print(" 기존 소스(이 클러스터·같은 타입):")
for e in existing:
print(" - %-20s [%s] %s" % (e["name"], e.get("mode") or "?", e.get("status") or "?"))
print(" ↑ 기존 소스에 붙이려면 environments.<env>.requires 에 { name: <논리>, clusterName: <이름> }")
else:
print(" 기존 소스 없음(같은 타입) — 위 모드 중 골라 requires 에 선언")
def cmd_bind(a) -> None:
m = load_manifest(); w = find_workload(m, a.workload)
if not any(r.get("name") == a.source for r in (m.get("requires", []) or [])):
# 검증은 병합 뷰(base + 현재 환경 오버레이) 기준 — info/datasource ls 와 일관. binds 는 논리 이름 참조.
if not any(r.get("name") == a.source for r in resolve_requires(m)):
sys.exit(" ✗ requires 에 소스 '%s' 없음 — 'yakcloud source add %s <type>' 먼저" % (a.source, a.source))
binds = w.get("binds", []) or []; w["binds"] = binds
binds[:] = [b for b in binds if b.get("alias") != a.alias]
@ -377,11 +488,22 @@ def build_parser() -> argparse.ArgumentParser:
so = sub.add_parser("source", help="데이터 소스: ls(클러스터 목록) · add · rm")
sosub = so.add_subparsers(dest="action", required=True)
sol = sosub.add_parser("ls", help="클러스터에서 이용 가능한 소스 목록"); sol.set_defaults(fn=cmd_source_ls)
sol = sosub.add_parser("ls", help="클러스터 소스 목록(운영 조회 --env prod, 에이전트 --json)")
sol.add_argument("--env", choices=["dev", "val", "prod"], help="대상 환경(environments.<env>.cluster) — 기본=현재 환경")
sol.add_argument("--cluster", help="대상 클러스터 이름/id 직접 지정(--env 무시)")
sol.add_argument("--json", action="store_true", help="기계판독 JSON 출력(에이전트용)")
sol.set_defaults(fn=cmd_source_ls)
soo = sosub.add_parser("options", help="타입별 모드(local/remote/shared)+공유 가용성+기존 소스 — 운영 모드 결정용")
soo.add_argument("type", help="데이터 소스 타입(postgresql·mysql·mongodb·redis·minio·rabbitmq·solr·oracle·mariadb)")
soo.add_argument("--env", choices=["dev", "val", "prod"], help="대상 환경(기존 소스 조회용)")
soo.add_argument("--cluster", help="대상 클러스터 이름/id 직접 지정")
soo.add_argument("--json", action="store_true", help="기계판독 JSON 출력(에이전트용)")
soo.set_defaults(fn=cmd_source_options)
soa = sosub.add_parser("add"); soa.add_argument("name"); soa.add_argument("type")
soa.add_argument("plan", nargs="?", default="small")
soa.add_argument("--shared", action="store_true",
help="공유 외부 소스(mode:shared) — 백엔드가 클러스터별 격리 DB/계정 민팅(소용량 클러스터용, 현재 mongodb)")
help="공유 외부 소스(mode:shared) — 백엔드가 격리 DB/계정 민팅. 지원=pg·mysql·mariadb·mongodb·rabbitmq·solr "
"(배포별 가용성은 'yakcloud datasource options <type>' 로 확인)")
soa.set_defaults(fn=cmd_source)
sor = sosub.add_parser("rm"); sor.add_argument("name"); sor.set_defaults(fn=cmd_source)

88
.yakcloud/deploy.py
View File

@ -94,7 +94,13 @@ def _ready_sources() -> str:
def reconcile_source(req: dict) -> str | None:
name, stype, plan = req["name"], req["type"].upper(), req.get("plan", "small")
logical = req["name"]
# 실제 클러스터 소스 이름 = clusterName(오버레이로 환경별 교체) 없으면 논리 이름.
# 앱 binds 는 논리 이름 고정 — 로컬↔운영 소스명이 달라도 매니페스트 수정 없이 매핑된다.
# 매칭·프로비저닝·READY 대기는 전부 이 실제 이름(name) 기준. 반환 id 는 호출부가 논리 이름으로 키.
name = req.get("clusterName") or logical
disp = logical if name == logical else f"{logical}→{name}"
stype, plan = req["type"].upper(), req.get("plan", "small")
# 기본 mode: 대부분 shared(공유 외부에 격리 DB 민팅). 단 Redis 는 공유 격리가 ACL 키/채널 '프리픽스'(비투명 —
# 앱이 맨 키/채널로 pub/sub 하면 NOPERM)라 **기본을 전용(local)**으로 → REDIS_URL 로 전권 사용(pub/sub 포함).
# 명시 mode 는 존중. shared redis 는 프리픽스-인지 앱 전용(mode: shared 로 명시).
@ -107,22 +113,22 @@ def reconcile_source(req: dict) -> str | None:
if match and match.get("status") == "READY":
cur = str(match.get("mode") or "").lower()
if cur and cur != mode:
log(f"⚠ source '{name}' 은 이미 mode={cur} 로 존재 → 매니페스트 mode={mode} 는 무시(재프로비저닝 안 함). "
f"바꾸려면 삭제 후 재배포: 'yakcloud source rm {name}'.")
log(f"⚠ source '{name}' ({stype}) 이미 READY → 재사용·바인딩 (id={match['id']}). "
log(f"⚠ source '{disp}' 은 이미 mode={cur} 로 존재 → 매니페스트 mode={mode} 는 무시(재프로비저닝 안 함). "
f"바꾸려면 콘솔에서 라이브 소스 '{name}' 삭제 후 재배포(매니페스트 항목 제거는 'yakcloud source rm {logical}').")
log(f"⚠ source '{disp}' ({stype}) 이미 READY → 재사용·바인딩 (id={match['id']}). "
f"본인 프로젝트가 만든 소스면 정상이나, 공용 클러스터라면 '다른 프로젝트의 소스'일 수 있음 "
f"— 이 앱의 마이그레이션이 그 소스(DB)에 테이블을 만든다. 전용이 필요하면 매니페스트에서 "
f"프로젝트 고유 이름(예: <project>-db)으로 바꿔 재배포하라.")
return match["id"]
if DRY:
if match:
log(f"source '{name}' 상태={match.get('status')} — READY 대기 필요")
log(f"source '{disp}' 상태={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()}")
log(f"source '{disp}' ({stype}, {plan_or_mode}) 없음 → 프로비저닝 예정(POST /services, name={name}, mode={mode}). 현재 READY: {_ready_sources()}")
return None
if not match:
log(f"source '{name}' ({stype}, {'shared' if mode=='shared' else plan}) 프로비저닝 시도… (mode={mode})")
log(f"source '{disp}' ({stype}, {'shared' if mode=='shared' else plan}) 프로비저닝 시도… (name={name}, mode={mode})")
try:
# 소스 생성 = POST /services (clusterId 는 body). /clusters/{id}/services 는 GET 전용.
# local=인클러스터 프로비저닝(size 사용) · shared=공유 외부(size 무관, 백엔드가 격리 DB/계정 민팅).
@ -132,17 +138,17 @@ def reconcile_source(req: dict) -> str | None:
api("POST", "/services", body)
except SystemExit as e:
raise SystemExit(
f"소스 '{name}' 생성 실패 — 콘솔에서 소스를 만든 뒤 requires[].name·binds[].source 를 그 소스명으로 "
f"지정하거나 다시 시도하세요.\n 현재 READY 소스: {_ready_sources()}\n (원인: {e})")
f"소스 '{name}' 생성 실패 — 콘솔에서 소스를 만든 뒤 requires[].name(또는 clusterName)·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']})")
log(f"source '{disp}' READY (id={s['id']})")
return s["id"]
if s and s.get("status") == "ERROR":
raise SystemExit(f"source '{name}' 프로비저닝 ERROR")
raise SystemExit(f"source '{disp}' 프로비저닝 ERROR")
time.sleep(5)
raise SystemExit(f"source '{name}' READY 대기 초과")
raise SystemExit(f"source '{disp}' READY 대기 초과")
def reconcile_domain(fqdn: str) -> None:
@ -243,8 +249,8 @@ def bind(dep_id: str | None, alias: str, service_id: str | None, source: str) ->
# ── 환경 오버레이: base(requires/workloads) + environments.<env> 이름 매칭 병합 ─────────────────
# 앱 binds 는 논리이름 고정 → 환경별로 소스 정체(type/mode/plan)·워크로드 필드만 덮어쓴다. promote 무편집.
# ※ dev.py 의 동일 함수와 규약 일치(파리티) — 한쪽 바꾸면 다른 쪽도.
# 앱 binds 는 논리이름(name) 고정 → 환경별로 소스 정체(type/mode/plan)·실제 소스명(clusterName)·
# 워크로드 필드만 덮어쓴다. promote 무편집. ※ dev.py 의 동일 함수와 규약 일치(파리티) — 한쪽 바꾸면 다른 쪽도.
def _merge_env_vars(base, ov):
def norm(e):
if isinstance(e, dict):
@ -256,18 +262,47 @@ def _merge_env_vars(base, ov):
return list(out.values())
# 지원 소스 타입(콘솔 카탈로그와 동일) — 매니페스트 사전검증용.
SOURCE_TYPES = {"postgresql", "mysql", "mariadb", "mongodb", "redis", "minio", "rabbitmq", "solr", "oracle"}
def _validate_requires(requires: list, env: str) -> None:
"""병합된 requires 사전검증 — 배포 도중 KeyError 대신 친절한 오류로 조기 차단.
★ 환경 오버레이는 base 하고만 병합되며 dev↔val↔prod 끼리 상속하지 않는다 →
type 은 base(권장) 또는 그 환경에 반드시 있어야 한다. (A안: 공유 정체는 base 에 한 번.)"""
for r in requires:
name = r.get("name")
t = str(r.get("type") or "").lower()
if not t:
raise SystemExit(
f"소스 '{name}' 에 type 이 없습니다(환경 '{env}') — base.requires 또는 "
f"environments.{env}.requires 에 type 을 지정하세요. 환경 오버레이는 base 하고만 "
f"병합되고 dev↔val↔prod 끼리 상속하지 않습니다(공유 정체는 base 에 두는 걸 권장).")
if t not in SOURCE_TYPES:
raise SystemExit(
f"소스 '{name}' 의 type '{r.get('type')}' 은 지원되지 않습니다(환경 '{env}') — "
f"{', '.join(sorted(SOURCE_TYPES))} 중 하나여야 합니다.")
def resolve_env(m: dict, env: str) -> tuple[list, list]:
ec = (m.get("environments") or {}).get(env) or {}
by: dict = {}
order: list = []
for r in (m.get("requires") or []):
by[r["name"]] = dict(r); order.append(r["name"])
nm = r.get("name")
if not nm:
raise SystemExit("requires 항목에 name 이 없습니다(base.requires) — 각 소스는 논리 이름 name 이 필수입니다.")
by[nm] = dict(r); order.append(nm)
for r in (ec.get("requires") or []):
if r["name"] in by:
by[r["name"]].update(r) # 필드 병합(type/mode/plan 덮어쓰기)
nm = r.get("name")
if not nm:
raise SystemExit(f"requires 항목에 name 이 없습니다(environments.{env}.requires) — name 은 필수입니다.")
if nm in by:
by[nm].update(r) # 필드 병합(type/mode/plan/clusterName 덮어쓰기)
else:
by[r["name"]] = dict(r); order.append(r["name"])
by[nm] = dict(r); order.append(nm)
requires = [by[n] for n in order]
_validate_requires(requires, env) # 병합 후 사전검증(type 누락·미지원 등)
wov = ec.get("workloads") or {} # {워크로드명: 패치}
workloads = []
for w in (m.get("workloads") or []):
@ -303,6 +338,16 @@ def main() -> None:
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 ''}")
requires, workloads = resolve_env(m, ENV) # base + environments.<ENV> 오버레이 병합
# 충돌 가드 — 서로 다른 논리 소스가 같은 '실제 소스 이름'(clusterName 우선)으로 뭉개지면 binds 가 한 물리 소스로
# 조용히 aliasing 되어 자격/DB 오배선 → 거부(dev.py cmd_deploy 와 동일 계약·파리티).
src_owner: dict[str, str] = {}
for req in requires:
actual = req.get("clusterName") or req["name"]
if actual in src_owner and src_owner[actual] != req["name"]:
raise SystemExit(
f"소스명 충돌 — '{src_owner[actual]}' 와 '{req['name']}' 가 같은 실제 소스명 '{actual}'(clusterName)으로 "
f"뭉개집니다. 논리 이름별로 실제 소스명(clusterName)을 구분하세요.")
src_owner[actual] = req["name"]
# 운영 승격 승인 요청만 등록(project confirm 이 호출) — 게이트 켜진 클러스터면 워크로드별 PENDING 생성 후 종료.
if os.environ.get("YAKCLOUD_ACTION") == "register-approval":
cinfo = unwrap(api("GET", f"/clusters/{CLUSTER}")) or {}
@ -324,6 +369,13 @@ def main() -> None:
# 바인딩은 매 배포마다 재적용(멱등·재바인딩) — 승격/소스변경 시 별칭이 '새 소스'를 가리키게.
# (기존엔 신규 생성 때만 bind → shared→전용 등 소스 교체가 반영 안 되고 옛 env 에 고착되던 버그)
for b in w.get("binds", []):
# binds[].source 는 requires 의 '논리 이름'이어야 한다(clusterName/실제명 아님). 미해결이면
# 조용히 no-op(옛 동작) 대신 명확히 실패시킨다 — dev.py deploy_workload 와 동일 계약(파리티).
if b["source"] not in src_ids:
raise SystemExit(
f"워크로드 '{w['name']}' bind '{b['alias']}' → source '{b['source']}' 미해결 — "
f"binds[].source 는 requires 의 논리 이름이어야 합니다(clusterName/실제명 아님). "
f"requires: {', '.join(r['name'] for r in requires) or '(없음)'}")
bind(dep_id, b["alias"], src_ids.get(b["source"]), b["source"])
log("완료 — 콘솔 앱 탭에서 배포/바인딩 확인" if not DRY else "완료(dry-run) — 실제 변경 없음")

53
.yakcloud/dev.py
View File

@ -287,8 +287,13 @@ def _mint_oracle(cname: str, user: str, pw: str) -> None:
def provision_source(ns: str, project: str, req: dict) -> dict | None:
"""requires 소스 → 공유 도커 컨테이너의 프로젝트별 격리 DB + no-selector Service/Endpoints. 접속정보 반환."""
name = req["name"]
logical = req["name"]
# 실제 로컬 소스 이름 = clusterName(오버레이) 없으면 논리 이름. 서비스/격리DB/Secret 은 이 실제 이름 기준,
# 앱 binds 는 논리 이름 고정(로컬↔운영 매핑). 호출부는 반환 conn 을 논리 이름으로 키.
name = req.get("clusterName") or logical
stype = req["type"].upper()
if name != logical:
log(f"source '{logical}' → 실제 소스 '{name}' (clusterName)")
if req.get("mode") == "local":
warn(f"source '{name}': mode=local(인클러스터)는 로컬 dev 에서 공유 컨테이너로 대체됩니다.")
if stype not in SUPPORTED:
@ -547,9 +552,10 @@ def _report_to_console(project: str, reqs: list, wls: list) -> None:
"project": project,
"context": CTX, # kubectl 컨텍스트(콘솔 연결·삭제 탭 안내용) — kind-yak-<name>
# secret = 노트북 kind 커넥션 Secret 이름 → 콘솔이 connSecretRef 로 저장, 에이전트가 이 이름으로 읽음.
# name=논리 이름(binds 매칭·표시용), secret=실제 소스명(clusterName 우선)의 Secret 이름.
# 콘솔이 secret 을 connSecretRef 로 저장 → 에이전트가 액션의 그 이름으로 조회(재계산 아님)하므로 정합 유지.
"sources": [{"name": r["name"], "type": str(r.get("type", "")).upper(),
"secret": f"yak-dev-src-{sanitize(r['name'])}"} for r in reqs],
"secret": f"yak-dev-src-{sanitize(r.get('clusterName') or r['name'])}"} for r in reqs],
"workloads": [{"name": sanitize(w["name"]), "image": w["image"].replace("${TAG}", TAG),
"hosts": _disp_hosts(project, w), "port": int(w.get("port", 8080)),
"path": (w.get("expose", {}) or {}).get("path", "/") or "/",
@ -642,8 +648,8 @@ def _ensure_agent() -> None:
# ── 환경 오버레이: base(requires/workloads) + environments.<env> 이름 매칭 병합 ─────────────────
# 앱 binds 는 논리이름 고정 → 환경별로 소스 정체(type/mode/plan)·워크로드 필드만. 로컬은 항상 env="dev".
# ※ deploy.py 의 동일 함수와 규약 일치(파리티).
# 앱 binds 는 논리이름(name) 고정 → 환경별로 소스 정체(type/mode/plan)·실제 소스명(clusterName)·
# 워크로드 필드만. 로컬은 항상 env="dev". ※ deploy.py 의 동일 함수와 규약 일치(파리티).
def _merge_env_vars(base, ov):
def norm(e):
if isinstance(e, dict):
@ -655,17 +661,40 @@ def _merge_env_vars(base, ov):
return list(out.values())
def _validate_requires(requires: list, env: str) -> None:
"""병합된 requires 사전검증 — 프로비저닝 도중 KeyError 대신 친절한 오류로 조기 차단.
★ 환경 오버레이는 base 하고만 병합되며 dev↔val↔prod 끼리 상속하지 않는다 →
type 은 base(권장) 또는 그 환경에 반드시 있어야 한다. ※ deploy.py 와 규약 일치(파리티)."""
for r in requires:
name = r.get("name")
t = str(r.get("type") or "").upper()
if not t:
die(f"소스 '{name}' 에 type 이 없습니다(환경 '{env}') — base.requires 또는 "
f"environments.{env}.requires 에 type 을 지정하세요. 환경 오버레이는 base 하고만 "
f"병합되고 dev↔val↔prod 끼리 상속하지 않습니다(공유 정체는 base 에 두는 걸 권장).")
if t not in SUPPORTED:
die(f"소스 '{name}' 의 type '{r.get('type')}' 은 미지원(환경 '{env}') — "
f"{', '.join(sorted(s.lower() for s in SUPPORTED))} 중 하나여야 합니다.")
def resolve_env(m: dict, env: str):
ec = (m.get("environments") or {}).get(env) or {}
by, order = {}, []
for r in (m.get("requires") or []):
by[r["name"]] = dict(r); order.append(r["name"])
nm = r.get("name")
if not nm:
die("requires 항목에 name 이 없습니다(base.requires) — 각 소스는 논리 이름 name 이 필수입니다.")
by[nm] = dict(r); order.append(nm)
for r in (ec.get("requires") or []):
if r["name"] in by:
by[r["name"]].update(r)
nm = r.get("name")
if not nm:
die(f"requires 항목에 name 이 없습니다(environments.{env}.requires) — name 은 필수입니다.")
if nm in by:
by[nm].update(r)
else:
by[r["name"]] = dict(r); order.append(r["name"])
by[nm] = dict(r); order.append(nm)
requires = [by[n] for n in order]
_validate_requires(requires, env) # 병합 후 사전검증(type 누락·미지원 등)
wov = ec.get("workloads") or {}
workloads = []
for w in (m.get("workloads") or []):
@ -688,9 +717,10 @@ def cmd_deploy(m: dict) -> None:
reqs, wls = resolve_env(m, "dev") # base + environments.dev 오버레이(로컬은 항상 dev 환경)
# 충돌 가드 — 서로 다른 소스명/워크로드명이 같은 서비스명으로 뭉개지면 자격 오배선 → 거부.
# 서비스명은 '실제 소스 이름'(clusterName 우선)으로 만들어지므로 그 기준으로 검사.
svc_owner: dict[str, str] = {}
for r in reqs:
s = sanitize(r["name"])
s = sanitize(r.get("clusterName") or r["name"])
if s in svc_owner and svc_owner[s] != r["name"]:
die(f"소스명 충돌 — '{svc_owner[s]}' 와 '{r['name']}' 가 같은 서비스명 '{s}' 로 뭉개집니다. 이름을 구분하세요.")
svc_owner[s] = r["name"]
@ -709,7 +739,8 @@ def cmd_deploy(m: dict) -> None:
results = [deploy_workload(project, ns, w, src_conns) for w in wls]
# 매니페스트에서 빠진 소스/워크로드 정리(고아 방지 — add-only 가 아니라 리컨실).
_prune(ns, "source", "yakcloud.dev/source", {sanitize(r["name"]) for r in reqs})
# 소스 라벨(yakcloud.dev/source)=실제 소스명(clusterName 우선)의 sanitize → desired 도 그 기준.
_prune(ns, "source", "yakcloud.dev/source", {sanitize(r.get("clusterName") or r["name"]) for r in reqs})
_prune(ns, "workload", "app", {sanitize(w["name"]) for w in wls})
_report_to_console(project, reqs, wls) # 콘솔 LOCAL 클러스터 탭(데이터소스·앱) 리포트