xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #15325 from taosdata/test/cluster_vnode_case 

test : cluster test case about vnode
This commit is contained in:
Hui Li 2022-07-23 12:45:49 +08:00 committed by GitHub
parent 3543e98891 950cf5dc35
commit c57a04a6c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 3720 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/pytest/util/common.py
View File

@ -457,14 +457,14 @@ class TDCom:
def newcon(self,host='localhost',port=6030,user='root',password='taosdata'):
con=taos.connect(host=host, user=user, password=password, port=port)
print(con)
# print(con)
return con
def newcur(self,host='localhost',port=6030,user='root',password='taosdata'):
cfgPath = self.getClientCfgPath()
con=taos.connect(host=host, user=user, password=password, config=cfgPath, port=port)
cur=con.cursor()
print(cur)
# print(cur)
return cur
def newTdSql(self, host='localhost',port=6030,user='root',password='taosdata'):

138
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_createDb_replica1.py Normal file
View File

@ -0,0 +1,138 @@
# 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 = {}
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 getConnection(self, dnode):
host = dnode.cfgDict["fqdn"]
port = dnode.cfgDict["serverPort"]
config_dir = dnode.cfgDir
return taos.connect(host=host, port=int(port), config=config_dir)
def run(self):
self.check_setup_cluster_status()
self.create_db_check_vgroups()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

179
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica1_insertdatas.py Normal file
View File

@ -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 = 1
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_1_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_1_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())

179
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas.py Normal file
View File

@ -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())

496
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_sync.py Normal file
View File

@ -0,0 +1,496 @@
# 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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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())

497
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync.py Normal file
View File

@ -0,0 +1,497 @@
# 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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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())

521
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_follower_unsync_force_stop.py Normal file
View File

@ -0,0 +1,521 @@
# 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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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].forcestop()
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)
while not self.stop_dnode_id:
time.sleep(0.5)
self.stop_dnode_id = self._get_stop_dnode_id(dbname)
# begin restart dnode
# force stop taosd by kill -9
self.force_stop_dnode(self.stop_dnode_id)
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 force_stop_dnode(self, dnode_id ):
tdSql.query("show dnodes")
port = None
for dnode_info in tdSql.queryResult:
if dnode_id == dnode_info[0]:
port = dnode_info[1].split(":")[-1]
break
else:
continue
if port:
tdLog.info(" ==== dnode {} will be force stop by kill -9 ====".format(dnode_id))
psCmd = '''netstat -anp|grep -w LISTEN|grep -w %s |grep -o "LISTEN.*"|awk '{print $2}'|cut -d/ -f1|head -n1''' %(port)
processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
ps_kill_taosd = ''' kill -9 {} '''.format(processID)
# print(ps_kill_taosd)
os.system(ps_kill_taosd)
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())

558
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader.py Normal file
View File

@ -0,0 +1,558 @@
# 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 = 10
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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(vgroup_info[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
tdDnodes[self.stop_dnode_id-1].stoptaosd()
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)
# # get leader info before stop
# before_leader_infos = self.get_leader_infos(db_name)
# 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)
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())

580
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_stop_leader_forece_stop.py Normal file
View File

@ -0,0 +1,580 @@
# 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 = 10
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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))
while not tdSql.queryResult:
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)
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))
while not tdSql.queryResult:
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(vgroup_info[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 force_stop_dnode(self, dnode_id ):
tdSql.query("show dnodes")
port = None
for dnode_info in tdSql.queryResult:
if dnode_id == dnode_info[0]:
port = dnode_info[1].split(":")[-1]
break
else:
continue
if port:
tdLog.info(" ==== dnode {} will be force stop by kill -9 ====".format(dnode_id))
psCmd = '''netstat -anp|grep -w LISTEN|grep -w %s |grep -o "LISTEN.*"|awk '{print $2}'|cut -d/ -f1|head -n1''' %(port)
processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
ps_kill_taosd = ''' kill -9 {} '''.format(processID)
# print(ps_kill_taosd)
os.system(ps_kill_taosd)
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
# force stop taosd by kill -9
self.force_stop_dnode(self.stop_dnode_id)
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
# force stop taosd by kill -9
# get leader info before stop
before_leader_infos = self.get_leader_infos(db_name)
self.force_stop_dnode(self.stop_dnode_id)
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)
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)
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)
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())

206
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_vgroups.py Normal file
View File

