[Cherry-pick][Test][RayCluster] Test redis cleanup job in the e2e compatibility test (#2026) (#2034)

Author: kevin85421

tests/compatibility-test.py

@@ -12,6 +12,7 @@
 )
 
 from framework.utils import (
+    delete_all_cr,
     get_head_pod,
     get_pod,
     pod_exec_command,
@@ -86,8 +87,9 @@ class RayFTTestCase(unittest.TestCase):
     cluster_template = CONST.REPO_ROOT.joinpath("tests/config/ray-cluster.ray-ft.yaml.template")
     ray_cluster_ns = "default"
 
-    @classmethod
-    def setUpClass(cls):
+    # We need to delete the RayCluster at the end of each test case to test cleanup_redis,
+    # therefore, we use setUp, instead of setUpClass, here to re-create RayCluster for each test case.
+    def setUp(self):
         if not utils.is_feature_supported(ray_version, CONST.RAY_FT):
             raise unittest.SkipTest(f"{CONST.RAY_FT} is not supported")
         K8S_CLUSTER_MANAGER.cleanup()
@@ -96,6 +98,13 @@ def setUpClass(cls):
         operator_manager.prepare_operator()
         utils.create_ray_cluster(RayFTTestCase.cluster_template, ray_version, ray_image)
 
+    def cleanup_redis(self):
+        delete_all_cr(CONST.RAY_CLUSTER_CRD, RayFTTestCase.ray_cluster_ns)
+        utils.wait_for_condition(
+            lambda: shell_subprocess_run("test $(kubectl exec deploy/redis -- redis-cli --no-auth-warning -a 5241590000000000 DBSIZE) = '0'") == 0,
+            timeout=300, retry_interval_ms=1000,
+        )
+
     def test_ray_serve(self):
         """Kill GCS process on the head Pod and then test a deployed Ray Serve model."""
         if not utils.is_feature_supported(ray_version, CONST.RAY_SERVE_FT):
@@ -153,6 +162,8 @@ def test_ray_serve(self):
             show_cluster_info(RayFTTestCase.ray_cluster_ns)
             self.fail(f"Fail to execute test_ray_serve_2.py. The exit code is {exit_code}.")
 
+        self.cleanup_redis()
+
     @unittest.skipIf(
         ray_version == '2.8.0',
         'test_detached_actor is too flaky with Ray 2.8.0 due to '
@@ -233,6 +244,8 @@ def test_detached_actor(self):
             show_cluster_info(RayFTTestCase.ray_cluster_ns)
             self.fail(f"Fail to execute test_detached_actor_2.py. The exit code is {exit_code}.")
 
+        self.cleanup_redis()
+
 class KubeRayHealthCheckTestCase(unittest.TestCase):
     """Test KubeRay health check"""
     cluster_template = CONST.REPO_ROOT.joinpath("tests/config/ray-cluster.sidecar.yaml.template")

tests/framework/prototype.py

@@ -424,11 +424,6 @@ def clean_up(self):
             show_cluster_info(self.namespace)
             raise Exception("RayClusterAddCREvent clean_up() timeout")
 
-        """Make sure the external redis has been cleaned"""
-        if search_path(self.custom_resource_object, ['metadata', 'annotations', 'ray.io/ft-enabled']) == 'true':
-            if shell_subprocess_run("test $(kubectl exec deploy/redis -- redis-cli --no-auth-warning -a $(kubectl get secret redis-password-secret -o jsonpath='{.data.password}' | base64 --decode) DBSIZE) = '0'") != 0:
-                raise Exception("The external redis is not cleaned")
-
         """Delete other resources in the yaml"""
         if self.filepath:
             logger.info("Delete other resources in the YAML")

tests/framework/utils.py

@@ -412,6 +412,11 @@ def pod_exec_command(pod_name, namespace, exec_command, check=True):
         f"kubectl exec {pod_name} -n {namespace} -- {exec_command}", check
     )
 
+def delete_all_cr(crd_name, namespace, check=True):
+    return shell_subprocess_run(
+        f"kubectl delete {crd_name} --all -n {namespace}", check
+    )
+
 
 def start_curl_pod(name: str, namespace: str, timeout_s: int = -1):
     shell_subprocess_run(