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

cli v0.9.0: migrate 확장 redis/solr/rabbitmq — redis(SCAN+DUMP/RESTORE·TTL보존), solr(cursorMark export·fq 블랙리스트), rabbitmq(정의 export/import·메시지제외). 3종 docker E2E(캡처·제외·apply-skip·reset복원) 검증. oracle=다음

Browse Source
This commit is contained in:
yakenator
2026-08-27 22:00:42 +09:00
parent a498a402e6
commit d8fe2b0ea3
4 changed files with 168 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
scripts/yakcloud_migrate.py
View File

@ -24,9 +24,48 @@ import yakcloud_dev as dev # noqa: E402 (동일 scripts/ 의 dev 엔진 헬퍼
SNAP = "db" # 프로젝트 루트 db/<source>/ (git 추적 — .gitignore 아님)
IGNORE = ".migrateignore"
SUPPORTED = ("postgresql", "mysql", "mariadb", "mongodb", "minio")
SUPPORTED = ("postgresql", "mysql", "mariadb", "mongodb", "minio", "redis", "solr", "rabbitmq")
IMG = {"postgresql": "postgres:16-alpine", "mysql": "mysql:8.4", "mariadb": "mariadb:11.4",
"mongodb": "mongo:7", "minio": "minio/mc:latest"}
"mongodb": "mongo:7", "minio": "minio/mc:latest",
"redis": "redis:7-alpine", "solr": "solr:9", "rabbitmq": "rabbitmq:3.13-management"}
UNSUPPORTED_NEXT = ("oracle",) # 다음 단계(무거움/설계 보완 필요)
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:
@ -110,11 +149,94 @@ def list_units(m: dict, source: str, stype: str, cr: dict) -> list[str]:
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 []
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)}
# ── 미리보기 ────────────────────────────────────────────────────────────
def preview(m: dict, targets: list[tuple[str, str]]) -> dict:
seed = dev.ensure_seed()
@ -123,11 +245,20 @@ def preview(m: dict, targets: list[tuple[str, str]]) -> dict:
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)
inc = [u for u in units if not _excluded(u, bl)]
exc = [u for u in units if _excluded(u, bl)]
plan[source] = {"type": stype, "include": inc, "exclude": exc, "blacklist": bl}
unit_word = {"minio": "오브젝트"}.get(stype, "테이블/컬렉션")
unit_word = {"minio": "오브젝트", "redis": "키", "rabbitmq": "큐/익스체인지"}.get(stype, "테이블/컬렉션")
print(f"\n ● {source} ({stype}) — {unit_word} {len(units)}개 · 포함 {len(inc)} / 제외 {len(exc)}")
for u in inc[:20]:
print(f" ✓ {u}")
@ -146,8 +277,11 @@ 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)
units = list_units(m, source, stype, cr)
excluded = [u for u in units if _excluded(u, bl)]
if stype == "solr": # solr 는 fq(서버측) 제외라 fnmatch 단위 계산 안 함
units, excluded = [], []
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']}"
@ -174,6 +308,19 @@ def capture_source(m: dict, source: str, stype: str, cr: dict) -> dict:
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))
else:
die(f"미지원 타입(캡처): {stype}")
json.dump(info, open(os.path.join(outdir, "manifest.json"), "w"), ensure_ascii=False, indent=2)
@ -182,6 +329,8 @@ def capture_source(m: dict, source: str, stype: str, cr: dict) -> dict:
# ── 타깃 비어있음 검사 + 적용(부트스트랩) ─────────────────────────────────
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
@ -207,6 +356,16 @@ def apply_source(m: dict, source: str, stype: str, cr: dict) -> str:
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")])
else:
die(f"미지원 타입(적용): {stype}")
return "적용 완료(부트스트랩)"
@ -224,7 +383,7 @@ def select_targets(m: dict, names: list[str], require_running: bool) -> list[tup
out = []
for s, t in chosen:
if t not in SUPPORTED:
print(f" · {s}({t}) — 마이그레이션 미지원(다음 단계: redis/solr/rabbitmq/oracle). 건너뜀")
print(f" · {s}({t}) — 마이그레이션 미지원(다음 단계: {', '.join(UNSUPPORTED_NEXT)}). 건너뜀")
continue
if require_running and not _running(m, s):
print(f" · {s} — dev 컨테이너 미기동('yakcloud dev up {s}' 먼저). 건너뜀")