@ -0,0 +1,206 @@
# 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
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 = 1
self.vgroups = 2
self.tb_nums = 10
self.row_nums = 100
self.max_vote_time_cost = 10 # seconds
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 check_vgroups_init_done(self,dbname):
status = True
tdSql.query("show {}.vgroups".format(dbname))
for vgroup_info in tdSql.queryResult:
vgroup_id = vgroup_info[0]
vgroup_status = []
for ind , role in enumerate(vgroup_info[3:-4]):
if ind%2==0:
continue
else:
vgroup_status.append(role)
if vgroup_status.count("leader")!=1 or vgroup_status.count("follower")!=2:
status = False
return status
return status
def vote_leader_time_costs(self,dbname):
start = time.time()
status = self.check_vgroups_init_done(dbname)
while not status:
time.sleep(0.1)
status = self.check_vgroups_init_done(dbname)
# tdLog.info("=== database {} show vgroups vote the leader is in progress ===".format(dbname))
end = time.time()
cost_time = end - start
tdLog.info(" ==== database %s vote the leaders success , cost time is %.3f second ====="%(dbname,cost_time) )
# os.system("taos -s 'show {}.vgroups;'".format(dbname))
if cost_time >= self.max_vote_time_cost:
tdLog.exit(" ==== database %s vote the leaders cost too large time , cost time is %.3f second ===="%(dbname,cost_time) )
return cost_time
def test_init_vgroups_time_costs(self):
tdLog.info(" ====start check time cost about vgroups vote leaders ==== ")
tdLog.info(" ==== current max time cost is set value : {} =======".format(self.max_vote_time_cost))
# create database replica 3 vgroups 1
db1 = 'db_1'
create_db_replica_3_vgroups_1 = "create database {} replica 3 vgroups 1".format(db1)
tdLog.info('=======database {} replica 3 vgroups 1 ======'.format(db1))
tdSql.execute(create_db_replica_3_vgroups_1)
self.vote_leader_time_costs(db1)
# create database replica 3 vgroups 10
db2 = 'db_2'
create_db_replica_3_vgroups_10 = "create database {} replica 3 vgroups 10".format(db2)
tdLog.info('=======database {} replica 3 vgroups 10 ======'.format(db2))
tdSql.execute(create_db_replica_3_vgroups_10)
self.vote_leader_time_costs(db2)
# create database replica 3 vgroups 100
db3 = 'db_3'
create_db_replica_3_vgroups_100 = "create database {} replica 3 vgroups 100".format(db3)
tdLog.info('=======database {} replica 3 vgroups 100 ======'.format(db3))
tdSql.execute(create_db_replica_3_vgroups_100)
self.vote_leader_time_costs(db3)
def run(self):
self.check_setup_cluster_status()
self.test_init_vgroups_time_costs()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

364
tests/system-test/6-cluster/vnode/4dnode1mnode_basic_replica3_vgroups_stopOne.py Normal file
View File

