tmqDnodeRestart.py 12.6 KB
Newer Older
P
plum-lihui 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

import taos
import sys
import time
import socket
import os
import threading
from enum import Enum

from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
sys.path.append("./7-tmq")
from tmqCommon import *

class TDTestCase:
    def __init__(self):
P
plum-lihui 已提交
19
        self.snapshot   = 0
P
plum-lihui 已提交
20
        self.vgroups    = 2
P
plum-lihui 已提交
21
        self.ctbNum     = 100
P
plum-lihui 已提交
22
        self.rowsPerTbl = 1000
P
plum-lihui 已提交
23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40
        
    def init(self, conn, logSql):
        tdLog.debug(f"start to excute {__file__}")
        tdSql.init(conn.cursor(), False)

    def prepareTestEnv(self):
        tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ")
        paraDict = {'dbName':     'dbt',
                    'dropFlag':   1,
                    'event':      '',
                    'vgroups':    3,
                    'stbName':    'stb',
                    'colPrefix':  'c',
                    'tagPrefix':  't',
                    'colSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}],
                    'tagSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
                    'ctbPrefix':  'ctb',
                    'ctbStartIdx': 0,
P
plum-lihui 已提交
41
                    'ctbNum':     100,
P
plum-lihui 已提交
42
                    'rowsPerTbl': 1000,
P
plum-lihui 已提交
43
                    'batchNum':   10,
P
plum-lihui 已提交
44 45 46 47 48 49
                    'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
                    'pollDelay':  3,
                    'showMsg':    1,
                    'showRow':    1,
                    'snapshot':   0}

P
plum-lihui 已提交
50
        paraDict['snapshot'] = self.snapshot
P
plum-lihui 已提交
51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
        paraDict['vgroups'] = self.vgroups
        paraDict['ctbNum'] = self.ctbNum
        paraDict['rowsPerTbl'] = self.rowsPerTbl
        
        tmqCom.initConsumerTable()
        tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
        tdLog.info("create stb")
        tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
        tdLog.info("create ctb")
        tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
                             ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
        tdLog.info("insert data")
        tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
                                               ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
                                               startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
        
        tdLog.info("restart taosd to ensure that the data falls into the disk")
        # tdDnodes.stop(1)
        # tdDnodes.start(1)
        tdSql.query("flush database %s"%(paraDict['dbName']))
        return

    def tmqCase1(self):
        tdLog.printNoPrefix("======== test case 1: ")
        paraDict = {'dbName':     'dbt',
                    'dropFlag':   1,
                    'event':      '',
                    'vgroups':    4,
                    'stbName':    'stb',
                    'colPrefix':  'c',
                    'tagPrefix':  't',
                    'colSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}],
                    'tagSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
                    'ctbPrefix':  'ctb',
                    'ctbStartIdx': 0,
P
plum-lihui 已提交
86 87
                    'ctbNum':     1000,
                    'rowsPerTbl': 1000,
P
plum-lihui 已提交
88
                    'batchNum':   100,
P
plum-lihui 已提交
89 90 91 92
                    'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
                    'pollDelay':  5,
                    'showMsg':    1,
                    'showRow':    1,
P
plum-lihui 已提交
93
                    'snapshot':   0}
P
plum-lihui 已提交
94

P
plum-lihui 已提交
95 96 97 98
        paraDict['snapshot'] = self.snapshot
        paraDict['vgroups'] = self.vgroups
        paraDict['ctbNum'] = self.ctbNum
        paraDict['rowsPerTbl'] = self.rowsPerTbl
P
plum-lihui 已提交
99
        
P
plum-lihui 已提交
100 101 102 103 104 105 106 107 108 109 110
        # tmqCom.initConsumerTable()
        # tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
        # tdLog.info("create stb")
        # tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
        # tdLog.info("create ctb")
        # tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
        #                      ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
        # tdLog.info("insert data")
        # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
        #                                        ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
        #                                        startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
P
plum-lihui 已提交
111 112 113 114 115 116 117 118 119

        tdLog.info("create topics from stb1")
        topicFromStb1 = 'topic_stb1'                
        queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
        tdLog.info("create topic sql: %s"%sqlString)
        tdSql.execute(sqlString)
        
        consumerId     = 0
P
plum-lihui 已提交
120
        expectrowcnt   = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2
P
plum-lihui 已提交
121 122 123 124 125
        topicList      = topicFromStb1
        ifcheckdata    = 0
        ifManualCommit = 0
        keyList        = 'group.id:cgrp1,\
                        enable.auto.commit:true,\
P
plum-lihui 已提交
126
                        auto.commit.interval.ms:3000,\
P
plum-lihui 已提交
127 128 129 130 131 132 133
                        auto.offset.reset:earliest'
        tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)

        tdLog.info("start consume processor")
        tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
        
        # time.sleep(3)
P
plum-lihui 已提交
134
        tmqCom.getStartConsumeNotifyFromTmqsim()
P
plum-lihui 已提交
135
        tdLog.info("================= restart dnode ===========================")
P
plum-lihui 已提交
136 137
        tdDnodes.stoptaosd(1)
        tdDnodes.starttaosd(1)
P
plum-lihui 已提交
138
        # time.sleep(3)
P
plum-lihui 已提交
139

P
plum-lihui 已提交
140
        tdLog.info(" restart taosd end and wait to check consume result")
P
plum-lihui 已提交
141 142 143 144 145 146 147
        expectRows = 1
        resultList = tmqCom.selectConsumeResult(expectRows)
        totalConsumeRows = 0
        for i in range(expectRows):
            totalConsumeRows += resultList[i]

        tdSql.query(queryString)
