[test]Add concurrent test in ci (#27633)

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

tests/python_client/chaos/checker.py

@@ -342,7 +342,7 @@ class Checker:
         checker_name = self.__class__.__name__
         checkers_result = f"{checker_name}, succ_rate: {succ_rate:.2f}, total: {total:03d}, average_time: {average_time:.4f}, max_time: {max_time:.4f}, min_time: {min_time:.4f}"
         log.info(checkers_result)
-        log.info(f"{checker_name} rsp times: {self.rsp_times}")
+        log.debug(f"{checker_name} rsp times: {self.rsp_times}")
         if len(self.fail_records) > 0:
             log.info(f"{checker_name} failed at {self.fail_records}")
         return checkers_result

tests/python_client/chaos/conftest.py

@@ -10,7 +10,6 @@ def pytest_addoption(parser):
     parser.addoption("--target_number", action="store", default="1", help="target_number")
     parser.addoption("--chaos_duration", action="store", default="7m", help="chaos_duration")
     parser.addoption("--chaos_interval", action="store", default="2m", help="chaos_interval")
-    parser.addoption("--request_duration", action="store", default="10m", help="request_duration")
     parser.addoption("--is_check", action="store", type=bool, default=False, help="is_check")
     parser.addoption("--wait_signal", action="store", type=bool, default=True, help="wait_signal")
 
@@ -55,11 +54,6 @@ def chaos_interval(request):
     return request.config.getoption("--chaos_interval")
 
 
-@pytest.fixture
-def request_duration(request):
-    return request.config.getoption("--request_duration")
-
-
 @pytest.fixture
 def is_check(request):
     return request.config.getoption("--is_check")

tests/python_client/conftest.py

@@ -47,6 +47,7 @@ def pytest_addoption(parser):
     parser.addoption('--minio_host', action='store', default="localhost", help="minio service's ip")
     parser.addoption('--uri', action='store', default="", help="uri for high level api")
     parser.addoption('--token', action='store', default="", help="token for high level api")
+    parser.addoption("--request_duration", action="store", default="10m", help="request_duration")
 
 
 @pytest.fixture
@@ -81,7 +82,7 @@ def secure(request):
 
 @pytest.fixture
 def milvus_ns(request):
-    return request.config.getoption("--milvus_ns") 
+    return request.config.getoption("--milvus_ns")
 
 
 @pytest.fixture
@@ -185,6 +186,10 @@ def uri(request):
 def token(request):
     return request.config.getoption("--token")
 
+@pytest.fixture
+def request_duration(request):
+    return request.config.getoption("--request_duration")
+
 
 """ fixture func """
 

tests/python_client/testcases/test_concurrent.py

@@ -0,0 +1,100 @@
+import time
+import pytest
+import json
+from time import sleep
+from pymilvus import connections
+from chaos.checker import (InsertChecker,
+                           SearchChecker,
+                           QueryChecker,
+                           DeleteChecker,
+                           Op,
+                           ResultAnalyzer
+                           )
+from utils.util_log import test_log as log
+from chaos import chaos_commons as cc
+from common import common_func as cf
+from chaos.chaos_commons import assert_statistic
+from common.common_type import CaseLabel
+from chaos import constants
+from delayed_assert import assert_expectations
+
+
+def get_all_collections():
+    try:
+        with open("/tmp/ci_logs/all_collections.json", "r") as f:
+            data = json.load(f)
+            all_collections = data["all"]
+    except Exception as e:
+        log.error(f"get_all_collections error: {e}")
+        return [None]
+    return all_collections
+
+
+class TestBase:
+    expect_create = constants.SUCC
+    expect_insert = constants.SUCC
+    expect_flush = constants.SUCC
+    expect_compact = constants.SUCC
+    expect_search = constants.SUCC
+    expect_query = constants.SUCC
+    host = '127.0.0.1'
+    port = 19530
+    _chaos_config = None
+    health_checkers = {}
+
+
+class TestOperations(TestBase):
+
+    @pytest.fixture(scope="function", autouse=True)
+    def connection(self, host, port, user, password, milvus_ns):
+        if user and password:
+            connections.connect('default', host=host, port=port, user=user, password=password, secure=True)
+        else:
+            connections.connect('default', host=host, port=port)
+        if connections.has_connection("default") is False:
+            raise Exception("no connections")
+        log.info("connect to milvus successfully")
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+
+    def init_health_checkers(self, collection_name=None):
+        c_name = collection_name
+        checkers = {
+            Op.insert: InsertChecker(collection_name=c_name),
+            Op.search: SearchChecker(collection_name=c_name),
+            Op.query: QueryChecker(collection_name=c_name),
+            Op.delete: DeleteChecker(collection_name=c_name),
+        }
+        self.health_checkers = checkers
+
+    @pytest.fixture(scope="function", params=get_all_collections())
+    def collection_name(self, request):
+        if request.param == [] or request.param == "":
+            pytest.skip("The collection name is invalid")
+        yield request.param
+
+    @pytest.mark.tags(CaseLabel.L3)
+    def test_operations(self, request_duration, collection_name):
+        # start the monitor threads to check the milvus ops
+        log.info("*********************Test Start**********************")
+        log.info(connections.get_connection_addr('default'))
+        c_name = collection_name if collection_name else cf.gen_unique_str("Checker_")
+        self.init_health_checkers(collection_name=c_name)
+        cc.start_monitor_threads(self.health_checkers)
+        log.info("*********************Load Start**********************")
+        request_duration = request_duration.replace("h", "*3600+").replace("m", "*60+").replace("s", "")
+        if request_duration[-1] == "+":
+            request_duration = request_duration[:-1]
+        request_duration = eval(request_duration)
+        for i in range(10):
+            sleep(request_duration//10)
+            for k, v in self.health_checkers.items():
+                v.check_result()
+        time.sleep(60)
+        ra = ResultAnalyzer()
+        ra.get_stage_success_rate()
+        assert_statistic(self.health_checkers)
+        assert_expectations()
+        log.info("*********************Test Completed**********************")

tests/scripts/ci_e2e_4am.sh

@@ -61,13 +61,13 @@ if [ ! -d "${CI_LOG_PATH}" ]; then
   mkdir -p ${CI_LOG_PATH}
 fi
 
-echo "prepare e2e test"  
+echo "prepare e2e test"
-install_pytest_requirements  
+install_pytest_requirements
 
 
 # Pytest is not able to have both --timeout & --workers, so do not add --timeout or --workers in the shell script
 if [[ -n "${TEST_TIMEOUT:-}" ]]; then
-  
+
   timeout  "${TEST_TIMEOUT}" pytest --host ${MILVUS_SERVICE_NAME} --port ${MILVUS_SERVICE_PORT} \
                                      --html=${CI_LOG_PATH}/report.html  --self-contained-html ${@:-}
 else
@@ -77,10 +77,20 @@ fi
 
 # Run bulk insert test
 if [[ -n "${TEST_TIMEOUT:-}" ]]; then
-  
+
   timeout  "${TEST_TIMEOUT}" pytest testcases/test_bulk_insert.py --host ${MILVUS_SERVICE_NAME} --port ${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} \
                                      --html=${CI_LOG_PATH}/report_bulk_insert.html  --self-contained-html
 else
   pytest testcases/test_bulk_insert.py --host ${MILVUS_SERVICE_NAME} --port ${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} \
                                      --html=${CI_LOG_PATH}/report_bulk_insert.html --self-contained-html
-fi
+fi
+
+# Run concurrent test with 10 processes
+if [[ -n "${TEST_TIMEOUT:-}" ]]; then
+
+  timeout  "${TEST_TIMEOUT}" pytest testcases/test_concurrent.py --host ${MILVUS_SERVICE_NAME} --port ${MILVUS_SERVICE_PORT} --count 10 -n 10 \
+                                     --html=${CI_LOG_PATH}/report_concurrent.html  --self-contained-html
+else
+  pytest testcases/test_concurrent.py --host ${MILVUS_SERVICE_NAME} --port ${MILVUS_SERVICE_PORT} --count 10 -n 10 \
+                                     --html=${CI_LOG_PATH}/report_concurrent.html --self-contained-html
+fi