test: use container kill for CDC failover scenarios (#51479)

## What changed - Replace Pod deletion with Chaos Mesh `container-kill` in CDC force-promote and source-down failover tests. - Kill every regular container in matching Pods while preserving the Pod objects. - Verify that Pod UIDs remain unchanged and each container's restart count increases. - Clean up the generated `PodChaos` resources after fault injection. ## Why Deleting Pods changes their identity and tests Pod recreation rather than container-level recovery. These scenarios need to verify CDC failover behavior when containers restart inside the existing Pods. ## Impact CDC failover tests now exercise container failures while explicitly validating that Kubernetes does not recreate the affected Pods. ## Validation - `ruff check` passed for all modified files. - `ruff format --check` passed for all modified files. - Python compilation passed for all modified files. - Full CDC E2E was not run locally because it requires a Kubernetes environment with Chaos Mesh. --------- Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>
2026-07-21 02:05:41 +00:00 · 2026-07-17 11:08:41 +08:00
parent 09c44f2dbb
commit 43264fc57a
3 changed files with 154 additions and 45 deletions
@@ -1,4 +1,6 @@
+import json
 import logging
+import re
 import time
 from concurrent.futures import ThreadPoolExecutor, wait

@@ -328,35 +330,157 @@ def switchover_helper(request, upstream_client, downstream_client):

@pytest.fixture(scope="session")
 def kubectl_helper(milvus_ns):
-    """Helpers for pod-kill failover scenarios."""
+    """Helpers for container-kill failover scenarios."""
    import subprocess as _sp
    import time as _time

-    def delete_pods(instance_label):
+    def get_pods(instance_label):
        cmd = [
            "kubectl",
-            "delete",
+            "get",
            "pods",
            "-l",
            f"app.kubernetes.io/instance={instance_label}",
            "-n",
            milvus_ns,
-            "--grace-period=0",
-            "--force",
+            "-o",
+            "json",
        ]
        result = _sp.run(cmd, capture_output=True, text=True, check=False)
-        logger.info(
-            f"[KUBECTL] delete pods {instance_label}: rc={result.returncode}, "
-            f"stdout={result.stdout!r}, stderr={result.stderr!r}"
-        )
-        return result
+        if result.returncode != 0:
+            raise RuntimeError(
+                f"failed to list pods for {instance_label}: stdout={result.stdout!r}, stderr={result.stderr!r}"
+            )
+        pods = json.loads(result.stdout).get("items", [])
+        if not pods:
+            raise RuntimeError(f"no pods matched instance {instance_label}")
+        return pods
+
+    def snapshot_containers(pods):
+        snapshot = {}
+        for pod in pods:
+            metadata = pod["metadata"]
+            pod_name = metadata["name"]
+            statuses = {status["name"]: status for status in pod.get("status", {}).get("containerStatuses", [])}
+            containers = {}
+            for container in pod.get("spec", {}).get("containers", []):
+                container_name = container["name"]
+                containers[container_name] = statuses.get(container_name, {}).get("restartCount", -1)
+            snapshot[pod_name] = {
+                "uid": metadata["uid"],
+                "containers": containers,
+            }
+        return snapshot
+
+    def kill_containers(instance_label, timeout=120):
+        """Kill every matching container without deleting its Pod object."""
+        baseline = snapshot_containers(get_pods(instance_label))
+        missing_status = [
+            f"{pod_name}/{container_name}"
+            for pod_name, pod in baseline.items()
+            for container_name, restart_count in pod["containers"].items()
+            if restart_count < 0
+        ]
+        if missing_status:
+            raise RuntimeError(f"containers have no initial status: {missing_status}")
+
+        pods_by_containers = {}
+        for pod_name, pod in baseline.items():
+            container_names = tuple(sorted(pod["containers"]))
+            pods_by_containers.setdefault(container_names, []).append(pod_name)
+
+        safe_instance = re.sub(r"[^a-z0-9-]", "-", instance_label.lower()).strip("-")[:20]
+        run_id = str(_time.time_ns())[-10:]
+        chaos_names = []
+        try:
+            for index, (container_names, pod_names) in enumerate(pods_by_containers.items(), start=1):
+                chaos_name = f"cdc-ck-{safe_instance}-{run_id}-{index}"
+                chaos = {
+                    "apiVersion": "chaos-mesh.org/v1alpha1",
+                    "kind": "PodChaos",
+                    "metadata": {"name": chaos_name, "namespace": milvus_ns},
+                    "spec": {
+                        "selector": {"pods": {milvus_ns: sorted(pod_names)}},
+                        "mode": "all",
+                        "action": "container-kill",
+                        "containerNames": list(container_names),
+                    },
+                }
+                result = _sp.run(
+                    ["kubectl", "create", "-f", "-"],
+                    input=json.dumps(chaos),
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                logger.info(
+                    f"[CONTAINER_KILL] create {chaos_name}: rc={result.returncode}, "
+                    f"pods={sorted(pod_names)}, containers={list(container_names)}, "
+                    f"stdout={result.stdout!r}, stderr={result.stderr!r}"
+                )
+                if result.returncode != 0:
+                    raise RuntimeError(
+                        f"failed to create PodChaos {chaos_name}: stdout={result.stdout!r}, stderr={result.stderr!r}"
+                    )
+                chaos_names.append(chaos_name)
+
+            deadline = _time.time() + timeout
+            pending = []
+            while _time.time() < deadline:
+                current = snapshot_containers(get_pods(instance_label))
+                pending = []
+                for pod_name, original in baseline.items():
+                    observed = current.get(pod_name)
+                    if observed is None:
+                        raise RuntimeError(
+                            f"Pod {pod_name} disappeared during container-kill; the fault must preserve Pod objects"
+                        )
+                    if observed["uid"] != original["uid"]:
+                        raise RuntimeError(
+                            f"Pod {pod_name} was recreated during container-kill: "
+                            f"old UID={original['uid']}, new UID={observed['uid']}"
+                        )
+                    for container_name, restart_count in original["containers"].items():
+                        observed_count = observed["containers"].get(container_name, -1)
+                        if observed_count <= restart_count:
+                            pending.append(f"{pod_name}/{container_name}({restart_count}->{observed_count})")
+                if not pending:
+                    logger.info(
+                        f"[CONTAINER_KILL] all containers restarted for {instance_label}; "
+                        f"Pod UIDs unchanged: {sorted(baseline)}"
+                    )
+                    return baseline
+                _time.sleep(1)
+
+            raise TimeoutError(
+                f"container-kill was not observed for {instance_label} within {timeout}s; pending={pending}"
+            )
+        finally:
+            for chaos_name in chaos_names:
+                result = _sp.run(
+                    [
+                        "kubectl",
+                        "delete",
+                        "podchaos",
+                        chaos_name,
+                        "-n",
+                        milvus_ns,
+                        "--ignore-not-found",
+                        "--wait=false",
+                    ],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                logger.info(f"[CONTAINER_KILL] delete {chaos_name}: rc={result.returncode}")

    def wait_for_pods_ready(instance_label, timeout=300):
        """Wait for pods matching the label to be Ready.

-        Two phases because right after `kubectl delete pods`, the operator
-        hasn't recreated pods yet — `kubectl wait` would error with "no
-        matching resources found". So:
+        Keep the existence poll so this helper also handles an unexpected Pod
+        recreation without letting `kubectl wait` fail with "no matching
+        resources found". The expected container-kill path preserves Pod UIDs.
+        So:
          1. Poll until at least one pod matches.
          2. Then `kubectl wait` for Ready on the remaining time budget.
        """
@@ -408,7 +532,7 @@ def kubectl_helper(milvus_ns):
    class KubectlHelper:
        pass

-    KubectlHelper.delete_pods = staticmethod(delete_pods)
+    KubectlHelper.kill_containers = staticmethod(kill_containers)
    KubectlHelper.wait_for_pods_ready = staticmethod(wait_for_pods_ready)

    return KubectlHelper()
@@ -242,9 +242,9 @@ class TestCDCForcePromote(TestCDCSyncBase):

            assert self.wait_for_sync(check_synced, sync_timeout, f"initial sync {c_name}")

-            # Kill source
-            logger.info(f"[FAILOVER] Killing source pods (instance={source_cluster_id})...")
-            kubectl_helper.delete_pods(source_cluster_id)
+            # Kill source containers while preserving Pod objects.
+            logger.info(f"[FAILOVER] Killing source containers (instance={source_cluster_id})...")
+            kubectl_helper.kill_containers(source_cluster_id)

            # Promote downstream — retry until writable
            self.promote_call_until_writable(
@@ -300,7 +300,7 @@ class TestCDCForcePromote(TestCDCSyncBase):
        downstream_token,
        pchannel_num,
    ):
-        """Promote downstream while downstream pods are bouncing.
+        """Promote downstream while downstream containers are restarting.

        Mirrors snippets test_restart_b_during_force_promote.py: promote runs
        async, restart happens in main thread mid-promote.
@@ -369,10 +369,10 @@ class TestCDCForcePromote(TestCDCSyncBase):
            th = threading.Thread(target=promote_thread, daemon=True)
            th.start()

-            # Mid-promote: kill target pods, wait for them to come back
+            # Mid-promote: kill target containers, wait for them to come back.
            time.sleep(2)
-            logger.info("[FAILOVER] Killing target pods mid-promote...")
-            kubectl_helper.delete_pods(target_cluster_id)
+            logger.info("[FAILOVER] Killing target containers mid-promote...")
+            kubectl_helper.kill_containers(target_cluster_id)
            time.sleep(2)
            kubectl_helper.wait_for_pods_ready(target_cluster_id, timeout=300)

@@ -437,7 +437,7 @@ class TestCDCForcePromote(TestCDCSyncBase):
        downstream_token,
        pchannel_num,
    ):
-        """Promote downstream while upstream pods are bouncing.
+        """Promote downstream while upstream containers are restarting.

        Mirrors snippets test_restart_a_during_force_promote.py.
        """
@@ -505,8 +505,8 @@ class TestCDCForcePromote(TestCDCSyncBase):
            th.start()

            time.sleep(2)
-            logger.info("[FAILOVER] Killing source pods mid-promote...")
-            kubectl_helper.delete_pods(source_cluster_id)
+            logger.info("[FAILOVER] Killing source containers mid-promote...")
+            kubectl_helper.kill_containers(source_cluster_id)
            # Don't wait-ready here — promotion should succeed independently
            # of whether the old primary is back.

@@ -3,7 +3,6 @@ CDC sync tests for topology switchover and failover scenarios.
 """

 import random
-import subprocess
 import threading
 import time

@@ -670,10 +669,10 @@ class TestCDCSyncSwitchover(TestCDCSyncBase):
        switchover_helper,
        source_cluster_id,
        target_cluster_id,
-        milvus_ns,
+        kubectl_helper,
    ):
        """
-        Failover when source goes down: insert 500 records, verify sync, kill source pods,
+        Failover when source goes down: insert 500 records, verify sync, kill source containers,
        verify target count >= 500, wait for source to recover and verify count >= 500.
        """
        start_time = time.time()
@@ -727,26 +726,12 @@ class TestCDCSyncSwitchover(TestCDCSyncBase):
                f"Downstream did not receive 500 records within {sync_timeout}s"
            )

-            # Kill source pods forcefully
-            logger.info(f"[FAILOVER] Killing source pods (instance={source_cluster_id}, ns={milvus_ns})...")
-            kill_cmd = [
-                "kubectl",
-                "delete",
-                "pods",
-                "-l",
-                f"app.kubernetes.io/instance={source_cluster_id}",
-                "-n",
-                milvus_ns,
-                "--grace-period=0",
-                "--force",
-            ]
-            result = subprocess.run(kill_cmd, capture_output=True, text=True)
-            logger.info(
-                f"[FAILOVER] kubectl output: stdout={result.stdout!r}, stderr={result.stderr!r}, rc={result.returncode}"
-            )
+            # Kill source containers while preserving Pod objects.
+            logger.info(f"[FAILOVER] Killing source containers (instance={source_cluster_id})...")
+            kubectl_helper.kill_containers(source_cluster_id)

            # Wait 60 s and verify target still has data
-            logger.info("[FAILOVER] Waiting 60s after pod kill...")
+            logger.info("[FAILOVER] Waiting 60s after container kill...")
            time.sleep(60)

            def check_target_intact():