P
plum-lihui 已提交
148
        totalRowsFromQury = tdSql.getRows()
P
plum-lihui 已提交
149
        
P
plum-lihui 已提交
150 151
        tdLog.info("act consume rows: %d, act query rows: %d"%(totalConsumeRows, totalRowsFromQury))
        if not (totalConsumeRows == totalRowsFromQury):
P
plum-lihui 已提交
152 153
            tdLog.exit("tmq consume rows error!")

P
plum-lihui 已提交
154
        tmqCom.waitSubscriptionExit(tdSql, topicFromStb1)
P
plum-lihui 已提交
155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171
        tdSql.query("drop topic %s"%topicFromStb1)

        tdLog.printNoPrefix("======== test case 1 end ...... ")

    def tmqCase2(self):
        tdLog.printNoPrefix("======== test case 2: ")  
        paraDict = {'dbName':     'dbt',
                    'dropFlag':   1,
                    'event':      '',
                    'vgroups':    4,
                    'stbName':    'stb',
                    'colPrefix':  'c',
                    'tagPrefix':  't',
                    'colSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}],
                    'tagSchema':   [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
                    'ctbPrefix':  'ctb',
                    'ctbStartIdx': 0,
P
plum-lihui 已提交
172
                    'ctbNum':     100,
P
plum-lihui 已提交
173
                    'rowsPerTbl': 1000,
P
plum-lihui 已提交
174
                    'batchNum':   100,
P
plum-lihui 已提交
175
                    'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
P
plum-lihui 已提交
176
                    'pollDelay':  20,
P
plum-lihui 已提交
177 178
                    'showMsg':    1,
                    'showRow':    1,
P
plum-lihui 已提交
179
                    'snapshot':   0}
P
plum-lihui 已提交
180

P
plum-lihui 已提交
181 182 183 184
        paraDict['snapshot'] = self.snapshot
        paraDict['vgroups'] = self.vgroups
        paraDict['ctbNum'] = self.ctbNum
        paraDict['rowsPerTbl'] = self.rowsPerTbl
P
plum-lihui 已提交
185 186
        
        tmqCom.initConsumerTable()
P
plum-lihui 已提交
187 188 189 190 191 192 193 194 195 196
        # tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
        # tdLog.info("create stb")
        # tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
        # tdLog.info("create ctb")
        # tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
        #                      ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
        # tdLog.info("insert data")
        # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
        #                                        ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
        #                                        startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
P
plum-lihui 已提交
197 198 199 200 201 202 203
        tdLog.info("create topics from stb1")
        topicFromStb1 = 'topic_stb1'                
        queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
        sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
        tdLog.info("create topic sql: %s"%sqlString)
        tdSql.execute(sqlString)
        
P
plum-lihui 已提交
204
        consumerId     = 1
P
plum-lihui 已提交
205
        expectrowcnt   = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2
P
plum-lihui 已提交
206 207 208 209 210
        topicList      = topicFromStb1
        ifcheckdata    = 0
        ifManualCommit = 0
        keyList        = 'group.id:cgrp1,\
                        enable.auto.commit:true,\
P
plum-lihui 已提交
211
                        auto.commit.interval.ms:3000,\
P
plum-lihui 已提交
212 213 214 215 216 217
                        auto.offset.reset:earliest'
        tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)

        tdLog.info("start consume processor")
        tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])

P
plum-lihui 已提交
218
        tmqCom.getStartConsumeNotifyFromTmqsim()
P
plum-lihui 已提交
219
        tdLog.info("================= restart dnode ===========================")
P
plum-lihui 已提交
220 221
        tdDnodes.stoptaosd(1)
        tdDnodes.starttaosd(1)
P
plum-lihui 已提交
222
        # time.sleep(3)
P
plum-lihui 已提交
223

P
plum-lihui 已提交
224 225 226
        tdLog.info("create some new child table and insert data ")
        paraDict["batchNum"] = 100
        paraDict["ctbPrefix"] = 'newCtb'
P
plum-lihui 已提交
227
        tmqCom.insert_data_with_autoCreateTbl(tdSql,paraDict["dbName"],paraDict["stbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"])
P
plum-lihui 已提交
228
        
P
plum-lihui 已提交
229 230 231 232 233 234 235 236
        tdLog.info("insert process end, and start to check consume result")
        expectRows = 1
        resultList = tmqCom.selectConsumeResult(expectRows)
        totalConsumeRows = 0
        for i in range(expectRows):
            totalConsumeRows += resultList[i]

        tdSql.query(queryString)
P
plum-lihui 已提交
237
        totalRowsFromQuery = tdSql.getRows()
P
plum-lihui 已提交
238
        
P
plum-lihui 已提交
239 240
        tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsFromQuery))
        if totalConsumeRows != totalRowsFromQuery:
P
plum-lihui 已提交
241 242
            tdLog.exit("tmq consume rows error!")

P
plum-lihui 已提交
243
        tmqCom.waitSubscriptionExit(tdSql, topicFromStb1)
P
plum-lihui 已提交
244 245 246 247 248
        tdSql.query("drop topic %s"%topicFromStb1)

        tdLog.printNoPrefix("======== test case 2 end ...... ")

    def run(self):
P
plum-lihui 已提交
249
        # tdSql.prepare()
P
plum-lihui 已提交
250
        self.prepareTestEnv()
P
plum-lihui 已提交
251
        self.tmqCase1()
P
plum-lihui 已提交
252 253 254 255 256 257 258 259 260 261
        self.tmqCase2() 

    def stop(self):
        tdSql.close()
        tdLog.success(f"{__file__} successfully executed")

event = threading.Event()

tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())