From 7cce7da526801eaee99a8b39837ac6058dfc9b6e Mon Sep 17 00:00:00 2001 From: "wenzhouwww@live.cn" Date: Fri, 22 Jul 2022 16:29:22 +0800 Subject: [PATCH] add test case stop follower and leader --- ...4dnode1mnode_basic_replica3_insertdatas.py | 179 ++++++ ...replica3_insertdatas_stop_follower_sync.py | 484 ++++++++++++++++ ...lica3_insertdatas_stop_follower_unsync.py} | 126 +++- ..._basic_replica3_insertdatas_stop_leader.py | 548 ++++++++++++++++++ 4 files changed, 1332 insertions(+), 5 deletions(-) create mode 100644 tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas.py create mode 100644 tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_sync.py rename tests/system-test/6-cluster/vnode/{4dnode1mnode_basic_replica3_insertdatas_stop_follower.py => 4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync.py} (71%) create mode 100644 tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader.py diff --git a/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas.py b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas.py new file mode 100644 index 0000000000..00bd8a48d9 --- /dev/null +++ b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas.py @@ -0,0 +1,179 @@ +# author : wenzhouwww +from ssl import ALERT_DESCRIPTION_CERTIFICATE_UNOBTAINABLE +import taos +import sys +import time +import os + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import TDDnodes +from util.dnodes import TDDnode +from util.cluster import * + +import time +import socket +import subprocess +sys.path.append(os.path.dirname(__file__)) + +class TDTestCase: + def init(self,conn ,logSql): + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + self.host = socket.gethostname() + self.mnode_list = {} + self.dnode_list = {} + self.ts = 1483200000000 + self.db_name ='testdb' + self.replica = 3 + self.vgroups = 2 + self.tb_nums = 10 + self.row_nums = 100 + + def getBuildPath(self): + selfPath = os.path.dirname(os.path.realpath(__file__)) + if ("community" in selfPath): + projPath = selfPath[:selfPath.find("community")] + else: + projPath = selfPath[:selfPath.find("tests")] + + for root, dirs, files in os.walk(projPath): + if ("taosd" in files): + rootRealPath = os.path.dirname(os.path.realpath(root)) + if ("packaging" not in rootRealPath): + buildPath = root[:len(root) - len("/build/bin")] + break + return buildPath + + def check_setup_cluster_status(self): + tdSql.query("show mnodes") + for mnode in tdSql.queryResult: + name = mnode[1] + info = mnode + self.mnode_list[name] = info + + tdSql.query("show dnodes") + for dnode in tdSql.queryResult: + name = dnode[1] + info = dnode + self.dnode_list[name] = info + + count = 0 + is_leader = False + mnode_name = '' + for k,v in self.mnode_list.items(): + count +=1 + # only for 1 mnode + mnode_name = k + + if v[2] =='leader': + is_leader=True + + if count==1 and is_leader: + tdLog.info("===== depoly cluster success with 1 mnode as leader =====") + else: + tdLog.exit("===== depoly cluster fail with 1 mnode as leader =====") + + for k ,v in self.dnode_list.items(): + if k == mnode_name: + if v[3]==0: + tdLog.info("===== depoly cluster mnode only success at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + tdLog.exit("===== depoly cluster mnode only fail at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + continue + + def create_db_check_vgroups(self): + + tdSql.execute("drop database if exists test") + tdSql.execute("create database if not exists test replica 1 duration 300") + tdSql.execute("use test") + tdSql.execute( + '''create table stb1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + tags (t1 int) + ''' + ) + tdSql.execute( + ''' + create table t1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + ''' + ) + + for i in range(5): + tdSql.execute("create table sub_tb_{} using stb1 tags({})".format(i,i)) + tdSql.query("show stables") + tdSql.checkRows(1) + tdSql.query("show tables") + tdSql.checkRows(6) + + tdSql.query("show test.vgroups;") + vgroups_infos = {} # key is id: value is info list + for vgroup_info in tdSql.queryResult: + vgroup_id = vgroup_info[0] + tmp_list = [] + for role in vgroup_info[3:-4]: + if role in ['leader','follower']: + tmp_list.append(role) + vgroups_infos[vgroup_id]=tmp_list + + for k , v in vgroups_infos.items(): + if len(v) ==1 and v[0]=="leader": + tdLog.info(" === create database replica only 1 role leader check success of vgroup_id {} ======".format(k)) + else: + tdLog.exit(" === create database replica only 1 role leader check fail of vgroup_id {} ======".format(k)) + + def create_db_replica_3_insertdatas(self, dbname, replica_num ,vgroup_nums ,tb_nums , row_nums ): + drop_db_sql = "drop database if exists {}".format(dbname) + create_db_sql = "create database {} replica {} vgroups {}".format(dbname,replica_num,vgroup_nums) + + tdLog.info(" ==== create database {} and insert rows begin =====".format(dbname)) + tdSql.execute(drop_db_sql) + tdSql.execute(create_db_sql) + tdSql.execute("use {}".format(dbname)) + tdSql.execute( + '''create table stb1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(32),c9 nchar(32), c10 timestamp) + tags (t1 int) + ''' + ) + tdSql.execute( + ''' + create table t1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(32),c9 nchar(32), c10 timestamp) + ''' + ) + + for i in range(tb_nums): + sub_tbname = "sub_tb_{}".format(i) + tdSql.execute("create table {} using stb1 tags({})".format(sub_tbname,i)) + # insert datas about new database + + for row_num in range(row_nums): + ts = self.ts + 1000*row_num + tdSql.execute(f"insert into {sub_tbname} values ({ts}, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + + tdLog.info(" ==== create database {} and insert rows execute end =====".format(dbname)) + + def check_insert_status(self, dbname, tb_nums , row_nums): + tdSql.execute("use {}".format(dbname)) + tdSql.query("select count(*) from {}.{}".format(dbname,'stb1')) + tdSql.checkData(0 , 0 , tb_nums*row_nums) + tdSql.query("select distinct tbname from {}.{}".format(dbname,'stb1')) + tdSql.checkRows(tb_nums) + + def run(self): + self.check_setup_cluster_status() + self.create_db_check_vgroups() + self.create_db_replica_3_insertdatas(self.db_name , self.replica , self.vgroups , self.tb_nums , self.row_nums) + self.check_insert_status(self.db_name , self.tb_nums , self.row_nums) + + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) \ No newline at end of file diff --git a/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_sync.py b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_sync.py new file mode 100644 index 0000000000..5db1ced199 --- /dev/null +++ b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_sync.py @@ -0,0 +1,484 @@ +# author : wenzhouwww +from ssl import ALERT_DESCRIPTION_CERTIFICATE_UNOBTAINABLE +import taos +import sys +import time +import os + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import TDDnodes +from util.dnodes import TDDnode +from util.cluster import * + +import datetime +import inspect +import time +import socket +import subprocess +import threading +sys.path.append(os.path.dirname(__file__)) + +class TDTestCase: + def init(self,conn ,logSql): + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + self.host = socket.gethostname() + self.mnode_list = {} + self.dnode_list = {} + self.ts = 1483200000000 + self.ts_step =1000 + self.db_name ='testdb' + self.replica = 3 + self.vgroups = 1 + self.tb_nums = 10 + self.row_nums = 100 + self.stop_dnode_id = None + self.loop_restart_times = 5 + self.current_thread = None + self.max_restart_time = 10 + self.try_check_times = 10 + + def getBuildPath(self): + selfPath = os.path.dirname(os.path.realpath(__file__)) + if ("community" in selfPath): + projPath = selfPath[:selfPath.find("community")] + else: + projPath = selfPath[:selfPath.find("tests")] + + for root, dirs, files in os.walk(projPath): + if ("taosd" in files): + rootRealPath = os.path.dirname(os.path.realpath(root)) + if ("packaging" not in rootRealPath): + buildPath = root[:len(root) - len("/build/bin")] + break + return buildPath + + def check_setup_cluster_status(self): + tdSql.query("show mnodes") + for mnode in tdSql.queryResult: + name = mnode[1] + info = mnode + self.mnode_list[name] = info + + tdSql.query("show dnodes") + for dnode in tdSql.queryResult: + name = dnode[1] + info = dnode + self.dnode_list[name] = info + + count = 0 + is_leader = False + mnode_name = '' + for k,v in self.mnode_list.items(): + count +=1 + # only for 1 mnode + mnode_name = k + + if v[2] =='leader': + is_leader=True + + if count==1 and is_leader: + tdLog.info("===== depoly cluster success with 1 mnode as leader =====") + else: + tdLog.exit("===== depoly cluster fail with 1 mnode as leader =====") + + for k ,v in self.dnode_list.items(): + if k == mnode_name: + if v[3]==0: + tdLog.info("===== depoly cluster mnode only success at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + tdLog.exit("===== depoly cluster mnode only fail at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + continue + + def create_db_check_vgroups(self): + + tdSql.execute("drop database if exists test") + tdSql.execute("create database if not exists test replica 1 duration 300") + tdSql.execute("use test") + tdSql.execute( + '''create table stb1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + tags (t1 int) + ''' + ) + tdSql.execute( + ''' + create table t1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + ''' + ) + + for i in range(5): + tdSql.execute("create table sub_tb_{} using stb1 tags({})".format(i,i)) + tdSql.query("show stables") + tdSql.checkRows(1) + tdSql.query("show tables") + tdSql.checkRows(6) + + tdSql.query("show test.vgroups;") + vgroups_infos = {} # key is id: value is info list + for vgroup_info in tdSql.queryResult: + vgroup_id = vgroup_info[0] + tmp_list = [] + for role in vgroup_info[3:-4]: + if role in ['leader','follower']: + tmp_list.append(role) + vgroups_infos[vgroup_id]=tmp_list + + for k , v in vgroups_infos.items(): + if len(v) ==1 and v[0]=="leader": + tdLog.info(" === create database replica only 1 role leader check success of vgroup_id {} ======".format(k)) + else: + tdLog.exit(" === create database replica only 1 role leader check fail of vgroup_id {} ======".format(k)) + + def create_database(self, dbname, replica_num ,vgroup_nums ): + drop_db_sql = "drop database if exists {}".format(dbname) + create_db_sql = "create database {} replica {} vgroups {}".format(dbname,replica_num,vgroup_nums) + + tdLog.info(" ==== create database {} and insert rows begin =====".format(dbname)) + tdSql.execute(drop_db_sql) + tdSql.execute(create_db_sql) + tdSql.execute("use {}".format(dbname)) + + def create_stable_insert_datas(self,dbname ,stablename , tb_nums , row_nums): + tdSql.execute("use {}".format(dbname)) + tdSql.execute( + '''create table {} + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(32),c9 nchar(32), c10 timestamp) + tags (t1 int) + '''.format(stablename) + ) + + for i in range(tb_nums): + sub_tbname = "sub_{}_{}".format(stablename,i) + tdSql.execute("create table {} using {} tags({})".format(sub_tbname, stablename ,i)) + # insert datas about new database + + for row_num in range(row_nums): + ts = self.ts + self.ts_step*row_num + tdSql.execute(f"insert into {sub_tbname} values ({ts}, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + + tdLog.info(" ==== stable {} insert rows execute end =====".format(stablename)) + + def append_rows_of_exists_tables(self,dbname ,stablename , tbname , append_nums ): + + tdSql.execute("use {}".format(dbname)) + + for row_num in range(append_nums): + tdSql.execute(f"insert into {tbname} values (now, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + # print(f"insert into {tbname} values (now, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + tdLog.info(" ==== append new rows of table {} belongs to stable {} execute end =====".format(tbname,stablename)) + os.system("taos -s 'select count(*) from {}.{}';".format(dbname,stablename)) + + def check_insert_rows(self, dbname, stablename , tb_nums , row_nums, append_rows): + + tdSql.execute("use {}".format(dbname)) + + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) + + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups;'".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + tdLog.info(" ==== check insert rows first failed , this is {}_th retry check rows of database {}".format(count , dbname)) + count += 1 + + + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups;'".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + tdLog.info(" ==== check insert tbnames first failed , this is {}_th retry check tbnames of database {}".format(count , dbname)) + count += 1 + + def _get_stop_dnode_id(self,dbname): + tdSql.query("show {}.vgroups".format(dbname)) + vgroup_infos = tdSql.queryResult + for vgroup_info in vgroup_infos: + leader_infos = vgroup_info[3:-4] + # print(vgroup_info) + for ind ,role in enumerate(leader_infos): + if role =='follower': + # print(ind,leader_infos) + self.stop_dnode_id = leader_infos[ind-1] + break + + + return self.stop_dnode_id + + def wait_stop_dnode_OK(self): + + def _get_status(): + newTdSql=tdCom.newTdSql() + + status = "" + newTdSql.query("show dnodes") + dnode_infos = newTdSql.queryResult + for dnode_info in dnode_infos: + id = dnode_info[0] + dnode_status = dnode_info[4] + if id == self.stop_dnode_id: + status = dnode_status + break + return status + + status = _get_status() + while status !="offline": + time.sleep(0.1) + status = _get_status() + # tdLog.info("==== stop dnode has not been stopped , endpoint is {}".format(self.stop_dnode)) + tdLog.info("==== stop_dnode has stopped , id is {}".format(self.stop_dnode_id)) + + def wait_start_dnode_OK(self): + + def _get_status(): + newTdSql=tdCom.newTdSql() + status = "" + newTdSql.query("show dnodes") + dnode_infos = newTdSql.queryResult + for dnode_info in dnode_infos: + id = dnode_info[0] + dnode_status = dnode_info[4] + if id == self.stop_dnode_id: + status = dnode_status + break + return status + + status = _get_status() + while status !="ready": + time.sleep(0.1) + status = _get_status() + # tdLog.info("==== stop dnode has not been stopped , endpoint is {}".format(self.stop_dnode)) + tdLog.info("==== stop_dnode has restart , id is {}".format(self.stop_dnode_id)) + + def _parse_datetime(self,timestr): + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S.%f') + except ValueError: + pass + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S') + except ValueError: + pass + + def mycheckRowCol(self, sql, row, col): + caller = inspect.getframeinfo(inspect.stack()[2][0]) + if row < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is smaller than zero" % args) + if col < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is smaller than zero" % args) + if row > tdSql.queryRows: + args = (caller.filename, caller.lineno, sql, row, tdSql.queryRows) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is larger than queryRows:%d" % args) + if col > tdSql.queryCols: + args = (caller.filename, caller.lineno, sql, col, tdSql.queryCols) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is larger than queryCols:%d" % args) + + def mycheckData(self, sql ,row, col, data): + check_status = True + self.mycheckRowCol(sql ,row, col) + if tdSql.queryResult[row][col] != data: + if tdSql.cursor.istype(col, "TIMESTAMP"): + # suppose user want to check nanosecond timestamp if a longer data passed + if (len(data) >= 28): + if pd.to_datetime(tdSql.queryResult[row][col]) == pd.to_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%d == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + if tdSql.queryResult[row][col] == self._parse_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + + if str(tdSql.queryResult[row][col]) == str(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + elif isinstance(data, float) and abs(tdSql.queryResult[row][col] - data) <= 0.000001: + tdLog.info("sql:%s, row:%d col:%d data:%f == expect:%f" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, row, col, tdSql.queryResult[row][col], data) + tdLog.info("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args) + + check_status = False + + if data is None: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, str): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, datetime.date): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, float): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%d" % + (sql, row, col, tdSql.queryResult[row][col], data)) + + return check_status + + def mycheckRows(self, sql, expectRows): + check_status = True + if len(tdSql.queryResult) == expectRows: + tdLog.info("sql:%s, queryRows:%d == expect:%d" % (sql, len(tdSql.queryResult), expectRows)) + return True + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, len(tdSql.queryResult), expectRows) + tdLog.info("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args) + check_status = False + return check_status + + def sync_run_case(self): + # stop follower and insert datas , update tables and create new stables + tdDnodes=cluster.dnodes + for loop in range(self.loop_restart_times): + db_name = "sync_db_{}".format(loop) + stablename = 'stable_{}'.format(loop) + self.create_database(dbname = db_name ,replica_num= self.replica , vgroup_nums= 1) + self.create_stable_insert_datas(dbname = db_name , stablename = stablename , tb_nums= 10 ,row_nums= 10 ) + self.stop_dnode_id = self._get_stop_dnode_id(db_name) + + # check rows of datas + + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + # begin stop dnode + start = time.time() + tdDnodes[self.stop_dnode_id-1].stoptaosd() + + self.wait_stop_dnode_OK() + + # append rows of stablename when dnode stop + + tbname = "sub_{}_{}".format(stablename , 0) + tdLog.info(" ==== begin append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.append_rows_of_exists_tables(db_name ,stablename , tbname , 100 ) + tdLog.info(" ==== check append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=100) + + # create new stables + tdLog.info(" ==== create new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb1' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb1' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + # begin start dnode + tdDnodes[self.stop_dnode_id-1].starttaosd() + self.wait_start_dnode_OK() + end = time.time() + time_cost = int(end -start) + if time_cost > self.max_restart_time: + tdLog.exit(" ==== restart dnode {} cost too much time , please check ====".format(self.stop_dnode_id)) + + # create new stables again + tdLog.info(" ==== create new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb2' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb2' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + def unsync_run_case(self): + + def _restart_dnode_of_db_unsync(dbname): + start = time.time() + tdDnodes=cluster.dnodes + self.stop_dnode_id = self._get_stop_dnode_id(dbname) + # begin restart dnode + tdDnodes[self.stop_dnode_id-1].stoptaosd() + self.wait_stop_dnode_OK() + tdDnodes[self.stop_dnode_id-1].starttaosd() + self.wait_start_dnode_OK() + end = time.time() + time_cost = int(end-start) + + if time_cost > self.max_restart_time: + tdLog.exit(" ==== restart dnode {} cost too much time , please check ====".format(self.stop_dnode_id)) + + + def _create_threading(dbname): + self.current_thread = threading.Thread(target=_restart_dnode_of_db_unsync, args=(dbname,)) + return self.current_thread + + + ''' + in this mode , it will be extra threading control start or stop dnode , insert will always going with not care follower online or alive + ''' + tdDnodes=cluster.dnodes + for loop in range(self.loop_restart_times): + db_name = "unsync_db_{}".format(loop) + stablename = 'stable_{}'.format(loop) + self.create_database(dbname = db_name ,replica_num= self.replica , vgroup_nums= 1) + self.create_stable_insert_datas(dbname = db_name , stablename = stablename , tb_nums= 10 ,row_nums= 10 ) + + tdLog.info(" ===== restart dnode of database {} in an unsync threading ===== ".format(db_name)) + + # create sync threading and start it + self.current_thread = _create_threading(db_name) + self.current_thread.start() + + # check rows of datas + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + tbname = "sub_{}_{}".format(stablename , 0) + tdLog.info(" ==== begin append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.append_rows_of_exists_tables(db_name ,stablename , tbname , 100 ) + tdLog.info(" ==== check append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=100) + + # create new stables + tdLog.info(" ==== create new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb1' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb1' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + # create new stables again + tdLog.info(" ==== create new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb2' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb2' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + self.current_thread.join() + + + def run(self): + + # basic insert and check of cluster + self.check_setup_cluster_status() + self.create_db_check_vgroups() + self.sync_run_case() + # self.unsync_run_case() + + + + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) \ No newline at end of file diff --git a/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower.py b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync.py similarity index 71% rename from tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower.py rename to tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync.py index 63d560587d..854a28b48a 100644 --- a/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower.py +++ b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync.py @@ -12,6 +12,8 @@ from util.dnodes import TDDnodes from util.dnodes import TDDnode from util.cluster import * +import datetime +import inspect import time import socket import subprocess @@ -35,7 +37,8 @@ class TDTestCase: self.stop_dnode_id = None self.loop_restart_times = 5 self.current_thread = None - self.max_restart_time = 20 + self.max_restart_time = 10 + self.try_check_times = 10 def getBuildPath(self): selfPath = os.path.dirname(os.path.realpath(__file__)) @@ -173,10 +176,37 @@ class TDTestCase: def check_insert_rows(self, dbname, stablename , tb_nums , row_nums, append_rows): tdSql.execute("use {}".format(dbname)) + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) - tdSql.checkData(0 , 0 , tb_nums*row_nums+append_rows) + + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups; '".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + tdLog.info(" ==== check insert rows first failed , this is {}_th retry check rows of database {}".format(count , dbname)) + count += 1 + + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) - tdSql.checkRows(tb_nums) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups;'".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + tdLog.info(" ==== check insert tbnames first failed , this is {}_th retry check tbnames of database {}".format(count , dbname)) + count += 1 def _get_stop_dnode_id(self,dbname): tdSql.query("show {}.vgroups".format(dbname)) @@ -238,6 +268,92 @@ class TDTestCase: # tdLog.info("==== stop dnode has not been stopped , endpoint is {}".format(self.stop_dnode)) tdLog.info("==== stop_dnode has restart , id is {}".format(self.stop_dnode_id)) + def _parse_datetime(self,timestr): + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S.%f') + except ValueError: + pass + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S') + except ValueError: + pass + + def mycheckRowCol(self, sql, row, col): + caller = inspect.getframeinfo(inspect.stack()[2][0]) + if row < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is smaller than zero" % args) + if col < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is smaller than zero" % args) + if row > tdSql.queryRows: + args = (caller.filename, caller.lineno, sql, row, tdSql.queryRows) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is larger than queryRows:%d" % args) + if col > tdSql.queryCols: + args = (caller.filename, caller.lineno, sql, col, tdSql.queryCols) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is larger than queryCols:%d" % args) + + def mycheckData(self, sql ,row, col, data): + check_status = True + self.mycheckRowCol(sql ,row, col) + if tdSql.queryResult[row][col] != data: + if tdSql.cursor.istype(col, "TIMESTAMP"): + # suppose user want to check nanosecond timestamp if a longer data passed + if (len(data) >= 28): + if pd.to_datetime(tdSql.queryResult[row][col]) == pd.to_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%d == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + if tdSql.queryResult[row][col] == self._parse_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + + if str(tdSql.queryResult[row][col]) == str(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + elif isinstance(data, float) and abs(tdSql.queryResult[row][col] - data) <= 0.000001: + tdLog.info("sql:%s, row:%d col:%d data:%f == expect:%f" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, row, col, tdSql.queryResult[row][col], data) + tdLog.info("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args) + + check_status = False + + if data is None: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, str): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, datetime.date): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, float): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%d" % + (sql, row, col, tdSql.queryResult[row][col], data)) + + return check_status + + def mycheckRows(self, sql, expectRows): + check_status = True + if len(tdSql.queryResult) == expectRows: + tdLog.info("sql:%s, queryRows:%d == expect:%d" % (sql, len(tdSql.queryResult), expectRows)) + return True + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, len(tdSql.queryResult), expectRows) + tdLog.info("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args) + check_status = False + return check_status + def sync_run_case(self): # stop follower and insert datas , update tables and create new stables tdDnodes=cluster.dnodes @@ -251,7 +367,7 @@ class TDTestCase: # check rows of datas self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=0) - + # begin stop dnode start = time.time() tdDnodes[self.stop_dnode_id-1].stoptaosd() @@ -354,7 +470,7 @@ class TDTestCase: # basic insert and check of cluster self.check_setup_cluster_status() self.create_db_check_vgroups() - self.sync_run_case() + # self.sync_run_case() self.unsync_run_case() diff --git a/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader.py b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader.py new file mode 100644 index 0000000000..9dde4a6382 --- /dev/null +++ b/tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader.py @@ -0,0 +1,548 @@ +# author : wenzhouwww +from ssl import ALERT_DESCRIPTION_CERTIFICATE_UNOBTAINABLE +import taos +import sys +import time +import os + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import TDDnodes +from util.dnodes import TDDnode +from util.cluster import * + +import time +import socket +import subprocess +import threading +sys.path.append(os.path.dirname(__file__)) + +class TDTestCase: + def init(self,conn ,logSql): + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + self.host = socket.gethostname() + self.mnode_list = {} + self.dnode_list = {} + self.ts = 1483200000000 + self.ts_step =1000 + self.db_name ='testdb' + self.replica = 3 + self.vgroups = 1 + self.tb_nums = 10 + self.row_nums = 100 + self.stop_dnode_id = None + self.loop_restart_times = 50 + self.current_thread = None + self.max_restart_time = 5 + + def getBuildPath(self): + selfPath = os.path.dirname(os.path.realpath(__file__)) + if ("community" in selfPath): + projPath = selfPath[:selfPath.find("community")] + else: + projPath = selfPath[:selfPath.find("tests")] + + for root, dirs, files in os.walk(projPath): + if ("taosd" in files): + rootRealPath = os.path.dirname(os.path.realpath(root)) + if ("packaging" not in rootRealPath): + buildPath = root[:len(root) - len("/build/bin")] + break + return buildPath + + def _parse_datetime(self,timestr): + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S.%f') + except ValueError: + pass + try: + return datetime.datetime.strptime(timestr, '%Y-%m-%d %H:%M:%S') + except ValueError: + pass + + def mycheckRowCol(self, sql, row, col): + caller = inspect.getframeinfo(inspect.stack()[2][0]) + if row < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is smaller than zero" % args) + if col < 0: + args = (caller.filename, caller.lineno, sql, row) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is smaller than zero" % args) + if row > tdSql.queryRows: + args = (caller.filename, caller.lineno, sql, row, tdSql.queryRows) + tdLog.exit("%s(%d) failed: sql:%s, row:%d is larger than queryRows:%d" % args) + if col > tdSql.queryCols: + args = (caller.filename, caller.lineno, sql, col, tdSql.queryCols) + tdLog.exit("%s(%d) failed: sql:%s, col:%d is larger than queryCols:%d" % args) + + def mycheckData(self, sql ,row, col, data): + check_status = True + self.mycheckRowCol(sql ,row, col) + if tdSql.queryResult[row][col] != data: + if tdSql.cursor.istype(col, "TIMESTAMP"): + # suppose user want to check nanosecond timestamp if a longer data passed + if (len(data) >= 28): + if pd.to_datetime(tdSql.queryResult[row][col]) == pd.to_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%d == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + if tdSql.queryResult[row][col] == self._parse_datetime(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + + if str(tdSql.queryResult[row][col]) == str(data): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + elif isinstance(data, float) and abs(tdSql.queryResult[row][col] - data) <= 0.000001: + tdLog.info("sql:%s, row:%d col:%d data:%f == expect:%f" % + (sql, row, col, tdSql.queryResult[row][col], data)) + return + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, row, col, tdSql.queryResult[row][col], data) + tdLog.info("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args) + + check_status = False + + if data is None: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, str): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, datetime.date): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + elif isinstance(data, float): + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % + (sql, row, col, tdSql.queryResult[row][col], data)) + else: + tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%d" % + (sql, row, col, tdSql.queryResult[row][col], data)) + + return check_status + + def mycheckRows(self, sql, expectRows): + check_status = True + if len(tdSql.queryResult) == expectRows: + tdLog.info("sql:%s, queryRows:%d == expect:%d" % (sql, len(tdSql.queryResult), expectRows)) + return True + else: + caller = inspect.getframeinfo(inspect.stack()[1][0]) + args = (caller.filename, caller.lineno, sql, len(tdSql.queryResult), expectRows) + tdLog.info("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args) + check_status = False + return check_status + + def check_setup_cluster_status(self): + tdSql.query("show mnodes") + for mnode in tdSql.queryResult: + name = mnode[1] + info = mnode + self.mnode_list[name] = info + + tdSql.query("show dnodes") + for dnode in tdSql.queryResult: + name = dnode[1] + info = dnode + self.dnode_list[name] = info + + count = 0 + is_leader = False + mnode_name = '' + for k,v in self.mnode_list.items(): + count +=1 + # only for 1 mnode + mnode_name = k + + if v[2] =='leader': + is_leader=True + + if count==1 and is_leader: + tdLog.info("===== depoly cluster success with 1 mnode as leader =====") + else: + tdLog.exit("===== depoly cluster fail with 1 mnode as leader =====") + + for k ,v in self.dnode_list.items(): + if k == mnode_name: + if v[3]==0: + tdLog.info("===== depoly cluster mnode only success at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + tdLog.exit("===== depoly cluster mnode only fail at {} , support_vnodes is {} ".format(mnode_name,v[3])) + else: + continue + + def create_db_check_vgroups(self): + + tdSql.execute("drop database if exists test") + tdSql.execute("create database if not exists test replica 1 duration 300") + tdSql.execute("use test") + tdSql.execute( + '''create table stb1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + tags (t1 int) + ''' + ) + tdSql.execute( + ''' + create table t1 + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(16),c9 nchar(32), c10 timestamp) + ''' + ) + + for i in range(5): + tdSql.execute("create table sub_tb_{} using stb1 tags({})".format(i,i)) + tdSql.query("show stables") + tdSql.checkRows(1) + tdSql.query("show tables") + tdSql.checkRows(6) + + tdSql.query("show test.vgroups;") + vgroups_infos = {} # key is id: value is info list + for vgroup_info in tdSql.queryResult: + vgroup_id = vgroup_info[0] + tmp_list = [] + for role in vgroup_info[3:-4]: + if role in ['leader','follower']: + tmp_list.append(role) + vgroups_infos[vgroup_id]=tmp_list + + for k , v in vgroups_infos.items(): + if len(v) ==1 and v[0]=="leader": + tdLog.info(" === create database replica only 1 role leader check success of vgroup_id {} ======".format(k)) + else: + tdLog.exit(" === create database replica only 1 role leader check fail of vgroup_id {} ======".format(k)) + + def create_database(self, dbname, replica_num ,vgroup_nums ): + drop_db_sql = "drop database if exists {}".format(dbname) + create_db_sql = "create database {} replica {} vgroups {}".format(dbname,replica_num,vgroup_nums) + + tdLog.info(" ==== create database {} and insert rows begin =====".format(dbname)) + tdSql.execute(drop_db_sql) + tdSql.execute(create_db_sql) + tdSql.execute("use {}".format(dbname)) + + def create_stable_insert_datas(self,dbname ,stablename , tb_nums , row_nums): + tdSql.execute("use {}".format(dbname)) + tdSql.execute( + '''create table {} + (ts timestamp, c1 int, c2 bigint, c3 smallint, c4 tinyint, c5 float, c6 double, c7 bool, c8 binary(32),c9 nchar(32), c10 timestamp) + tags (t1 int) + '''.format(stablename) + ) + + for i in range(tb_nums): + sub_tbname = "sub_{}_{}".format(stablename,i) + tdSql.execute("create table {} using {} tags({})".format(sub_tbname, stablename ,i)) + # insert datas about new database + + for row_num in range(row_nums): + ts = self.ts + self.ts_step*row_num + tdSql.execute(f"insert into {sub_tbname} values ({ts}, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + + tdLog.info(" ==== stable {} insert rows execute end =====".format(stablename)) + + def append_rows_of_exists_tables(self,dbname ,stablename , tbname , append_nums ): + + tdSql.execute("use {}".format(dbname)) + + for row_num in range(append_nums): + tdSql.execute(f"insert into {tbname} values (now, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + # print(f"insert into {tbname} values (now, {row_num} ,{row_num}, 10 ,1 ,{row_num} ,{row_num},true,'bin_{row_num}','nchar_{row_num}',now) ") + tdLog.info(" ==== append new rows of table {} belongs to stable {} execute end =====".format(tbname,stablename)) + os.system("taos -s 'select count(*) from {}.{}';".format(dbname,stablename)) + + def check_insert_rows(self, dbname, stablename , tb_nums , row_nums, append_rows): + + tdSql.execute("use {}".format(dbname)) + + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) + + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups;'".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select count(*) from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckData("select count(*) from {}.{}".format(dbname,stablename) ,0 , 0 , tb_nums*row_nums+append_rows) + tdLog.info(" ==== check insert rows first failed , this is {}_th retry check rows of database {}".format(count , dbname)) + count += 1 + + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + count = 0 + while not status_OK : + if count > self.try_check_times: + os.system("taos -s ' show {}.vgroups;'".format(dbname)) + tdLog.exit(" ==== check insert rows failed after {} try check {} times of database {}".format(count , self.try_check_times ,dbname)) + break + time.sleep(0.1) + tdSql.query("select distinct tbname from {}.{}".format(dbname,stablename)) + status_OK = self.mycheckRows("select distinct tbname from {}.{}".format(dbname,stablename) ,tb_nums) + tdLog.info(" ==== check insert tbnames first failed , this is {}_th retry check tbnames of database {}".format(count , dbname)) + count += 1 + + def _get_stop_dnode_id(self,dbname): + newTdSql=tdCom.newTdSql() + newTdSql.query("show {}.vgroups".format(dbname)) + vgroup_infos = newTdSql.queryResult + for vgroup_info in vgroup_infos: + leader_infos = vgroup_info[3:-4] + # print(vgroup_info) + for ind ,role in enumerate(leader_infos): + if role =='leader': + # print(ind,leader_infos) + self.stop_dnode_id = leader_infos[ind-1] + break + + + return self.stop_dnode_id + + def wait_stop_dnode_OK(self): + + def _get_status(): + newTdSql=tdCom.newTdSql() + + status = "" + newTdSql.query("show dnodes") + dnode_infos = newTdSql.queryResult + for dnode_info in dnode_infos: + id = dnode_info[0] + dnode_status = dnode_info[4] + if id == self.stop_dnode_id: + status = dnode_status + break + return status + + status = _get_status() + while status !="offline": + time.sleep(0.1) + status = _get_status() + # tdLog.info("==== stop dnode has not been stopped , endpoint is {}".format(self.stop_dnode)) + tdLog.info("==== stop_dnode has stopped , id is {}".format(self.stop_dnode_id)) + + def wait_start_dnode_OK(self): + + def _get_status(): + newTdSql=tdCom.newTdSql() + status = "" + newTdSql.query("show dnodes") + dnode_infos = newTdSql.queryResult + for dnode_info in dnode_infos: + id = dnode_info[0] + dnode_status = dnode_info[4] + if id == self.stop_dnode_id: + status = dnode_status + break + return status + + status = _get_status() + while status !="ready": + time.sleep(0.1) + status = _get_status() + # tdLog.info("==== stop dnode has not been stopped , endpoint is {}".format(self.stop_dnode)) + tdLog.info("==== stop_dnode has restart , id is {}".format(self.stop_dnode_id)) + + def get_leader_infos(self ,dbname): + + newTdSql=tdCom.newTdSql() + newTdSql.query("show {}.vgroups".format(dbname)) + vgroup_infos = newTdSql.queryResult + + leader_infos = set() + for vgroup_info in vgroup_infos: + leader_infos.add(vgroup_info[3:-4]) + + return leader_infos + + def check_revote_leader_success(self, dbname, before_leader_infos , after_leader_infos): + check_status = False + vote_act = set(set(after_leader_infos)-set(before_leader_infos)) + if not vote_act: + print("=======before_revote_leader_infos ======\n" , before_leader_infos) + print("=======after_revote_leader_infos ======\n" , after_leader_infos) + tdLog.exit(" ===maybe revote not occured , there is no dnode offline ====") + else: + for vgroup_info in vote_act: + for ind , role in enumerate(vgroup_info): + if role==self.stop_dnode_id: + + if vgroup_info[ind+1] =="offline" and "leader" in vgroup_info: + tdLog.info(" === revote leader ok , leader is {} now ====".format(list(vgroup_info).index("leader")-1)) + check_status = True + elif vgroup_info[ind+1] !="offline": + tdLog.info(" === dnode {} should be offline ".format(self.stop_dnode_id)) + else: + continue + break + return check_status + + def sync_run_case(self): + # stop follower and insert datas , update tables and create new stables + tdDnodes=cluster.dnodes + for loop in range(self.loop_restart_times): + db_name = "sync_db_{}".format(loop) + stablename = 'stable_{}'.format(loop) + self.create_database(dbname = db_name ,replica_num= self.replica , vgroup_nums= 1) + self.create_stable_insert_datas(dbname = db_name , stablename = stablename , tb_nums= 10 ,row_nums= 10 ) + self.stop_dnode_id = self._get_stop_dnode_id(db_name) + + # check rows of datas + + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + # get leader info before stop + before_leader_infos = self.get_leader_infos(db_name) + + # begin stop dnode + + tdDnodes[self.stop_dnode_id-1].stoptaosd() + + self.wait_stop_dnode_OK() + + # vote leaders check + + # get leader info after stop + after_leader_infos = self.get_leader_infos(db_name) + + revote_status = self.check_revote_leader_success(db_name ,before_leader_infos , after_leader_infos) + + # append rows of stablename when dnode stop make sure revote leaders + + while not revote_status: + after_leader_infos = self.get_leader_infos(db_name) + revote_status = self.check_revote_leader_success(db_name ,before_leader_infos , after_leader_infos) + + + if revote_status: + tbname = "sub_{}_{}".format(stablename , 0) + tdLog.info(" ==== begin append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.append_rows_of_exists_tables(db_name ,stablename , tbname , 100 ) + tdLog.info(" ==== check append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=100) + + # create new stables + tdLog.info(" ==== create new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb1' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb1' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + else: + tdLog.info("===== leader of database {} is not ok , append rows fail =====".format(db_name)) + + # begin start dnode + start = time.time() + tdDnodes[self.stop_dnode_id-1].starttaosd() + self.wait_start_dnode_OK() + end = time.time() + time_cost = int(end -start) + if time_cost > self.max_restart_time: + tdLog.exit(" ==== restart dnode {} cost too much time , please check ====".format(self.stop_dnode_id)) + + # create new stables again + tdLog.info(" ==== create new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb2' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb2' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + def unsync_run_case(self): + + def _restart_dnode_of_db_unsync(dbname): + + tdDnodes=cluster.dnodes + self.stop_dnode_id = self._get_stop_dnode_id(dbname) + # begin restart dnode + + # get leader info before stop + before_leader_infos = self.get_leader_infos(db_name) + + tdDnodes[self.stop_dnode_id-1].stoptaosd() + + # self.wait_stop_dnode_OK() + + # check revote leader when restart servers + # # get leader info after stop + # after_leader_infos = self.get_leader_infos(db_name) + # revote_status = self.check_revote_leader_success(db_name ,before_leader_infos , after_leader_infos) + # # append rows of stablename when dnode stop make sure revote leaders + # while not revote_status: + # after_leader_infos = self.get_leader_infos(db_name) + # revote_status = self.check_revote_leader_success(db_name ,before_leader_infos , after_leader_infos) + tdDnodes[self.stop_dnode_id-1].starttaosd() + start = time.time() + self.wait_start_dnode_OK() + end = time.time() + time_cost = int(end-start) + + if time_cost > self.max_restart_time: + tdLog.exit(" ==== restart dnode {} cost too much time , please check ====".format(self.stop_dnode_id)) + + + def _create_threading(dbname): + self.current_thread = threading.Thread(target=_restart_dnode_of_db_unsync, args=(dbname,)) + return self.current_thread + + + ''' + in this mode , it will be extra threading control start or stop dnode , insert will always going with not care follower online or alive + ''' + for loop in range(self.loop_restart_times): + db_name = "unsync_db_{}".format(loop) + stablename = 'stable_{}'.format(loop) + self.create_database(dbname = db_name ,replica_num= self.replica , vgroup_nums= 1) + self.create_stable_insert_datas(dbname = db_name , stablename = stablename , tb_nums= 10 ,row_nums= 10 ) + + tdLog.info(" ===== restart dnode of database {} in an unsync threading ===== ".format(db_name)) + + # create sync threading and start it + self.current_thread = _create_threading(db_name) + self.current_thread.start() + + # check rows of datas + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + tbname = "sub_{}_{}".format(stablename , 0) + tdLog.info(" ==== begin append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.append_rows_of_exists_tables(db_name ,stablename , tbname , 100 ) + tdLog.info(" ==== check append rows of exists table {} when dnode {} offline ====".format(tbname , self.stop_dnode_id)) + self.check_insert_rows(db_name ,stablename ,tb_nums=10 , row_nums= 10 ,append_rows=100) + + # create new stables + tdLog.info(" ==== create new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb1' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} offline ====".format('new_stb1' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb1' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + # create new stables again + tdLog.info(" ==== create new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.create_stable_insert_datas(dbname = db_name , stablename = 'new_stb2' , tb_nums= 10 ,row_nums= 10 ) + tdLog.info(" ==== check new stable {} when dnode {} restart ====".format('new_stb2' , self.stop_dnode_id)) + self.check_insert_rows(db_name ,'new_stb2' ,tb_nums=10 , row_nums= 10 ,append_rows=0) + + self.current_thread.join() + + + def run(self): + + # basic insert and check of cluster + self.check_setup_cluster_status() + self.create_db_check_vgroups() + # self.sync_run_case() + self.unsync_run_case() + + + + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) \ No newline at end of file -- GitLab