#!/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 from ruamel.yaml.comments import CommentedMap # ----------------------------- # YAML 설정 # ----------------------------- yaml = YAML() yaml.preserve_quotes = True yaml.indent(mapping=2, sequence=4, offset=2) yaml.width = 4096 # ----------------------------- # kubectl ingress 조회 # ----------------------------- def kubectl_ingress(): out = subprocess.check_output( ["kubectl", "get", "ingress", "--all-namespaces", "-o", "yaml"] ) 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): """ annotations 하위에 그대로 삽입될 문자열을 생성 """ 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']}: {p['value']}") lines.append("") for c in converted: lines.append(f" {c['haproxy']}: {c['value']}") return "\n".join(lines) # ----------------------------- # ingress 변환 로직 # ----------------------------- def migrate_ingress(data, mapping, 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", {}) name = meta.get("name") namespace = meta.get("namespace", "default") ingress_id = f"{namespace}/{name}" if ingress_class: spec["ingressClassName"] = 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["annotations"] = CommentedMap() item["_rendered_annotations"] = render_annotations_block( unsupported, partial, converted ) return new, report # ----------------------------- # 리포트 생성 (Markdown) # ----------------------------- 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") # ----------------------------- # 파일 저장 (split) # ----------------------------- def save_split(data, out_dir): Path(out_dir).mkdir(parents=True, exist_ok=True) for item in data.get("items", []): rendered = item.pop("_rendered_annotations", "") name = item["metadata"]["name"] namespace = item["metadata"].get("namespace", "default") path = Path(out_dir) / f"{namespace}__{name}.yaml" with open(path, "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) # ----------------------------- # 파일 저장 (single) # ----------------------------- def save_single(data, out_path): rendered_map = {} for item in data.get("items", []): rendered_map[id(item)] = item.pop("_rendered_annotations", "") path = Path(out_path) with path.open("w") as f: yaml.dump(data, f) content = path.read_text() for item in data.get("items", []): rendered = rendered_map.get(id(item)) if rendered: 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("--single", action="store_true") parser.add_argument("--split", action="store_true") parser.add_argument("--out", default="output") parser.add_argument("--ingress-class") parser.add_argument( "--report", default="migration-report.md" ) args = parser.parse_args() if not args.single and not args.split: parser.error("one of --single or --split must be specified") ingress = kubectl_ingress() mapping = load_mapping(args.mapping) converted, report = migrate_ingress( ingress, mapping, ingress_class=args.ingress_class ) if args.single: out_file = args.out if args.out.endswith(".yaml") else f"{args.out}.yaml" save_single(converted, out_file) if args.split: 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"Report : {args.report}") if __name__ == "__main__": main()