Merge pull request #18535 from taosdata/enh/addTraceInfo

enh: add trace info

include/os/osSocket.h

@@ -169,6 +169,9 @@ void        taosSetMaskSIGPIPE();
 uint32_t    taosInetAddr(const char *ipAddr);
 const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len);
 
+uint64_t taosHton64(uint64_t val);
+uint64_t taosNtoh64(uint64_t val);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/transport/inc/transComm.h

@@ -169,6 +169,7 @@ typedef struct {
   char spi : 2;
   char hasEpSet : 2;  // contain epset or not, 0(default): no epset, 1: contain epset
 
+  uint64_t timestamp;
   char     user[TSDB_UNI_LEN];
   uint32_t magicNum;
   STraceId traceId;

source/libs/transport/src/transCli.c

@@ -758,6 +758,14 @@ static void cliSendCb(uv_write_t* req, int status) {
   SCliConn* pConn = transReqQueueRemove(req);
   if (pConn == NULL) return;
 
+  SCliMsg* pMsg = !transQueueEmpty(&pConn->cliMsgs) ? transQueueGet(&pConn->cliMsgs, 0) : NULL;
+  if (pMsg != NULL) {
+    int64_t cost = taosGetTimestampUs() - pMsg->st;
+    if (cost > 1000) {
+      tWarn("%s conn %p send cost:%dus, send exception", CONN_GET_INST_LABEL(pConn), pConn, (int)cost);
+    }
+  }
+
   if (status == 0) {
     tTrace("%s conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn);
   } else {
@@ -806,6 +814,7 @@ void cliSend(SCliConn* pConn) {
     pHead->traceId = pMsg->info.traceId;
     pHead->magicNum = htonl(TRANS_MAGIC_NUM);
   }
+  pHead->timestamp = taosHton64(taosGetTimestampUs());
 
   if (pHead->persist == 1) {
     CONN_SET_PERSIST_BY_APP(pConn);
@@ -1662,6 +1671,7 @@ int transReleaseCliHandle(void* handle) {
 
   SCliMsg* cmsg = taosMemoryCalloc(1, sizeof(SCliMsg));
   cmsg->msg = tmsg;
+  cmsg->st = taosGetTimestampUs();
   cmsg->type = Release;
   cmsg->ctx = pCtx;
 

source/libs/transport/src/transSvr.c

@@ -231,14 +231,29 @@ static bool uvHandleReq(SSvrConn* pConn) {
     }
   }
   STraceId* trace = &pHead->traceId;
+
+  int64_t        cost = taosGetTimestampUs() - taosNtoh64(pHead->timestamp);
+  static int64_t EXCEPTION_LIMIT_US = 100 * 1000;
+
   if (pConn->status == ConnNormal && pHead->noResp == 0) {
     transRefSrvHandle(pConn);
-
-    tGDebug("%s conn %p %s received from %s, local info:%s, len:%d", transLabel(pTransInst), pConn,
-            TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen);
+    if (cost >= EXCEPTION_LIMIT_US) {
+      tGWarn("%s conn %p %s received from %s, local info:%s, len:%d, cost:%dus, recv exception", transLabel(pTransInst),
+             pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, (int)cost);
+    } else {
+      tGDebug("%s conn %p %s received from %s, local info:%s, len:%d, cost:%dus", transLabel(pTransInst), pConn,
+              TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, (int)cost);
+    }
   } else {
-    tGDebug("%s conn %p %s received from %s, local info:%s, len:%d, resp:%d, code:%d", transLabel(pTransInst), pConn,
-            TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, pHead->noResp, transMsg.code);
+    if (cost >= EXCEPTION_LIMIT_US) {
+      tGWarn("%s conn %p %s received from %s, local info:%s, len:%d, resp:%d, code:%d, cost:%dus, recv exception",
+             transLabel(pTransInst), pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, pHead->noResp,
+             transMsg.code, (int)(cost));
+    } else {
+      tGWarn("%s conn %p %s received from %s, local info:%s, len:%d, resp:%d, code:%d, cost:%dus",
+             transLabel(pTransInst), pConn, TMSG_INFO(transMsg.msgType), pConn->dst, pConn->src, msgLen, pHead->noResp,
+             transMsg.code, (int)(cost));
+    }
   }
 
   // pHead->noResp = 1,

source/os/src/osSocket.c

@@ -1103,3 +1103,30 @@ void taosWinSocketInit() {
 #else
 #endif
 }
+
+uint64_t taosHton64(uint64_t val) {
+#if defined(WINDOWS) || defined(DARWIN)
+  return ((val & 0x00000000000000ff) << 7 * 8) | ((val & 0x000000000000ff00) << 5 * 8) |
+         ((val & 0x0000000000ff0000) << 3 * 8) | ((val & 0x00000000ff000000) << 1 * 8) |
+         ((val & 0x000000ff00000000) >> 1 * 8) | ((val & 0x0000ff0000000000) >> 3 * 8) |
+         ((val & 0x00ff000000000000) >> 5 * 8) | ((val & 0xff00000000000000) >> 7 * 8);
+#else
+  if (__BYTE_ORDER == __LITTLE_ENDIAN) {
+    return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) | (unsigned int)htonl((int)(val >> 32));
+  } else if (__BYTE_ORDER == __BIG_ENDIAN) {
+    return val;
+  }
+#endif
+}
+
+uint64_t taosNtoh64(uint64_t val) {
+#if defined(WINDOWS) || defined(DARWIN)
+  return taosHton64(val);
+#else
+  if (__BYTE_ORDER == __LITTLE_ENDIAN) {
+    return (((uint64_t)htonl((int)((val << 32) >> 32))) << 32) | (unsigned int)htonl((int)(val >> 32));
+  } else if (__BYTE_ORDER == __BIG_ENDIAN) {
+    return val;
+  }
+#endif
+}

tests/system-test/0-others/compatibility.py

@@ -15,155 +15,159 @@ from util.cluster import *
 
 
 class TDTestCase:
-    def caseDescription(self):
-        '''
-        3.0 data compatibility test 
-        case1: basedata version is 3.0.1.0
-        '''
-        return
-
-    def init(self, conn, logSql, replicaVar=1):
-        self.replicaVar = int(replicaVar)
-        tdLog.debug(f"start to excute {__file__}")
-        tdSql.init(conn.cursor())
-
-
-    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 or "taosd.exe" 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 getCfgPath(self):
-        buildPath = self.getBuildPath()
-        selfPath = os.path.dirname(os.path.realpath(__file__))
-
-        if ("community" in selfPath):
-            cfgPath = buildPath + "/../sim/dnode1/cfg/"
-        else:
-            cfgPath = buildPath + "/../sim/dnode1/cfg/"
-
-        return cfgPath
-
-    def installTaosd(self,bPath,cPath):
-        # os.system(f"rmtaos && mkdir -p {self.getBuildPath()}/build/lib/temp &&  mv {self.getBuildPath()}/build/lib/libtaos.so*  {self.getBuildPath()}/build/lib/temp/ ")
-        # os.system(f" mv {bPath}/build  {bPath}/build_bak ")
-        # os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so  {self.getBuildPath()}/build/lib/libtaos.so_bak ")
-        # os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so.1  {self.getBuildPath()}/build/lib/libtaos.so.1_bak ")
-
-        packagePath="/usr/local/src/"
-        packageName="TDengine-server-3.0.1.0-Linux-x64.tar.gz"
-        os.system(f"cd {packagePath} &&  tar xvf  TDengine-server-3.0.1.0-Linux-x64.tar.gz && cd TDengine-server-3.0.1.0 && ./install.sh  -e no  " )
-        tdDnodes.stop(1)
-        print(f"start taosd: nohup taosd -c {cPath} & ")
-        os.system(f" nohup taosd -c {cPath} & " )
-        sleep(1)
-
-
-        
-    def buildTaosd(self,bPath):
-        # os.system(f"mv {bPath}/build_bak  {bPath}/build ")
-        os.system(f" cd {bPath}  &&  make install ")
-
-
-    def run(self):
-        bPath=self.getBuildPath()
-        cPath=self.getCfgPath()
-        dbname = "test"
-        stb = f"{dbname}.meters"
-        self.installTaosd(bPath,cPath)
-        os.system("echo 'debugFlag 143' > /etc/taos/taos.cfg ")
-        tableNumbers=100
-        recordNumbers1=100
-        recordNumbers2=1000
-        tdsqlF=tdCom.newTdSql()
-        print(tdsqlF)
-        tdsqlF.query(f"SELECT SERVER_VERSION();")
-        print(tdsqlF.query(f"SELECT SERVER_VERSION();"))
-        oldServerVersion=tdsqlF.queryResult[0][0]
-        tdLog.info(f"Base server version is {oldServerVersion}")
-        tdsqlF.query(f"SELECT CLIENT_VERSION();")
-        
-        # the oldClientVersion can't be updated in the same python process,so the version is new compiled verison
-        oldClientVersion=tdsqlF.queryResult[0][0]
-        tdLog.info(f"Base client version is {oldClientVersion}")
-
-        tdLog.printNoPrefix(f"==========step1:prepare and check data in old version-{oldServerVersion}")
-        tdLog.info(f" LD_LIBRARY_PATH=/usr/lib  taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
-        os.system(f"LD_LIBRARY_PATH=/usr/lib taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
-        sleep(3)
-
-        # tdsqlF.query(f"select count(*) from {stb}")
-        # tdsqlF.checkData(0,0,tableNumbers*recordNumbers1)
-        os.system("pkill taosd")
-        sleep(2)
-
-        print(f"start taosd: nohup taosd -c {cPath} & ")
-        os.system(f" nohup taosd -c {cPath} & " )
-        sleep(10)
-        tdLog.info(" LD_LIBRARY_PATH=/usr/lib  taosBenchmark -f 0-others/compa4096.json -y  ")
-        os.system("LD_LIBRARY_PATH=/usr/lib  taosBenchmark -f 0-others/compa4096.json -y")
-        os.system("pkill -9 taosd")
-
-
-
-        tdLog.printNoPrefix("==========step2:update new version ")
-        self.buildTaosd(bPath)
-        tdDnodes.start(1)
-        sleep(1)
-        tdsql=tdCom.newTdSql()
-        print(tdsql)
-
-
-        tdsql.query(f"SELECT SERVER_VERSION();")
-        nowServerVersion=tdsql.queryResult[0][0]
-        tdLog.info(f"New server version is {nowServerVersion}")
-        tdsql.query(f"SELECT CLIENT_VERSION();")
-        nowClientVersion=tdsql.queryResult[0][0]
-        tdLog.info(f"New client version is {nowClientVersion}")
-
-        tdLog.printNoPrefix(f"==========step3:prepare and check data in new version-{nowServerVersion}")
-        tdsql.query(f"select count(*) from {stb}")
-        tdsql.checkData(0,0,tableNumbers*recordNumbers1)        
-        os.system(f"taosBenchmark -t {tableNumbers} -n {recordNumbers2} -y  ")
-        tdsql.query(f"select count(*) from {stb}")
-        tdsql.checkData(0,0,tableNumbers*recordNumbers2)
-
-        tdsql=tdCom.newTdSql()
-        tdLog.printNoPrefix(f"==========step4:verify backticks in taos Sql-TD18542")
-        tdsql.execute("drop database if exists db")
-        tdsql.execute("create database db")
-        tdsql.execute("use db")
-        tdsql.execute("create stable db.stb1 (ts timestamp, c1 int) tags (t1 int);")
-        tdsql.execute("insert into db.ct1 using db.stb1 TAGS(1) values(now(),11);")
-        tdsql.error(" insert into `db.ct2` using db.stb1 TAGS(9) values(now(),11);")
-        tdsql.error(" insert into db.`db.ct2` using db.stb1 TAGS(9) values(now(),11);")
-        tdsql.execute("insert into `db`.ct3 using db.stb1 TAGS(3) values(now(),13);")
-        tdsql.query("select * from db.ct3")
-        tdsql.checkData(0,1,13)
-        tdsql.execute("insert into db.`ct4` using db.stb1 TAGS(4) values(now(),14);")
-        tdsql.query("select * from db.ct4")
-        tdsql.checkData(0,1,14)
-        tdsql.query("describe  information_schema.ins_databases;")
-        qRows=tdsql.queryRows   
-        for i in  range(qRows) :
-            if tdsql.queryResult[i][0]=="retentions" :
-                return True
-            else:
-                return False
-    def stop(self):
-        tdSql.close()
-        tdLog.success(f"{__file__} successfully executed")
+	def caseDescription(self):
+		'''
+		3.0 data compatibility test 
+		case1: basedata version is 3.0.1.0
+		'''
+		return
+
+	def init(self, conn, logSql, replicaVar=1):
+		self.replicaVar = int(replicaVar)
+		tdLog.debug(f"start to excute {__file__}")
+		tdSql.init(conn.cursor())
+
+
+	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 or "taosd.exe" 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 getCfgPath(self):
+		buildPath = self.getBuildPath()
+		selfPath = os.path.dirname(os.path.realpath(__file__))
+
+		if ("community" in selfPath):
+			cfgPath = buildPath + "/../sim/dnode1/cfg/"
+		else:
+			cfgPath = buildPath + "/../sim/dnode1/cfg/"
+
+		return cfgPath
+
+	def installTaosd(self,bPath,cPath):
+		# os.system(f"rmtaos && mkdir -p {self.getBuildPath()}/build/lib/temp &&  mv {self.getBuildPath()}/build/lib/libtaos.so*  {self.getBuildPath()}/build/lib/temp/ ")
+		# os.system(f" mv {bPath}/build  {bPath}/build_bak ")
+		# os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so  {self.getBuildPath()}/build/lib/libtaos.so_bak ")
+		# os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so.1  {self.getBuildPath()}/build/lib/libtaos.so.1_bak ")
+
+		packagePath="/usr/local/src/"
+		packageName="TDengine-server-3.0.1.0-Linux-x64.tar.gz"
+		os.system(f"cd {packagePath} &&  tar xvf  TDengine-server-3.0.1.0-Linux-x64.tar.gz && cd TDengine-server-3.0.1.0 && ./install.sh  -e no  " )
+		tdDnodes.stop(1)
+		print(f"start taosd: nohup taosd -c {cPath} & ")
+		os.system(f" nohup taosd -c {cPath} & " )
+		sleep(1)
+
+
+		
+	def buildTaosd(self,bPath):
+		# os.system(f"mv {bPath}/build_bak  {bPath}/build ")
+		os.system(f" cd {bPath}  &&  make install ")
+
+
+	def run(self):
+		print(f"start taosd run")
+		bPath=self.getBuildPath()
+		cPath=self.getCfgPath()
+		dbname = "test"
+		stb = f"{dbname}.meters"
+		self.installTaosd(bPath,cPath)
+		os.system("echo 'debugFlag 143' > /etc/taos/taos.cfg ")
+		tableNumbers=100
+		recordNumbers1=100
+		recordNumbers2=1000
+		#tdsqlF=tdCom.newTdSql()
+		#print(tdsqlF)
+
+		oldServerVersion = '3.0.1.0' 
+		#tdsqlF.query(f"SELECT SERVER_VERSION();")
+		#print(tdsqlF.query(f"SELECT SERVER_VERSION();"))
+		#oldServerVersion=tdsqlF.queryResult[0][0]
+		#tdLog.info(f"Base server version is {oldServerVersion}")
+		#tdsqlF.query(f"SELECT CLIENT_VERSION();")
+		#	
+		## the oldClientVersion can't be updated in the same python process,so the version is new compiled verison
+		#oldClientVersion=tdsqlF.queryResult[0][0]
+
+		#tdLog.info(f"Base client version is {oldClientVersion}")
+
+		tdLog.printNoPrefix(f"==========step1:prepare and check data in old version-{oldServerVersion}")
+		tdLog.info(f" LD_LIBRARY_PATH=/usr/lib  taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
+		os.system(f"LD_LIBRARY_PATH=/usr/lib taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
+		sleep(3)
+
+		# tdsqlF.query(f"select count(*) from {stb}")
+		# tdsqlF.checkData(0,0,tableNumbers*recordNumbers1)
+		os.system("pkill taosd")
+		sleep(2)
+
+		print(f"start taosd: nohup taosd -c {cPath} & ")
+		os.system(f" nohup taosd -c {cPath} & " )
+		sleep(10)
+		tdLog.info(" LD_LIBRARY_PATH=/usr/lib  taosBenchmark -f 0-others/compa4096.json -y  ")
+		os.system("LD_LIBRARY_PATH=/usr/lib  taosBenchmark -f 0-others/compa4096.json -y")
+		os.system("pkill -9 taosd")
+
+
+
+		tdLog.printNoPrefix("==========step2:update new version ")
+		self.buildTaosd(bPath)
+		tdDnodes.start(1)
+		sleep(1)
+		tdsql=tdCom.newTdSql()
+		print(tdsql)
+
+
+		tdsql.query(f"SELECT SERVER_VERSION();")
+		nowServerVersion=tdsql.queryResult[0][0]
+		tdLog.info(f"New server version is {nowServerVersion}")
+		tdsql.query(f"SELECT CLIENT_VERSION();")
+		nowClientVersion=tdsql.queryResult[0][0]
+		tdLog.info(f"New client version is {nowClientVersion}")
+
+		tdLog.printNoPrefix(f"==========step3:prepare and check data in new version-{nowServerVersion}")
+		tdsql.query(f"select count(*) from {stb}")
+		tdsql.checkData(0,0,tableNumbers*recordNumbers1)		
+		os.system(f"taosBenchmark -t {tableNumbers} -n {recordNumbers2} -y  ")
+		tdsql.query(f"select count(*) from {stb}")
+		tdsql.checkData(0,0,tableNumbers*recordNumbers2)
+
+		tdsql=tdCom.newTdSql()
+		tdLog.printNoPrefix(f"==========step4:verify backticks in taos Sql-TD18542")
+		tdsql.execute("drop database if exists db")
+		tdsql.execute("create database db")
+		tdsql.execute("use db")
+		tdsql.execute("create stable db.stb1 (ts timestamp, c1 int) tags (t1 int);")
+		tdsql.execute("insert into db.ct1 using db.stb1 TAGS(1) values(now(),11);")
+		tdsql.error(" insert into `db.ct2` using db.stb1 TAGS(9) values(now(),11);")
+		tdsql.error(" insert into db.`db.ct2` using db.stb1 TAGS(9) values(now(),11);")
+		tdsql.execute("insert into `db`.ct3 using db.stb1 TAGS(3) values(now(),13);")
+		tdsql.query("select * from db.ct3")
+		tdsql.checkData(0,1,13)
+		tdsql.execute("insert into db.`ct4` using db.stb1 TAGS(4) values(now(),14);")
+		tdsql.query("select * from db.ct4")
+		tdsql.checkData(0,1,14)
+		tdsql.query("describe  information_schema.ins_databases;")
+		qRows=tdsql.queryRows   
+		for i in  range(qRows) :
+			if tdsql.queryResult[i][0]=="retentions" :
+				return True
+			else:
+				return False
+	def stop(self):
+		tdSql.close()
+		tdLog.success(f"{__file__} successfully executed")
 
 tdCases.addLinux(__file__, TDTestCase())
 tdCases.addWindows(__file__, TDTestCase())