#!/usr/bin/env python3 import argparse import subprocess from copy import deepcopy from pathlib import Path from collections import defaultdict from ruamel.yaml import YAML # ----------------------------- # YAML 설정 # ----------------------------- yaml = YAML() yaml.preserve_quotes = True yaml.indent(mapping=2, sequence=4, offset=2) yaml.width = 4096 # ----------------------------- # kubectl ingress 조회 # ----------------------------- def kubectl_ingress(namespace=None): cmd = ["kubectl", "get", "ingress"] if namespace: cmd += ["-n", namespace] else: cmd.append("--all-namespaces") cmd += ["-o", "yaml"] out = subprocess.check_output(cmd) return yaml.load(out) # ----------------------------- # mapping.yaml 로드 # ----------------------------- def load_mapping(path): with open(path) as f: return yaml.load(f) # ----------------------------- # annotation 렌더링 # ----------------------------- def render_annotations_block(unsupported, partial, converted): lines = [] if unsupported: note = unsupported[0]["note"] lines.append(f" # {note}") for u in unsupported: lines.append(f" #{u['key']}: {u['value']}") lines.append("") if partial: for p in partial: lines.append(f" # PARTIAL SUPPORT: {p['note']}") lines.append(f" {p['haproxy']}: \"{str(p['value'])}\"") lines.append("") for c in converted: lines.append(f" {c['haproxy']}: \"{str(c['value'])}\"") return "\n".join(lines) # ----------------------------- # metadata 정리 # ----------------------------- def cleanup_metadata(meta): for k in [ "creationTimestamp", "generation", "resourceVersion", "uid", "managedFields", ]: meta.pop(k, None) # ----------------------------- # ingressClassName 기준 필터링 # ----------------------------- def filter_by_ingress_class(data, from_ingress_class): filtered = deepcopy(data) filtered["items"] = [] for item in data.get("items", []): spec = item.get("spec", {}) current_class = spec.get("ingressClassName", "nginx") if current_class == from_ingress_class: filtered["items"].append(item) return filtered # ----------------------------- # 변환 전 원본 백업 (backup/ns/ingress.yaml) # ----------------------------- def backup_original_ingress(data, backup_dir="./backup"): base = Path(backup_dir) base.mkdir(parents=True, exist_ok=True) for item in data.get("items", []): meta = item.get("metadata", {}) name = meta.get("name") namespace = meta.get("namespace", "default") ns_dir = base / namespace ns_dir.mkdir(parents=True, exist_ok=True) with (ns_dir / f"{name}.yaml").open("w") as f: yaml.dump(item, f) # ----------------------------- # ingress 변환 로직 # ----------------------------- def migrate_ingress(data, mapping, to_ingress_class=None): report = { "converted": [], "partial": [], "unsupported": [], "detail": defaultdict(lambda: {"partial": [], "unsupported": []}), } new = deepcopy(data) for item in new.get("items", []): meta = item.setdefault("metadata", {}) spec = item.setdefault("spec", {}) cleanup_metadata(meta) name = meta.get("name") namespace = meta.get("namespace", "default") ingress_id = f"{namespace}/{name}" if to_ingress_class: spec["ingressClassName"] = to_ingress_class anns = meta.get("annotations", {}) or {} unsupported = [] partial = [] converted = [] for k, v in anns.items(): rule = mapping.get(k) if not rule: converted.append({"haproxy": k, "value": v}) continue support = rule.get("support", "unsupported") haproxy_key = rule.get("haproxy") note = rule.get("note", "no HAProxy equivalent") if support == "full" and haproxy_key: converted.append({"haproxy": haproxy_key, "value": v}) report["converted"].append(k) elif support == "partial" and haproxy_key: partial.append({"haproxy": haproxy_key, "value": v, "note": note}) report["partial"].append(k) report["detail"][ingress_id]["partial"].append( {"nginx": k, "haproxy": haproxy_key, "note": note} ) else: unsupported.append({"key": k, "value": v, "note": note}) report["unsupported"].append(k) report["detail"][ingress_id]["unsupported"].append( {"nginx": k, "value": v, "note": note} ) meta.pop("annotations", None) item["_rendered_annotations"] = render_annotations_block( unsupported, partial, converted ) return new, report # ----------------------------- # 리포트 생성 # ----------------------------- def write_report(report, path): with open(path, "w") as f: f.write("# Ingress Migration Report\n\n") f.write("## Summary\n") f.write(f"- Converted : {len(report['converted'])}\n") f.write(f"- Partial : {len(report['partial'])}\n") f.write(f"- Unsupported: {len(report['unsupported'])}\n\n") f.write("---\n\n") for ingress, detail in report["detail"].items(): if not detail["partial"] and not detail["unsupported"]: continue f.write(f"## {ingress}\n\n") for p in detail["partial"]: f.write( f"- PARTIAL {p['nginx']} → {p['haproxy']}\n" f" - note: {p['note']}\n" ) for u in detail["unsupported"]: f.write( f"- UNSUPPORTED {u['nginx']}\n" f" - note: {u['note']}\n" ) f.write("\n") # ----------------------------- # 파일 저장 (yaml/ns/ingress.yaml) # ----------------------------- def save_split(data, out_dir="./yaml"): base = Path(out_dir) base.mkdir(parents=True, exist_ok=True) for item in data.get("items", []): rendered = item.pop("_rendered_annotations", "") meta = item["metadata"] name = meta["name"] namespace = meta.get("namespace", "default") ns_dir = base / namespace ns_dir.mkdir(parents=True, exist_ok=True) path = ns_dir / f"{name}.yaml" with path.open("w") as f: yaml.dump(item, f) if rendered: content = path.read_text() content = content.replace( "metadata:\n", "metadata:\n annotations:\n" + rendered + "\n", 1, ) path.write_text(content) # ----------------------------- # main # ----------------------------- def main(): parser = argparse.ArgumentParser( description="nginx ingress → HAProxy ingress migration utility" ) parser.add_argument("--mapping", default="mapping.yaml") parser.add_argument("--split", action="store_true") parser.add_argument("--out", default="./yaml") parser.add_argument("--ingress-class", help="target ingressClassName") parser.add_argument( "--from-ingress-class", default="nginx", help="source ingressClassName (default: nginx)", ) parser.add_argument( "--namespace", help="target namespace (default: all namespaces)", ) parser.add_argument("--report", default="migration-report.md") args = parser.parse_args() if not args.split: parser.error("--split must be specified") # 🔹 기본 디렉토리 보장 Path("./backup").mkdir(parents=True, exist_ok=True) Path(args.out).mkdir(parents=True, exist_ok=True) ingress = kubectl_ingress(args.namespace) ingress = filter_by_ingress_class( ingress, args.from_ingress_class, ) backup_original_ingress(ingress) mapping = load_mapping(args.mapping) converted, report = migrate_ingress( ingress, mapping, to_ingress_class=args.ingress_class, ) save_split(converted, args.out) write_report(report, args.report) print("\n=== Migration Summary ===") print(f"Converted : {len(report['converted'])}") print(f"Partial : {len(report['partial'])}") print(f"Unsupported: {len(report['unsupported'])}") print(f"Backup : ./backup//.yaml") print(f"Output : {args.out}//.yaml") print(f"Report : {args.report}") if __name__ == "__main__": main()