[test]Add all succ check after rolling update (#26638)

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>

tests/python_client/chaos/chaos_commons.py

@@ -110,6 +110,6 @@ def assert_statistic(
                 f"Expect Succ: {str(k)} succ rate {succ_rate}, total: {total}, average time: {average_time:.4f}"
             )
             pytest.assume(
-                succ_rate > succ_rate_threshold and total > 2,
+                succ_rate >= succ_rate_threshold and total > 2,
                 f"Expect Succ: {str(k)} succ rate {succ_rate}, total: {total}, average time: {average_time:.4f}",
             )

tests/python_client/chaos/checker.py

@@ -1,3 +1,4 @@
+import pytest
 from enum import Enum
 from random import randint
 import time
@@ -164,10 +165,15 @@ class Checker:
         self._keep_running = False
         time.sleep(10)
 
+    def resume(self):
+        self._keep_running = True
+        time.sleep(10)
+
     def reset(self):
         self._succ = 0
         self._fail = 0
         self.rsp_times = []
+        self.fail_records = []
         self.average_time = 0
 
     def get_rto(self):
@@ -395,7 +401,7 @@ class InsertChecker(Checker):
         for r in res:
             d = r[f"{ct.default_int64_field_name}"]
             data_in_server.append(d)
-        assert set(data_in_server) == set(data_in_client)
+        pytest.assume(set(data_in_server) == set(data_in_client))
 
 
 class CreateChecker(Checker):

tests/python_client/chaos/testcases/test_single_request_operation.py

@@ -1,3 +1,5 @@
+import time
+
 import pytest
 from time import sleep
 from pymilvus import connections
@@ -11,6 +13,7 @@ from chaos.checker import (CreateChecker,
                            DropChecker,
                            Op)
 from utils.util_log import test_log as log
+from utils.util_k8s import wait_pods_ready, get_milvus_instance_name
 from chaos import chaos_commons as cc
 from common.common_type import CaseLabel
 from chaos.chaos_commons import assert_statistic
@@ -34,7 +37,7 @@ class TestBase:
 class TestOperations(TestBase):
 
     @pytest.fixture(scope="function", autouse=True)
-    def connection(self, host, port, user, password):
+    def connection(self, host, port, user, password, milvus_ns):
         if user and password:
             # log.info(f"connect to {host}:{port} with user {user} and password {password}")
             connections.connect('default', host=host, port=port, user=user, password=password, secure=True)
@@ -47,6 +50,8 @@ class TestOperations(TestBase):
         self.port = port
         self.user = user
         self.password = password
+        self.milvus_ns = milvus_ns
+        self.release_name = get_milvus_instance_name(self.milvus_ns, milvus_sys=self.milvus_sys)
 
     def init_health_checkers(self, collection_name=None):
         c_name = collection_name
@@ -83,4 +88,7 @@ class TestOperations(TestBase):
         if is_check:
             assert_statistic(self.health_checkers, succ_rate_threshold=0.98)
             assert_expectations()
+        # wait all pod ready
+        wait_pods_ready(self.milvus_ns, f"app.kubernetes.io/instance={self.release_name}")
+        time.sleep(60)
         log.info("*********************Chaos Test Completed**********************")

tests/python_client/chaos/testcases/test_single_request_operation_for_rolling_update.py

@@ -3,6 +3,8 @@ from pathlib import Path
 import subprocess
 import pytest
 from time import sleep
+
+from yaml import full_load
 from pymilvus import connections, utility
 from chaos.checker import (CreateChecker,
                            InsertChecker,
@@ -13,6 +15,7 @@ from chaos.checker import (CreateChecker,
                            DeleteChecker,
                            DropChecker,
                            Op)
+from utils.util_k8s import wait_pods_ready
 from utils.util_log import test_log as log
 from chaos import chaos_commons as cc
 from common.common_type import CaseLabel
@@ -120,7 +123,9 @@ class TestOperations(TestBase):
                 res = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                 stdout, stderr = res.communicate()
                 log.info(f"{cmd}, stdout: {stdout}, stderr: {stderr}")
-
+                # reset all
+                for k, v in self.health_checkers.items():
+                    v.reset()
             for k, v in self.health_checkers.items():
                 v.check_result()
         for k, v in self.health_checkers.items():
@@ -131,10 +136,8 @@ class TestOperations(TestBase):
             log.info(f"{k} failed request: {v.fail_records}")
         for k, v in self.health_checkers.items():  
             log.info(f"{k} rto: {v.get_rto()}")
-
         if is_check:
             assert_statistic(self.health_checkers, succ_rate_threshold=0.98)
-            assert_expectations()
             # get each checker's rto
             for k, v in self.health_checkers.items():
                 log.info(f"{k} rto: {v.get_rto()}")
@@ -146,4 +149,25 @@ class TestOperations(TestBase):
                 log.info("*********************Verify Data Completeness**********************")
                 self.health_checkers[Op.insert].verify_data_completeness()
 
-        log.info("*********************Chaos Test Completed**********************")
+        #
+        for k, v in self.health_checkers.items():
+            v.reset()
+        # wait all pod running
+        file_path = f"{str(Path(__file__).parent.parent.parent)}/deploy/milvus_crd.yaml"
+        with open(file_path, "r") as f:
+            config = full_load(f)
+        meta_name = config["metadata"]["name"]
+        label_selector = f"app.kubernetes.io/instance={meta_name}"
+        is_ready = wait_pods_ready("chaos-testing", label_selector)
+        pytest.assume(is_ready is True, f"expect all pods ready but got {is_ready}")
+        cc.start_monitor_threads(self.health_checkers)
+        sleep(60)
+        log.info("check succ rate after rolling update finished")
+        for k, v in self.health_checkers.items():
+            v.check_result()
+        for k, v in self.health_checkers.items():
+            log.info(f"{k} failed request: {v.fail_records}")
+        for k, v in self.health_checkers.items():
+            log.info(f"{k} rto: {v.get_rto()}")
+        assert_statistic(self.health_checkers, succ_rate_threshold=1.0)
+        log.info("*********************Test Completed**********************")