test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-03 04:19:18 +08:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity

[skip e2e]Set the case of load balance to L2 (#17223)

Browse Source 
Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>
This commit is contained in:
zhuwenxing 2022-05-25 21:20:00 +08:00 committed by GitHub
parent e44afc7aa3
commit 43792c0615
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
tests/python_client/testcases/test_utility.py
View File

@ -1515,7 +1515,7 @@ class TestUtilityAdvanced(TestcaseBase):
cnt += r.num_rows cnt += r.num_rows
assert cnt == nb assert cnt == nb
@pytest.mark.tags(CaseLabel.L3) @pytest.mark.tags(CaseLabel.L2)
def test_load_balance_normal(self): def test_load_balance_normal(self):
""" """
target: test load balance of collection target: test load balance of collection
@ -1523,6 +1523,10 @@ class TestUtilityAdvanced(TestcaseBase):
expected: sealed_segment_ids is subset of des_sealed_segment_ids expected: sealed_segment_ids is subset of des_sealed_segment_ids
""" """
# init a collection # init a collection
self._connect()
querynode_num = len(MilvusSys().query_nodes)
if querynode_num < 2:
pytest.skip("skip load balance testcase when querynode number less than 2")
c_name = cf.gen_unique_str(prefix) c_name = cf.gen_unique_str(prefix)
collection_w = self.init_collection_wrap(name=c_name) collection_w = self.init_collection_wrap(name=c_name)
ms = MilvusSys() ms = MilvusSys()
@ -1550,6 +1554,9 @@ class TestUtilityAdvanced(TestcaseBase):
# get segments distribution after load balance # get segments distribution after load balance
res, _ = self.utility_wrap.get_query_segment_info(c_name) res, _ = self.utility_wrap.get_query_segment_info(c_name)
segment_distribution = cf.get_segment_distribution(res) segment_distribution = cf.get_segment_distribution(res)
sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
# assert src node has no sealed segments
assert sealed_segment_ids_after_load_banalce == []
des_sealed_segment_ids = [] des_sealed_segment_ids = []
for des_node_id in des_node_ids: for des_node_id in des_node_ids:
des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"] des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
@ -1667,17 +1674,20 @@ class TestUtilityAdvanced(TestcaseBase):
check_task=CheckTasks.err_res, check_task=CheckTasks.err_res,
check_items={ct.err_code: 1, ct.err_msg: "is not exist"}) check_items={ct.err_code: 1, ct.err_msg: "is not exist"})
@pytest.mark.tags(CaseLabel.L3) @pytest.mark.tags(CaseLabel.L2)
def test_load_balance_in_one_group(self): def test_load_balance_in_one_group(self):
""" """
target: test load balance of collection in one group target: test load balance of collection in one group
method: init a collection, load with multi replicas and load balance among the querynodes in one group method: init a collection, load with multi replicas and load balance among the querynodes in one group
expected: load balance successfully expected: load balance successfully
""" """
self._connect()
querynode_num = len(MilvusSys().query_nodes)
if querynode_num < 3:
pytest.skip("skip load balance for multi replicas testcase when querynode number less than 3")
# init a collection # init a collection
c_name = cf.gen_unique_str(prefix) c_name = cf.gen_unique_str(prefix)
collection_w = self.init_collection_wrap(name=c_name) collection_w = self.init_collection_wrap(name=c_name)
ms = MilvusSys()
nb = 3000 nb = 3000
df = cf.gen_default_dataframe_data(nb) df = cf.gen_default_dataframe_data(nb)
collection_w.insert(df) collection_w.insert(df)
@ -1695,10 +1705,14 @@ class TestUtilityAdvanced(TestcaseBase):
if len(g.group_nodes) >= 2: if len(g.group_nodes) >= 2:
group_nodes = list(g.group_nodes) group_nodes = list(g.group_nodes)
break break
src_node_id = group_nodes[0]
dst_node_ids = group_nodes[1:]
res, _ = self.utility_wrap.get_query_segment_info(c_name) res, _ = self.utility_wrap.get_query_segment_info(c_name)
segment_distribution = cf.get_segment_distribution(res) segment_distribution = cf.get_segment_distribution(res)
group_nodes = sorted(group_nodes,
key=lambda x: len(
segment_distribution[x]["sealed"])
if x in segment_distribution else 0, reverse=True)
src_node_id = group_nodes[0]
dst_node_ids = group_nodes[1:]
sealed_segment_ids = segment_distribution[src_node_id]["sealed"] sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
# load balance # load balance
self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids) self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids)
@ -1706,7 +1720,7 @@ class TestUtilityAdvanced(TestcaseBase):
res, _ = self.utility_wrap.get_query_segment_info(c_name) res, _ = self.utility_wrap.get_query_segment_info(c_name)
segment_distribution = cf.get_segment_distribution(res) segment_distribution = cf.get_segment_distribution(res)
sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"] sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
# assert # assert src node has no sealed segments
assert sealed_segment_ids_after_load_banalce == [] assert sealed_segment_ids_after_load_banalce == []
des_sealed_segment_ids = [] des_sealed_segment_ids = []
for des_node_id in dst_node_ids: for des_node_id in dst_node_ids: