ROX-15724: preserve infra kubeconfig contexts during merge

scripts/dev/infra-sync-merge.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Merge infra kubeconfigs without losing contexts with identical source names."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+
+
+def load_config(path):
+    result = subprocess.run(
+        ["kubectl", "config", "view", "--kubeconfig", str(path), "--raw", "--flatten", "-o", "json"],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return json.loads(result.stdout)
+
+
+def unique_name(name, source, used, prefix=False):
+    candidate = f"{source}/{name}" if prefix or name in used else name
+    index = 2
+    while candidate in used:
+        candidate = f"{source}/{name}-{index}"
+        index += 1
+    used.add(candidate)
+    return candidate
+
+
+def merge_configs(inputs):
+    merged = {
+        "apiVersion": "v1",
+        "kind": "Config",
+        "preferences": {},
+        "clusters": [],
+        "users": [],
+        "contexts": [],
+        "current-context": "",
+    }
+    names = {kind: set() for kind in ("clusters", "users", "contexts")}
+
+    for path, infra in inputs:
+        config = load_config(path)
+        references = {}
+        for kind in ("clusters", "users"):
+            references[kind] = {}
+            for entry in config.get(kind, []):
+                old = entry["name"]
+                if old in references[kind]:
+                    raise ValueError(f"Duplicate {kind} name {old!r} in {path}")
+                new = unique_name(old, path.parent.name, names[kind], prefix=infra and kind == "users")
+                references[kind][old] = new
+                entry["name"] = new
+                merged[kind].append(entry)
+
+        for entry in config.get("contexts", []):
+            context = entry["context"]
+            cluster = context["cluster"]
+            if cluster not in references["clusters"]:
+                raise ValueError(f"Context {entry['name']!r} in {path} references missing cluster {cluster!r}")
+            name = cluster if infra else entry["name"]
+            if name in names["contexts"]:
+                raise ValueError(f"Context {name!r} occurs more than once (in {path}); refusing to overwrite it")
+            names["contexts"].add(name)
+            context["cluster"] = references["clusters"][cluster]
+            if context.get("user"):
+                user = context["user"]
+                if user not in references["users"]:
+                    raise ValueError(f"Context {entry['name']!r} in {path} references missing user {user!r}")
+                context["user"] = references["users"][user]
+            if config.get("current-context") == entry["name"] and not merged["current-context"]:
+                merged["current-context"] = name
+            entry["name"] = name
+            merged["contexts"].append(entry)
+
+    return merged
+
+
+def write_config(config, output):
+    fd, temporary = tempfile.mkstemp(prefix=".kubeconfig-", dir=output.parent)
+    try:
+        with os.fdopen(fd, "w") as stream:
+            json.dump(config, stream)
+            stream.write("\n")
+        if output.exists():
+            backup = Path(f"{output}.bkp")
+            print(f"Backing up previous infra kubeconfig {output} as {backup}")
+            shutil.copy2(output, backup)
+        os.replace(temporary, output)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def main():
+    clusters_dir = Path(sys.argv[1])
+    output = Path(sys.argv[2]).expanduser()
+    extras = [Path(path).expanduser() for path in sys.argv[3].split(os.pathsep) if path]
+    inputs = [(path, True) for path in sorted(clusters_dir.glob("*/kubeconfig"))]
+    inputs.extend((path, False) for path in extras)
+    write_config(merge_configs(inputs), output)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (OSError, ValueError, subprocess.CalledProcessError) as error:
+        sys.exit(f"Failed to merge kubeconfigs: {error}")

scripts/dev/infra-sync.sh

@@ -9,7 +9,7 @@ clusters_dir="${infra_dir}/clusters"
 infra_kubeconfig="${INFRA_KUBECONFIG:-${infra_dir}/kubeconfig}"
 extra_kubeconfigs="${EXTRA_KUBECONFIGS:-}"
 original_kubeconfig="${KUBECONFIG:-}"
-opts=($@)
+opts=("$@")
 
 clean_infra_cluster_cache() {
     echo "Deleting old cluster artifacts..."
@@ -74,33 +74,10 @@ fetch_infra_clusters() {
     fi
 }
 
-kubecfg_merge() {
-    # Backup previous kubeconfig file.
-    infra_kubeconfig_backup="${infra_kubeconfig}.bkp"
-    if [[ -e "${infra_kubeconfig}" ]]; then
-        echo "Backing up previous infra kubeconfig ${infra_kubeconfig} as ${infra_kubeconfig_backup}"
-        mv "${infra_kubeconfig}" "${infra_kubeconfig_backup}"
-    fi
-
-    # Merge all the kubeconfigs.
-    export KUBECONFIG="/dev/null"
-    for cluster_dir in "${clusters_dir}"/*; do
-        if [ -e "${cluster_dir}/kubeconfig" ]; then
-            KUBECONFIG="${KUBECONFIG}:${cluster_dir}/kubeconfig"
-        fi
-    done
-    if [[ -n "${extra_kubeconfigs}" ]]; then
-        KUBECONFIG="${KUBECONFIG}:${extra_kubeconfigs}"
-    fi
-
-    echo "Merging kube configurations into ${infra_kubeconfig}"
-    tmp_infra_kubeconfig=$(mktemp)
-    kubectl config view --flatten > "${tmp_infra_kubeconfig}"
-    mv "${tmp_infra_kubeconfig}" "${infra_kubeconfig}"
-}
-
 fetch_infra_clusters
-kubecfg_merge
+echo "Merging kube configurations into ${infra_kubeconfig}"
+python3 "$(dirname "${BASH_SOURCE[0]}")/infra-sync-merge.py" \
+    "${clusters_dir}" "${infra_kubeconfig}" "${extra_kubeconfigs}"
 
 if [[ "${original_kubeconfig}" != "${infra_kubeconfig}" ]]; then
     echo

scripts/dev/test_infra_sync_merge.py

@@ -0,0 +1,108 @@
+import importlib.util
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+
+HELPER = Path(__file__).with_name("infra-sync-merge.py")
+spec = importlib.util.spec_from_file_location("infra_sync_merge", HELPER)
+merge = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(merge)
+
+
+def kubeconfig(path, cluster, context="default", user="admin"):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({
+        "apiVersion": "v1",
+        "kind": "Config",
+        "clusters": [{"name": cluster, "cluster": {"server": f"https://{cluster}.example.com"}}],
+        "users": [{"name": user, "user": {"token": f"token-{cluster}"}}],
+        "contexts": [{"name": context, "context": {"cluster": cluster, "user": user}}],
+        "current-context": context,
+    }))
+
+
+@unittest.skipUnless(shutil.which("kubectl"), "kubectl is required")
+class MergeTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.root = Path(self.directory.name)
+
+    def test_same_context_name_keeps_both_clusters_and_credentials(self):
+        first = self.root / "clusters" / "first" / "kubeconfig"
+        second = self.root / "clusters" / "second" / "kubeconfig"
+        kubeconfig(first, "first")
+        kubeconfig(second, "second")
+
+        result = merge.merge_configs([(first, True), (second, True)])
+
+        self.assertEqual({item["name"] for item in result["contexts"]}, {"first", "second"})
+        self.assertEqual(result["current-context"], "first")
+        contexts = {item["name"]: item["context"] for item in result["contexts"]}
+        users = {item["name"]: item["user"] for item in result["users"]}
+        self.assertEqual(contexts["first"]["cluster"], "first")
+        self.assertEqual(contexts["second"]["cluster"], "second")
+        self.assertEqual(contexts["first"]["user"], "first/admin")
+        self.assertEqual(contexts["second"]["user"], "second/admin")
+        self.assertEqual(users[contexts["first"]["user"]]["token"], "token-first")
+        self.assertEqual(users[contexts["second"]["user"]]["token"], "token-second")
+
+    def test_extra_context_is_not_renamed_and_conflicting_references_are_distinct(self):
+        infra = self.root / "clusters" / "first" / "kubeconfig"
+        extra = self.root / "extra" / "kubeconfig"
+        kubeconfig(infra, "first")
+        kubeconfig(extra, "first", context="personal")
+
+        result = merge.merge_configs([(infra, True), (extra, False)])
+
+        self.assertEqual({item["name"] for item in result["contexts"]}, {"first", "personal"})
+        self.assertEqual(len({item["context"]["cluster"] for item in result["contexts"]}), 2)
+        self.assertEqual(len({item["context"]["user"] for item in result["contexts"]}), 2)
+
+    def test_infra_user_is_prefixed_without_a_collision(self):
+        infra = self.root / "clusters" / "first" / "kubeconfig"
+        kubeconfig(infra, "first", user="unique")
+
+        result = merge.merge_configs([(infra, True)])
+
+        self.assertEqual(result["users"][0]["name"], "first/unique")
+        self.assertEqual(result["contexts"][0]["context"]["user"], "first/unique")
+
+    def test_context_collision_does_not_replace_existing_output(self):
+        first = self.root / "clusters" / "first" / "kubeconfig"
+        second = self.root / "clusters" / "second" / "kubeconfig"
+        kubeconfig(first, "shared")
+        kubeconfig(second, "shared")
+        output = self.root / "kubeconfig"
+        output.write_text("previous")
+
+        with self.assertRaisesRegex(ValueError, "refusing to overwrite"):
+            merge.write_config(merge.merge_configs([(first, True), (second, True)]), output)
+
+        self.assertEqual(output.read_text(), "previous")
+        self.assertFalse(Path(f"{output}.bkp").exists())
+
+    def test_output_is_private_and_previous_config_is_backed_up(self):
+        first = self.root / "clusters" / "first" / "kubeconfig"
+        kubeconfig(first, "first")
+        output = self.root / "kubeconfig"
+        output.write_text("previous")
+
+        merge.write_config(merge.merge_configs([(first, True)]), output)
+
+        self.assertEqual(json.loads(output.read_text())["contexts"][0]["name"], "first")
+        self.assertEqual(Path(f"{output}.bkp").read_text(), "previous")
+        self.assertEqual(output.stat().st_mode & 0o777, 0o600)
+        parsed = subprocess.run(
+            ["kubectl", "config", "view", "--kubeconfig", str(output), "--raw", "-o", "json"],
+            check=True, capture_output=True, text=True,
+        )
+        self.assertEqual(json.loads(parsed.stdout)["contexts"][0]["name"], "first")
+
+
+if __name__ == "__main__":
+    unittest.main()