532 lines
26 KiB
Python
532 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""yakcloud datasource migrate — 데이터소스 스냅샷(블랙리스트) 캡처 + 빈 타깃 적용(부트스트랩).
|
||
|
||
모델(확정):
|
||
capture = 소스 전체 native 덤프 − 블랙리스트(db/<source>/.migrateignore, 글롭) → db/<source>/snapshot/.
|
||
**읽기전용**(항상 안전). 커밋 전 '무엇이 잡히는지' 미리보기.
|
||
apply = 타깃이 **비어 있을 때만** 복원. 데이터가 있으면 **스킵**(절대 안 덮음, force 없음).
|
||
즉 새 환경을 dev 상태로 채우는 부트스트랩/시딩 도구. (prod 적용은 백엔드 경유 — 별도 승인)
|
||
|
||
dev 채널: 러너를 DB 클라이언트 이미지 one-shot 으로 compose 네트워크에 join → 소스 컨테이너 직접(자격 미노출).
|
||
자격은 yakcloud_dev 의 결정적 파생(creds_for)을 그대로 사용. 지원 타입: postgresql·mysql·mariadb·mongodb·minio.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import fnmatch
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import yakcloud_dev as dev # noqa: E402 (동일 scripts/ 의 dev 엔진 헬퍼 재사용)
|
||
|
||
SNAP = "db" # 프로젝트 루트 db/<source>/ (git 추적 — .gitignore 아님)
|
||
IGNORE = ".migrateignore"
|
||
SUPPORTED = ("postgresql", "mysql", "mariadb", "mongodb", "minio", "redis", "solr", "rabbitmq", "oracle")
|
||
IMG = {"postgresql": "postgres:16-alpine", "mysql": "mysql:8.4", "mariadb": "mariadb:11.4",
|
||
"mongodb": "mongo:7", "minio": "minio/mc:latest",
|
||
"redis": "redis:7-alpine", "solr": "solr:9", "rabbitmq": "rabbitmq:3.13-management"}
|
||
UNSUPPORTED_NEXT = () # 9종 전부 지원(oracle=Data Pump)
|
||
|
||
|
||
def _fill(tmpl: str, **kw) -> str:
|
||
"""플레이스홀더 치환 — f-string/.format 대신(셸의 \\r\\n·$7·${..} 이스케이프를 보존)."""
|
||
for k, v in kw.items():
|
||
tmpl = tmpl.replace("{" + k + "}", str(v))
|
||
return tmpl
|
||
|
||
|
||
# redis 캡처: SCAN → 블랙리스트(키 글롭) 클라이언트 제외 → 키별 PTTL+DUMP(base64) → keys.b64. 읽기전용.
|
||
_REDIS_CAPTURE = r'''
|
||
R="redis-cli -h {host} -a {pw} --no-auth-warning -n {db}"; EX="{bl}"; : > /out/keys.b64
|
||
$R --scan | while IFS= read -r k; do
|
||
[ -z "$k" ] && continue
|
||
skip=0; for pat in $EX; do case "$k" in $pat) skip=1;; esac; done; [ "$skip" = 1 ] && continue
|
||
pttl=$($R PTTL "$k"); case "$pttl" in -*) pttl=0;; esac
|
||
kb=$(printf "%s" "$k" | base64 | tr -d "\n")
|
||
vb=$($R DUMP "$k" | head -c -1 | base64 | tr -d "\n")
|
||
printf "%s\t%s\t%s\n" "$pttl" "$kb" "$vb" >> /out/keys.b64
|
||
done
|
||
'''
|
||
|
||
# redis 복원: keys.b64 → RESP RESTORE(REPLACE) 스트림 → redis-cli --pipe(바이너리 안전). 빈 타깃 전제.
|
||
_REDIS_RESTORE = r'''
|
||
R="redis-cli -h {host} -a {pw} --no-auth-warning -n {db}"; TAB=$(printf '\t')
|
||
{ while IFS="$TAB" read -r pttl kb vb; do
|
||
[ -z "$kb" ] && continue
|
||
k=$(printf "%s" "$kb" | base64 -d)
|
||
vlen=$(printf "%s" "$vb" | base64 -d | wc -c | tr -d ' ')
|
||
klen=$(printf "%s" "$k" | wc -c | tr -d ' '); tlen=$(printf "%s" "$pttl" | wc -c | tr -d ' ')
|
||
printf '*5\r\n$7\r\nRESTORE\r\n'
|
||
printf '$%s\r\n%s\r\n' "$klen" "$k"
|
||
printf '$%s\r\n%s\r\n' "$tlen" "$pttl"
|
||
printf '$%s\r\n' "$vlen"; printf "%s" "$vb" | base64 -d; printf '\r\n'
|
||
printf '$7\r\nREPLACE\r\n'
|
||
done < /out/keys.b64; } | $R --pipe
|
||
'''
|
||
|
||
|
||
def die(m: str) -> None:
|
||
sys.exit(f" ✗ {m}")
|
||
|
||
|
||
def _net(m: dict) -> str:
|
||
return f"{dev.compose_project(m)}_default"
|
||
|
||
|
||
def _running(m: dict, source: str) -> bool:
|
||
j = dev._ps_state(m).get(source) or {}
|
||
return (j.get("State") or "") == "running"
|
||
|
||
|
||
def _srcdir(source: str) -> str:
|
||
return os.path.join(SNAP, source)
|
||
|
||
|
||
def _blacklist(source: str) -> list[str]:
|
||
p = os.path.join(_srcdir(source), IGNORE)
|
||
if not os.path.exists(p):
|
||
return []
|
||
out = []
|
||
for ln in open(p):
|
||
ln = ln.split("#", 1)[0].strip()
|
||
if ln:
|
||
out.append(ln)
|
||
return out
|
||
|
||
|
||
def _excluded(name: str, patterns: list[str]) -> bool:
|
||
return any(fnmatch.fnmatch(name, pat) for pat in patterns)
|
||
|
||
|
||
def drun(net: str, image: str, shell: str, mounts=None, capture=True, check=True):
|
||
"""docker run one-shot(sh -c) — 모든 명령을 쉘 경유(리다이렉트/파이프 지원)."""
|
||
args = ["docker", "run", "--rm", "--network", net]
|
||
for host, cont in (mounts or []):
|
||
os.makedirs(host, exist_ok=True)
|
||
args += ["-v", f"{os.path.abspath(host)}:{cont}"]
|
||
args += ["--entrypoint", "sh", image, "-c", shell]
|
||
r = subprocess.run(args, text=True,
|
||
stdout=subprocess.PIPE if capture else None,
|
||
stderr=subprocess.PIPE if capture else None)
|
||
if check and r.returncode != 0:
|
||
die(f"클라이언트 실행 실패({r.returncode}): {(r.stderr or r.stdout or '').strip()[:400]}")
|
||
return r
|
||
|
||
|
||
# ── 타입별: 대상 단위 나열(미리보기·글롭 확장용) ────────────────────────────
|
||
def list_units(m: dict, source: str, stype: str, cr: dict) -> list[str]:
|
||
net = _net(m)
|
||
if stype == "postgresql":
|
||
url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
|
||
q = "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename"
|
||
r = drun(net, IMG[stype], f"psql '{url}' -tAc \"{q}\"", check=False)
|
||
elif stype in ("mysql", "mariadb"):
|
||
q = f"SELECT table_name FROM information_schema.tables WHERE table_schema='{cr['db']}'"
|
||
r = drun(net, IMG[stype],
|
||
f"mysql -h {source} -u{cr['user']} -p{cr['password']} -N -B -e \"{q}\" 2>/dev/null", check=False)
|
||
elif stype == "mongodb":
|
||
js = "db.getCollectionNames().filter(c=>!c.startsWith('_yakcloud')).join('\\n')"
|
||
r = drun(net, IMG[stype],
|
||
f"mongosh --quiet --host {source} -u {cr['user']} -p {cr['password']} "
|
||
f"--authenticationDatabase {cr['db']} {cr['db']} --eval \"{js}\"", check=False)
|
||
elif stype == "minio":
|
||
# minio/mc 이미지엔 awk 가 없음 → --json 으로 받아 파이썬에서 key 파싱.
|
||
setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null 2>&1"
|
||
r = drun(net, IMG[stype],
|
||
f"{setup} && mc ls --recursive --json s/{cr['bucket']} 2>/dev/null", check=False)
|
||
keys = []
|
||
for ln in (r.stdout or "").splitlines():
|
||
ln = ln.strip()
|
||
if not ln:
|
||
continue
|
||
try:
|
||
j = json.loads(ln)
|
||
if j.get("key"):
|
||
keys.append(j["key"])
|
||
except Exception:
|
||
pass
|
||
return keys
|
||
elif stype == "redis":
|
||
r = drun(net, IMG[stype],
|
||
f"redis-cli -h {source} -a {cr['password']} --no-auth-warning -n {cr['db']} --scan", check=False)
|
||
return [x for x in (r.stdout or "").splitlines() if x.strip()]
|
||
elif stype == "rabbitmq":
|
||
ra = (f"rabbitmqadmin -H {source} -P 15672 -u {cr['user']} -p {cr['password']} "
|
||
f"-V {cr['vhost']} -f tsv -q")
|
||
q = drun(net, IMG[stype], f"{ra} list queues name", check=False)
|
||
e = drun(net, IMG[stype], f"{ra} list exchanges name", check=False)
|
||
qn = [x.strip() for x in (q.stdout or "").splitlines() if x.strip() and x.strip() != "name"]
|
||
en = [x.strip() for x in (e.stdout or "").splitlines()
|
||
if x.strip() and x.strip() != "name" and not x.strip().startswith("amq.")]
|
||
return qn + en
|
||
elif stype == "solr":
|
||
# source=core 1:1 → 단위='문서'. is_empty/미리보기용 표본 id(최대 10). 전체 수는 _solr_numfound.
|
||
r = drun(net, IMG[stype],
|
||
f'curl -sS "http://{source}:8983/solr/{cr["core"]}/select?q=*:*&rows=10&fl=id&wt=json"', check=False)
|
||
try:
|
||
return [str(d.get("id", "?")) for d in json.loads(r.stdout or "{}").get("response", {}).get("docs", [])]
|
||
except Exception:
|
||
return []
|
||
elif stype == "oracle":
|
||
return _ora_lines(_ora_sql(source, cr,
|
||
f"SELECT table_name FROM all_tables WHERE owner = UPPER('{cr['user']}');"))
|
||
else:
|
||
return []
|
||
return [x.strip() for x in (r.stdout or "").splitlines() if x.strip() and "Warning" not in x]
|
||
|
||
|
||
def _solr_numfound(m: dict, source: str, cr: dict) -> int:
|
||
r = drun(_net(m), IMG["solr"],
|
||
f'curl -sS "http://{source}:8983/solr/{cr["core"]}/select?q=*:*&rows=0&wt=json"', check=False)
|
||
try:
|
||
return int(json.loads(r.stdout or "{}").get("response", {}).get("numFound", 0))
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
def _rabbit_filter(path: str, bl: list[str], info: dict) -> None:
|
||
"""rabbitmq defs.json 에서 블랙리스트(큐/익스체인지 글롭) 항목 + 참조 bindings 를 제거."""
|
||
d = json.load(open(path))
|
||
rmq = {q["name"] for q in d.get("queues", []) if _excluded(q.get("name", ""), bl)}
|
||
rmx = {x["name"] for x in d.get("exchanges", []) if _excluded(x.get("name", ""), bl)}
|
||
d["queues"] = [q for q in d.get("queues", []) if q.get("name", "") not in rmq]
|
||
d["exchanges"] = [x for x in d.get("exchanges", []) if x.get("name", "") not in rmx]
|
||
keep = []
|
||
for b in d.get("bindings", []):
|
||
if b.get("source", "") in rmx:
|
||
continue
|
||
dt, dn = b.get("destination_type", ""), b.get("destination", "")
|
||
if dt == "queue" and dn in rmq:
|
||
continue
|
||
if dt == "exchange" and dn in rmx:
|
||
continue
|
||
keep.append(b)
|
||
d["bindings"] = keep
|
||
json.dump(d, open(path, "w"), ensure_ascii=False, indent=2)
|
||
info["excluded"] = sorted(rmq | rmx)
|
||
|
||
|
||
def _solr_capture(net: str, source: str, cr: dict, bl: list[str], outdir: str) -> dict:
|
||
"""cursorMark 딥페이징으로 전체 문서 export(블랙리스트=서버측 fq 제외), 내부필드 제거 → docs.json."""
|
||
core = cr["core"]
|
||
blfq = ""
|
||
if bl:
|
||
expr = " OR ".join(f"( {ln} )" for ln in bl)
|
||
blfq = f"--data-urlencode 'fq=-({expr})'"
|
||
internal = {"_version_", "_root_", "_nest_path_", "_nest_parent_"}
|
||
docs, mark, n = [], "*", 0
|
||
while True:
|
||
cmd = (f'curl -sS "http://{source}:8983/solr/{core}/select" -G --data-urlencode "q=*:*" {blfq} '
|
||
f'--data-urlencode "fl=*" --data-urlencode "sort=id asc" --data-urlencode "rows=1000" '
|
||
f'--data-urlencode "cursorMark={mark}" --data-urlencode "wt=json" -o /out/_page.json')
|
||
drun(net, IMG["solr"], cmd, [(outdir, "/out")])
|
||
j = json.load(open(os.path.join(outdir, "_page.json")))
|
||
for doc in j.get("response", {}).get("docs", []):
|
||
docs.append({k: v for k, v in doc.items() if k not in internal})
|
||
nxt = j.get("nextCursorMark", mark)
|
||
if nxt == mark:
|
||
break
|
||
mark = nxt
|
||
n += 1
|
||
p = os.path.join(outdir, "_page.json")
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
json.dump(docs, open(os.path.join(outdir, "docs.json"), "w"), ensure_ascii=False)
|
||
drun(net, IMG["solr"], f'curl -sS "http://{source}:8983/solr/{core}/schema?wt=json" -o /out/schema.json',
|
||
[(outdir, "/out")])
|
||
return {"file": "docs.json", "captured": len(docs)}
|
||
|
||
|
||
# ── oracle: Data Pump(expdp/impdp) — 파일이 DB 서버측이라 소스 컨테이너 exec + docker cp + REMAP_SCHEMA ──
|
||
def _ora_sql(source: str, cr: dict, sql: str) -> str:
|
||
script = "SET HEADING OFF PAGESIZE 0 FEEDBACK OFF VERIFY OFF ECHO OFF TRIMSPOOL ON\n" + sql + "\nEXIT\n"
|
||
r = subprocess.run(["docker", "exec", "-i", f"yakdev-{source}", "bash", "-lc",
|
||
f"sqlplus -s system/{cr['sys_pw']}@//localhost:1521/FREEPDB1"],
|
||
input=script, text=True, capture_output=True)
|
||
return r.stdout or ""
|
||
|
||
|
||
def _ora_lines(out: str) -> list[str]:
|
||
bad = ("ORA-", "SP2-", "SQL>", "altered", "PL/SQL", "Connected")
|
||
return [ln.strip() for ln in out.splitlines() if ln.strip() and not any(b in ln for b in bad)]
|
||
|
||
|
||
def _ora_dppath(source: str, cr: dict) -> str:
|
||
ls = _ora_lines(_ora_sql(source, cr,
|
||
"SELECT directory_path FROM dba_directories WHERE directory_name='DATA_PUMP_DIR';"))
|
||
return ls[-1] if ls else "/opt/oracle/admin/FREE/dpdump"
|
||
|
||
|
||
def _ora_datapump(source: str, cr: dict, parfile: str, tool: str) -> None:
|
||
cont = f"yakdev-{source}"
|
||
subprocess.run(["docker", "exec", "-i", cont, "bash", "-lc", "cat > /tmp/yak.par"],
|
||
input=parfile, text=True, capture_output=True)
|
||
r = subprocess.run(["docker", "exec", cont, "bash", "-lc",
|
||
f"{tool} system/{cr['sys_pw']}@//localhost:1521/FREEPDB1 parfile=/tmp/yak.par"],
|
||
text=True, capture_output=True)
|
||
out = (r.stdout or "") + (r.stderr or "") # expdp/impdp 는 진행상황을 stderr 로 출력
|
||
if "successfully completed" not in out:
|
||
die(f"{tool} 실패:\n{out[-600:]}")
|
||
|
||
|
||
# ── 미리보기 ────────────────────────────────────────────────────────────
|
||
def preview(m: dict, targets: list[tuple[str, str]]) -> dict:
|
||
seed = dev.ensure_seed()
|
||
plan = {}
|
||
print("캡처 미리보기 (읽기전용) — ✓=포함, ✗=제외(블랙리스트)")
|
||
for source, stype in targets:
|
||
cr = dev.creds_for(seed, source, stype)
|
||
bl = _blacklist(source)
|
||
if stype == "solr": # solr 블랙리스트=서버측 fq(문서 필터), fnmatch 미적용 → 문서 수만 표시
|
||
nf = _solr_numfound(m, source, cr)
|
||
plan[source] = {"type": stype, "numFound": nf, "blacklist": bl}
|
||
print(f"\n ● {source} (solr) — 문서 {nf}개"
|
||
+ (f" · 블랙리스트 {len(bl)}줄(서버측 fq 로 제외)" if bl else ""))
|
||
if not bl:
|
||
print(f" · 블랙리스트 없음 → 전 문서 포함. 제외는 "
|
||
f"{os.path.join(_srcdir(source), IGNORE)} 에 Solr 쿼리절(한 줄에 하나)로")
|
||
continue
|
||
units = list_units(m, source, stype, cr)
|
||
blm = [p.upper() for p in bl] if stype == "oracle" else bl # oracle 식별자=대문자
|
||
inc = [u for u in units if not _excluded(u, blm)]
|
||
exc = [u for u in units if _excluded(u, blm)]
|
||
plan[source] = {"type": stype, "include": inc, "exclude": exc, "blacklist": bl}
|
||
unit_word = {"minio": "오브젝트", "redis": "키", "rabbitmq": "큐/익스체인지",
|
||
"oracle": "테이블"}.get(stype, "테이블/컬렉션")
|
||
print(f"\n ● {source} ({stype}) — {unit_word} {len(units)}개 · 포함 {len(inc)} / 제외 {len(exc)}")
|
||
for u in inc[:20]:
|
||
print(f" ✓ {u}")
|
||
if len(inc) > 20:
|
||
print(f" … 외 {len(inc)-20}개")
|
||
for u in exc:
|
||
print(f" ✗ {u} (블랙리스트)")
|
||
if not bl:
|
||
print(f" · 블랙리스트 없음 → 전부 포함. 민감/개인정보·쓰레기는 "
|
||
f"{os.path.join(_srcdir(source), IGNORE)} 에 글롭으로 제외 권장")
|
||
return plan
|
||
|
||
|
||
# ── 캡처(전체 native 덤프 − 블랙리스트) ───────────────────────────────────
|
||
def capture_source(m: dict, source: str, stype: str, cr: dict) -> dict:
|
||
net, outdir = _net(m), os.path.join(_srcdir(source), "snapshot")
|
||
os.makedirs(outdir, exist_ok=True)
|
||
bl = _blacklist(source)
|
||
if stype == "solr": # solr 는 fq(서버측) 제외라 fnmatch 단위 계산 안 함
|
||
units, excluded = [], []
|
||
elif stype == "oracle": # oracle 식별자=대문자 → 블랙리스트 글롭도 대문자 정규화 매칭
|
||
units = list_units(m, source, stype, cr)
|
||
excluded = [u for u in units if _excluded(u, [p.upper() for p in bl])]
|
||
else:
|
||
units = list_units(m, source, stype, cr)
|
||
excluded = [u for u in units if _excluded(u, bl)]
|
||
info = {"type": stype, "excluded": excluded, "blacklist": bl}
|
||
if stype == "postgresql":
|
||
url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
|
||
# pg_dump 는 패턴(*)을 지원 → 블랙리스트 글롭을 그대로 넘겨 소유 시퀀스(<tbl>_id_seq 등)까지 제외.
|
||
ex = " ".join(f"--exclude-table='{g}'" for g in bl)
|
||
drun(net, IMG[stype], f"pg_dump '{url}' {ex} -f /out/snapshot.sql", [(outdir, "/out")])
|
||
info["file"] = "snapshot.sql"
|
||
elif stype in ("mysql", "mariadb"):
|
||
ig = " ".join(f"--ignore-table={cr['db']}.{e}" for e in excluded)
|
||
drun(net, IMG[stype],
|
||
f"mysqldump -h {source} -u{cr['user']} -p{cr['password']} --skip-comments {ig} "
|
||
f"{cr['db']} > /out/snapshot.sql", [(outdir, "/out")])
|
||
info["file"] = "snapshot.sql"
|
||
elif stype == "mongodb":
|
||
ex = " ".join(f"--excludeCollection={e}" for e in excluded)
|
||
drun(net, IMG[stype],
|
||
f"mongodump --host {source} -u {cr['user']} -p {cr['password']} "
|
||
f"--authenticationDatabase {cr['db']} --db {cr['db']} {ex} --archive=/out/dump.archive",
|
||
[(outdir, "/out")])
|
||
info["file"] = "dump.archive"
|
||
elif stype == "minio":
|
||
setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null"
|
||
exflag = " ".join(f"--exclude '{e}'" for e in bl)
|
||
drun(net, IMG[stype],
|
||
f"{setup} && mc mirror --overwrite {exflag} s/{cr['bucket']} /out/bucket", [(outdir, "/out")])
|
||
info["file"], info["bucket"] = "bucket/", cr["bucket"]
|
||
elif stype == "redis":
|
||
drun(net, IMG[stype], _fill(_REDIS_CAPTURE, host=source, pw=cr["password"], db=cr["db"], bl=" ".join(bl)),
|
||
[(outdir, "/out")])
|
||
info["file"] = "keys.b64"
|
||
elif stype == "rabbitmq":
|
||
drun(net, IMG[stype],
|
||
f"rabbitmqadmin -H {source} -P 15672 -u {cr['user']} -p {cr['password']} -V {cr['vhost']} "
|
||
f"export /out/defs.json", [(outdir, "/out")])
|
||
_rabbit_filter(os.path.join(outdir, "defs.json"), bl, info) # 블랙리스트 큐/익스체인지 + bindings 제거
|
||
info["file"] = "defs.json"
|
||
info["note"] = "메시지 본문 제외(전이성). 정의(큐/익스체인지/바인딩/정책)만 이관."
|
||
elif stype == "solr":
|
||
info.update(_solr_capture(net, source, cr, bl, outdir))
|
||
elif stype == "oracle":
|
||
cont, schema, dp = f"yakdev-{source}", cr["user"].upper(), _ora_dppath(source, cr)
|
||
subprocess.run(["docker", "exec", cont, "bash", "-lc", f"rm -f {dp}/yak.dmp"], capture_output=True)
|
||
par = f"schemas={schema}\ndirectory=DATA_PUMP_DIR\ndumpfile=yak.dmp\nreuse_dumpfiles=y\nnologfile=y\n"
|
||
if excluded:
|
||
lst = ", ".join(f"'{e.upper()}'" for e in excluded)
|
||
par += f'EXCLUDE=TABLE:"IN ({lst})"\n'
|
||
_ora_datapump(source, cr, par, "expdp")
|
||
cp = subprocess.run(["docker", "cp", f"{cont}:{dp}/yak.dmp", os.path.join(outdir, "data.dmp")],
|
||
capture_output=True, text=True)
|
||
if cp.returncode != 0:
|
||
die(f"덤프 추출 실패: {cp.stderr[-300:]}")
|
||
info["file"], info["schema"] = "data.dmp", schema
|
||
else:
|
||
die(f"미지원 타입(캡처): {stype}")
|
||
json.dump(info, open(os.path.join(outdir, "manifest.json"), "w"), ensure_ascii=False, indent=2)
|
||
return info
|
||
|
||
|
||
# ── 타깃 비어있음 검사 + 적용(부트스트랩) ─────────────────────────────────
|
||
def is_empty(m: dict, source: str, stype: str, cr: dict) -> bool:
|
||
if stype == "solr":
|
||
return _solr_numfound(m, source, cr) == 0
|
||
return len(list_units(m, source, stype, cr)) == 0
|
||
|
||
|
||
def apply_source(m: dict, source: str, stype: str, cr: dict) -> str:
|
||
net, outdir = _net(m), os.path.join(_srcdir(source), "snapshot")
|
||
if not os.path.isdir(outdir):
|
||
return "스냅샷 없음(먼저 capture)"
|
||
if not is_empty(m, source, stype, cr):
|
||
return "타깃에 데이터 있음 → 스킵(덮지 않음)"
|
||
if stype == "postgresql":
|
||
url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
|
||
drun(net, IMG[stype], f"psql '{url}' -v ON_ERROR_STOP=1 -f /out/snapshot.sql", [(outdir, "/out")])
|
||
elif stype in ("mysql", "mariadb"):
|
||
drun(net, IMG[stype],
|
||
f"mysql -h {source} -u{cr['user']} -p{cr['password']} {cr['db']} < /out/snapshot.sql",
|
||
[(outdir, "/out")])
|
||
elif stype == "mongodb":
|
||
drun(net, IMG[stype],
|
||
f"mongorestore --host {source} -u {cr['user']} -p {cr['password']} "
|
||
f"--authenticationDatabase {cr['db']} --archive=/out/dump.archive", [(outdir, "/out")])
|
||
elif stype == "minio":
|
||
setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null"
|
||
drun(net, IMG[stype],
|
||
f"{setup} && (mc mb -p s/{cr['bucket']} 2>/dev/null || true) && "
|
||
f"mc mirror --overwrite /out/bucket s/{cr['bucket']}", [(outdir, "/out")])
|
||
elif stype == "redis":
|
||
drun(net, IMG[stype], _fill(_REDIS_RESTORE, host=source, pw=cr["password"], db=cr["db"]), [(outdir, "/out")])
|
||
elif stype == "rabbitmq":
|
||
drun(net, IMG[stype],
|
||
f"rabbitmqadmin -H {source} -P 15672 -u {cr['user']} -p {cr['password']} -V {cr['vhost']} "
|
||
f"import /out/defs.json", [(outdir, "/out")])
|
||
elif stype == "solr":
|
||
drun(net, IMG[stype],
|
||
f'curl -sS "http://{source}:8983/solr/{cr["core"]}/update?commit=true" '
|
||
f'-H "Content-Type: application/json" --data-binary @/out/docs.json', [(outdir, "/out")])
|
||
elif stype == "oracle":
|
||
cont, dst, dp = f"yakdev-{source}", cr["user"].upper(), _ora_dppath(source, cr)
|
||
src = json.load(open(os.path.join(outdir, "manifest.json"))).get("schema", dst)
|
||
cp = subprocess.run(["docker", "cp", os.path.join(outdir, "data.dmp"), f"{cont}:{dp}/yak.dmp"],
|
||
capture_output=True, text=True)
|
||
if cp.returncode != 0:
|
||
die(f"덤프 주입 실패: {cp.stderr[-300:]}")
|
||
# docker cp 는 호스트 uid 로 파일 생성 → oracle 프로세스가 읽도록 root(-u 0)로 권한 부여.
|
||
subprocess.run(["docker", "exec", "-u", "0", cont, "bash", "-lc", f"chmod 644 {dp}/yak.dmp"],
|
||
capture_output=True)
|
||
_ora_sql(source, cr, f"ALTER USER {dst} QUOTA UNLIMITED ON USERS;") # 쿼터 보장(멱등)
|
||
# exclude=user: 대상 스키마가 이미 존재(gvenzl 생성) → CREATE USER 스킵(ORA-31684 benign 방지, 깨끗한 성공)
|
||
par = (f"directory=DATA_PUMP_DIR\ndumpfile=yak.dmp\nnologfile=y\nexclude=user\n"
|
||
f"remap_schema={src}:{dst}\ntable_exists_action=skip\n")
|
||
_ora_datapump(source, cr, par, "impdp")
|
||
else:
|
||
die(f"미지원 타입(적용): {stype}")
|
||
return "적용 완료(부트스트랩)"
|
||
|
||
|
||
def select_targets(m: dict, names: list[str], require_running: bool) -> list[tuple[str, str]]:
|
||
reqs = {r["name"]: r["type"] for r in dev.requires(m)}
|
||
if names:
|
||
miss = [n for n in names if n not in reqs]
|
||
if miss:
|
||
die(f"requires 에 없음: {', '.join(miss)}")
|
||
chosen = [(n, reqs[n]) for n in names]
|
||
else:
|
||
chosen = list(reqs.items())
|
||
out = []
|
||
for s, t in chosen:
|
||
if t not in SUPPORTED:
|
||
print(f" · {s}({t}) — 마이그레이션 미지원(다음 단계: {', '.join(UNSUPPORTED_NEXT)}). 건너뜀")
|
||
continue
|
||
if require_running and not _running(m, s):
|
||
print(f" · {s} — dev 컨테이너 미기동('yakcloud dev up {s}' 먼저). 건너뜀")
|
||
continue
|
||
out.append((s, t))
|
||
if not out:
|
||
die("대상 소스가 없습니다.")
|
||
return out
|
||
|
||
|
||
# ── 명령 ────────────────────────────────────────────────────────────────
|
||
def cmd_capture(a) -> None:
|
||
m = dev.load_manifest()
|
||
targets = select_targets(m, a.source, require_running=True)
|
||
preview(m, targets)
|
||
if a.dry_run:
|
||
print("\n(미리보기 전용 — 캡처 안 함)")
|
||
return
|
||
if not a.yes and sys.stdin.isatty():
|
||
if input("\n위 대상을 db/<source>/snapshot/ 에 캡처할까요? [y/N] ").strip().lower() not in ("y", "yes"):
|
||
print("취소.")
|
||
return
|
||
seed = dev.ensure_seed()
|
||
for source, stype in targets:
|
||
cr = dev.creds_for(seed, source, stype)
|
||
info = capture_source(m, source, stype, cr)
|
||
print(f" ✓ {source} 캡처 → {os.path.join(_srcdir(source), 'snapshot', info['file'])}"
|
||
f" (제외 {len(info['excluded'])})")
|
||
print(" · db/ 를 커밋하세요(리뷰 대상). 적용: 'yakcloud datasource migrate apply'(빈 타깃만)")
|
||
|
||
|
||
def cmd_apply(a) -> None:
|
||
m = dev.load_manifest()
|
||
targets = select_targets(m, a.source, require_running=True)
|
||
seed = dev.ensure_seed()
|
||
for source, stype in targets:
|
||
cr = dev.creds_for(seed, source, stype)
|
||
print(f" ▸ {source} ({stype}): {apply_source(m, source, stype, cr)}")
|
||
print(" · prod 적용은 콘솔 소스관리자 컨펌 + 백업 후(롤백 보장) — 별도 경로")
|
||
|
||
|
||
def cmd_status(a) -> None:
|
||
m = dev.load_manifest()
|
||
reqs = {r["name"]: r["type"] for r in dev.requires(m)}
|
||
names = a.source or list(reqs)
|
||
print("migrate 스냅샷 상태")
|
||
for s in names:
|
||
t = reqs.get(s, "?")
|
||
snap = os.path.join(_srcdir(s), "snapshot", "manifest.json")
|
||
bl = _blacklist(s)
|
||
if os.path.exists(snap):
|
||
info = json.load(open(snap))
|
||
print(f" {s} ({t}): 스냅샷 있음(제외 {len(info.get('excluded', []))}) · 블랙리스트 {len(bl)}줄")
|
||
else:
|
||
print(f" {s} ({t}): 스냅샷 없음 · 블랙리스트 {len(bl)}줄")
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(prog="yakcloud datasource migrate")
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
c = sub.add_parser("capture", help="소스 전체 상태 − 블랙리스트 를 db/<source>/snapshot/ 에 캡처(읽기전용)")
|
||
c.add_argument("source", nargs="*")
|
||
c.add_argument("--dry-run", action="store_true", help="미리보기만")
|
||
c.add_argument("-y", "--yes", action="store_true", help="확인 없이")
|
||
c.set_defaults(fn=cmd_capture)
|
||
ap = sub.add_parser("apply", help="스냅샷을 타깃에 적용(빈 타깃만, 데이터 있으면 스킵)")
|
||
ap.add_argument("source", nargs="*")
|
||
ap.set_defaults(fn=cmd_apply)
|
||
st = sub.add_parser("status", help="스냅샷·블랙리스트 상태")
|
||
st.add_argument("source", nargs="*")
|
||
st.set_defaults(fn=cmd_status)
|
||
return p
|
||
|
||
|
||
if __name__ == "__main__":
|
||
args = build_parser().parse_args()
|
||
args.fn(args)
|