@ -0,0 +1,364 @@
# author : wenzhouwww
from errno import ESOCKTNOSUPPORT
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 random
import socket
import subprocess
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 = 1
self.vgroups = 2
self.tb_nums = 10
self.row_nums = 100
self.max_vote_time_cost = 10 # seconds
self.stop_dnode = None
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 _get_stop_dnode(self):
only_dnode_list = self.dnode_list.keys() - self.mnode_list.keys()
self.stop_dnode = random.sample(only_dnode_list , 1 )[0]
return self.stop_dnode
def check_vgroups_revote_leader(self,dbname):
status = True
stop_dnode_id = self.dnode_list[self.stop_dnode][0]
tdSql.query("show {}.vgroups".format(dbname))
for vgroup_info in tdSql.queryResult:
vgroup_id = vgroup_info[0]
vgroup_status = []
vgroups_leader_follower = vgroup_info[3:-4]
for ind , role in enumerate(vgroups_leader_follower):
if ind%2==0:
if role == stop_dnode_id and vgroups_leader_follower[ind+1]=="offline":
tdLog.info("====== dnode {} has offline , endpoint is {}".format(stop_dnode_id , self.stop_dnode))
elif role == stop_dnode_id :
tdLog.exit("====== dnode {} has not offline , endpoint is {}".format(stop_dnode_id , self.stop_dnode))
else:
continue
else:
vgroup_status.append(role)
if vgroup_status.count("leader")!=1 or vgroup_status.count("follower")!=1 or vgroup_status.count("offline")!=1:
status = False
return status
return status
def wait_stop_dnode_OK(self):
def _get_status():
status = ""
tdSql.query("show dnodes")
dnode_infos = tdSql.queryResult
for dnode_info in dnode_infos:
endpoint = dnode_info[1]
dnode_status = dnode_info[4]
if endpoint == self.stop_dnode:
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 , endpoint is {}".format(self.stop_dnode))
def wait_start_dnode_OK(self):
def _get_status():
status = ""
tdSql.query("show dnodes")
dnode_infos = tdSql.queryResult
for dnode_info in dnode_infos:
endpoint = dnode_info[1]
dnode_status = dnode_info[4]
if endpoint == self.stop_dnode:
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 , endpoint is {}".format(self.stop_dnode))
def random_stop_One_dnode(self):
self.stop_dnode = self._get_stop_dnode()
stop_dnode_id = self.dnode_list[self.stop_dnode][0]
tdLog.info(" ==== dnode {} will offline ,endpoints is {} ====".format(stop_dnode_id , self.stop_dnode))
tdDnodes=cluster.dnodes
tdDnodes[stop_dnode_id-1].stoptaosd()
self.wait_stop_dnode_OK()
# os.system("taos -s 'show dnodes;'")
def Restart_stop_dnode(self):
tdDnodes=cluster.dnodes
stop_dnode_id = self.dnode_list[self.stop_dnode][0]
tdDnodes[stop_dnode_id-1].starttaosd()
self.wait_start_dnode_OK()
# os.system("taos -s 'show dnodes;'")
def check_vgroups_init_done(self,dbname):
status = True
tdSql.query("show {}.vgroups".format(dbname))
for vgroup_info in tdSql.queryResult:
vgroup_id = vgroup_info[0]
vgroup_status = []
for ind , role in enumerate(vgroup_info[3:-4]):
if ind%2==0:
continue
else:
vgroup_status.append(role)
if vgroup_status.count("leader")!=1 or vgroup_status.count("follower")!=2:
status = False
return status
return status
def vote_leader_time_costs(self,dbname):
start = time.time()
status = self.check_vgroups_init_done(dbname)
while not status:
time.sleep(0.1)
status = self.check_vgroups_init_done(dbname)
# tdLog.info("=== database {} show vgroups vote the leader is in progress ===".format(dbname))
end = time.time()
cost_time = end - start
tdLog.info(" ==== database %s vote the leaders success , cost time is %.3f second ====="%(dbname,cost_time) )
# os.system("taos -s 'show {}.vgroups;'".format(dbname))
if cost_time >= self.max_vote_time_cost:
tdLog.exit(" ==== database %s vote the leaders cost too large time , cost time is %.3f second ===="%(dbname,cost_time) )
return cost_time
def revote_leader_time_costs(self,dbname):
start = time.time()
status = self.check_vgroups_revote_leader(dbname)
while not status:
time.sleep(0.1)
status = self.check_vgroups_revote_leader(dbname)
# tdLog.info("=== database {} show vgroups vote the leader is in progress ===".format(dbname))
end = time.time()
cost_time = end - start
tdLog.info(" ==== database %s revote the leaders success , cost time is %.3f second ====="%(dbname,cost_time) )
# os.system("taos -s 'show {}.vgroups;'".format(dbname))
if cost_time >= self.max_vote_time_cost:
tdLog.exit(" ==== database %s revote the leaders cost too large time , cost time is %.3f second ===="%(dbname,cost_time) )
return cost_time
def exec_revote_action(self,dbname):
tdSql.query("show {}.vgroups".format(dbname))
before_revote = tdSql.queryResult
before_vgroups = set()
for vgroup_info in before_revote:
before_vgroups.add(vgroup_info[3:-4])
self.random_stop_One_dnode()
tdSql.query("show {}.vgroups".format(dbname))
after_revote = tdSql.queryResult
after_vgroups = set()
for vgroup_info in after_revote:
after_vgroups.add(vgroup_info[3:-4])
vote_act = set(set(after_vgroups)-set(before_vgroups))
if not vote_act:
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.dnode_list[self.stop_dnode][0]:
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))
elif vgroup_info[ind+1] !="offline":
tdLog.exit(" === dnode {} should be offline ".format(self.stop_dnode))
else:
continue
break
self.revote_leader_time_costs(dbname)
self.Restart_stop_dnode()
def test_init_vgroups_time_costs(self):
tdLog.info(" ====start check time cost about vgroups vote leaders ==== ")
tdLog.info(" ==== current max time cost is set value : {} =======".format(self.max_vote_time_cost))
# create database replica 3 vgroups 1
db1 = 'db_1'
create_db_replica_3_vgroups_1 = "create database {} replica 3 vgroups 1".format(db1)
tdLog.info('=======database {} replica 3 vgroups 1 ======'.format(db1))
tdSql.execute(create_db_replica_3_vgroups_1)
self.vote_leader_time_costs(db1)
self.exec_revote_action(db1)
# create database replica 3 vgroups 10
db2 = 'db_2'
create_db_replica_3_vgroups_10 = "create database {} replica 3 vgroups 10".format(db2)
tdLog.info('=======database {} replica 3 vgroups 10 ======'.format(db2))
tdSql.execute(create_db_replica_3_vgroups_10)
self.vote_leader_time_costs(db2)
self.exec_revote_action(db2)
# create database replica 3 vgroups 100
db3 = 'db_3'
create_db_replica_3_vgroups_100 = "create database {} replica 3 vgroups 100".format(db3)
tdLog.info('=======database {} replica 3 vgroups 100 ======'.format(db3))
tdSql.execute(create_db_replica_3_vgroups_100)
self.vote_leader_time_costs(db3)
self.exec_revote_action(db3)
def run(self):
self.check_setup_cluster_status()
self.test_init_vgroups_time_costs()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())