TD-10431 dnode test01

include/dnode/mnode/mnode.h

@@ -57,6 +57,7 @@ typedef struct {
   int32_t           sver;
   int32_t           statusInterval;
   int32_t           mnodeEqualVnodeNum;
+  int32_t           shellActivityTimer;
   char             *timezone;
   char             *locale;
   char             *charset;

source/dnode/mgmt/impl/src/dndMnode.c

@@ -334,6 +334,7 @@ static void dndInitMnodeOption(SDnode *pDnode, SMnodeOpt *pOption) {
   pOption->sver = pDnode->opt.sver;
   pOption->statusInterval = pDnode->opt.statusInterval;
   pOption->mnodeEqualVnodeNum = pDnode->opt.mnodeEqualVnodeNum;
+  pOption->shellActivityTimer = pDnode->opt.shellActivityTimer;
   pOption->timezone = pDnode->opt.timezone;
   pOption->charset = pDnode->opt.charset;
   pOption->locale = pDnode->opt.locale;
@@ -675,7 +676,7 @@ void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   SMnodeMgmt *pMgmt = &pDnode->mmgmt;
   SMnode     *pMnode = dndAcquireMnode(pDnode);
-  if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) {
+  if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pReadQ, pMsg) != 0) {
     SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
     rpcSendResponse(&rsp);
     rpcFreeCont(pMsg->pCont);

source/dnode/mgmt/impl/test/test01/CMakeLists.txt

@@ -2,8 +2,8 @@ add_executable(dndTest01 "")
 
 target_sources(dndTest01
     PRIVATE
-    "test01.cpp"
-    "../util/deploy.cpp"
+    "dndTest01.cpp"
+    "../util/dndTestDeploy.cpp"
 )
 
 target_link_libraries(
@@ -18,6 +18,7 @@ target_include_directories(dndTest01
     PUBLIC
     "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
     "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../util"
 )
 
 enable_testing()

source/dnode/mgmt/impl/test/test01/dndTest01.cpp

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "dndTestDeploy.h"
+
+class DndTest01 : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    pServer = createServer("/tmp/dndTest01");
+    pClient = createClient("root", "taosdata");
+  }
+  void TearDown() override {
+    dropServer(pServer);
+    dropClient(pClient);
+  }
+
+  SServer* pServer;
+  SClient* pClient;
+};
+
+TEST_F(DndTest01, connectMsg) {
+  SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
+  pReq->pid = htonl(1234);
+  strcpy(pReq->app, "test01");
+  strcpy(pReq->db, "");
+
+  SRpcMsg rpcMsg = {0};
+  rpcMsg.pCont = pReq;
+  rpcMsg.contLen = sizeof(SConnectMsg);
+  rpcMsg.msgType = TSDB_MSG_TYPE_CONNECT;
+
+  sendMsg(pClient, &rpcMsg);
+
+  SConnectRsp* pRsp = (SConnectRsp*)pClient->pRsp;
+  ASSERT(pRsp);
+  pRsp->acctId = htonl(pRsp->acctId);
+  pRsp->clusterId = htonl(pRsp->clusterId);
+  pRsp->connId = htonl(pRsp->connId);
+  pRsp->epSet.port[0] = htonl(pRsp->epSet.port[0]);
+
+  EXPECT_EQ(pRsp->acctId, 1);
+  EXPECT_GT(pRsp->clusterId, 0);
+  EXPECT_GT(pRsp->connId, 1);
+  EXPECT_EQ(pRsp->superAuth, 1);
+  EXPECT_EQ(pRsp->readAuth, 1);
+  EXPECT_EQ(pRsp->writeAuth, 1);
+
+  EXPECT_EQ(pRsp->epSet.inUse, 0);
+  EXPECT_EQ(pRsp->epSet.numOfEps, 1);
+  EXPECT_EQ(pRsp->epSet.port[0], 9527);
+  EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
+}
+
+// TEST_F(DndTest01, heartbeatMsg) {
+//   SHeartBeatMsg* pReq = (SHeartBeatMsg*)rpcMallocCont(sizeof(SHeartBeatMsg));
+//   pReq->connId = htonl(1);
+//   pReq->pid = htonl(1234);
+//   pReq->numOfQueries = htonl(0);
+//   pReq->numOfStreams = htonl(0);
+//   strcpy(pReq->app, "test01");
+
+//   SRpcMsg rpcMsg = {0};
+//   rpcMsg.pCont = pReq;
+//   rpcMsg.contLen = sizeof(SHeartBeatMsg);
+//   rpcMsg.msgType = TSDB_MSG_TYPE_HEARTBEAT;
+
+//   sendMsg(pClient, &rpcMsg);
+
+//   SHeartBeatRsp* pRsp = (SHeartBeatRsp*)pClient->pRsp;
+//   ASSERT(pRsp);
+  // pRsp->epSet.port[0] = htonl(pRsp->epSet.port[0]);
+
+//   EXPECT_EQ(htonl(pRsp->connId), 1);
+//   EXPECT_GT(htonl(pRsp->queryId), 0);
+//   EXPECT_GT(htonl(pRsp->streamId), 1);
+//   EXPECT_EQ(htonl(pRsp->totalDnodes), 1);
+//   EXPECT_EQ(htonl(pRsp->onlineDnodes), 1);
+//   EXPECT_EQ(pRsp->killConnection, 0);
+  // EXPECT_EQ(pRsp->epSet.inUse, 0);
+  // EXPECT_EQ(pRsp->epSet.numOfEps, 1);
+  // EXPECT_EQ(pRsp->epSet.port[0], 9527);
+  // EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
+// }

source/dnode/mgmt/impl/test/test01/test01.cpp → source/dnode/mgmt/impl/test/util/dndTestDeploy.cpp

@@ -13,22 +13,16 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <gtest/gtest.h>
-#include "os.h"
-
-#include "dnode.h"
-#include "taosmsg.h"
-#include "tconfig.h"
-#include "tglobal.h"
-#include "tnote.h"
-#include "trpc.h"
-#include "tthread.h"
-#include "ulog.h"
-
-typedef struct {
-  SDnode*    pDnode;
-  pthread_t* threadId;
-} SServer;
+#include "dndTestDeploy.h"
+
+void initLog(char *path) {
+  mDebugFlag = 207;
+  char temp[PATH_MAX];
+  snprintf(temp, PATH_MAX, "%s/taosdlog", path);
+  if (taosInitLog(temp, tsNumOfLogLines, 1) != 0) {
+    printf("failed to init log file\n");
+  }
+}
 
 void* runServer(void* param) {
   SServer* pServer = (SServer*)param;
@@ -38,7 +32,7 @@ void* runServer(void* param) {
   }
 }
 
-void initOption(SDnodeOpt* pOption) {
+void initOption(SDnodeOpt* pOption, char *path) {
   pOption->sver = 1;
   pOption->numOfCores = 1;
   pOption->numOfSupportMnodes = 1;
@@ -51,15 +45,18 @@ void initOption(SDnodeOpt* pOption) {
   pOption->maxShellConns = 1000;
   pOption->shellActivityTimer = 30;
   pOption->serverPort = 9527;
-  strncpy(pOption->dataDir, "./test01");
+  strcpy(pOption->dataDir, path);
   strcpy(pOption->localEp, "localhost:9527");
   strcpy(pOption->localFqdn, "localhost");
   strcpy(pOption->firstEp, "localhost:9527");
+
+  taosRemoveDir(path);
+  taosMkDir(path);
 }
 
-SServer* createServer() {
+SServer* createServer(char *path) {
   SDnodeOpt option = {0};
-  initOption(&option);
+  initOption(&option, path);
 
   SDnode* pDnode = dndInit(&option);
   ASSERT(pDnode);
@@ -80,24 +77,18 @@ void dropServer(SServer* pServer) {
   }
 }
 
-typedef struct {
-  void*    clientRpc;
-  SRpcMsg* pRsp;
-  tsem_t   sem;
-} SClient;
-
-static void processClientRsp(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
+void processClientRsp(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
   SClient* pClient = (SClient*)parent;
   pClient->pRsp = pMsg;
+  taosMsleep(100000);
   tsem_post(&pClient->sem);
 }
 
-SClient* createClient() {
+SClient* createClient(char *user, char *pass) {
   SClient* pClient = (SClient*)calloc(1, sizeof(SClient));
   ASSERT(pClient);
 
-  char  secretEncrypt[32] = {0};
-  char* pass = "taosdata";
+  char secretEncrypt[32] = {0};
   taosEncryptPass((uint8_t*)pass, strlen(pass), secretEncrypt);
 
   SRpcInit rpcInit;
@@ -108,7 +99,7 @@ SClient* createClient() {
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.idleTime = 30 * 1000;
-  rpcInit.user = "root";
+  rpcInit.user = user;
   rpcInit.ckey = "key";
   rpcInit.parent = pClient;
   rpcInit.secret = (char*)secretEncrypt;
@@ -119,6 +110,8 @@ SClient* createClient() {
   ASSERT(pClient->clientRpc);
 
   tsem_init(&pClient->sem, 0, 0);
+
+  return pClient;
 }
 
 void dropClient(SClient* pClient) {
@@ -136,43 +129,3 @@ void sendMsg(SClient* pClient, SRpcMsg* pMsg) {
   rpcSendRequest(pClient->clientRpc, &epSet, pMsg, NULL);
   tsem_wait(&pClient->sem);
 }
-
-class DndTest01 : public ::testing::Test {
- protected:
-  void SetUp() override {
-    pServer = createServer();
-    pClient = createClient();
-  }
-  void TearDown() override {
-    dropServer(pServer);
-    dropClient(pClient);
-  }
-
-  SServer* pServer;
-  SClient* pClient;
-};
-
-TEST_F(DndTest01, connectMsg) {
-  SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
-  pReq->pid = 1234;
-  strcpy(pReq->app, "test01");
-  strcpy(pReq->app, "");
-
-  SRpcMsg rpcMsg = {.pCont = pReq, .contLen = sizeof(SConnectMsg), .msgType = TSDB_MSG_TYPE_AUTH};
-
-  sendMsg(pClient, &rpcMsg);
-
-  SConnectRsp* pRsp = (SConnectRsp*)pClient->pRsp;
-  EXPECT_NE(pRsp, NULL);
-  EXPECT_EQ(pRsp->acctId, 1);
-  EXPECT_GT(pRsp->clusterId, 0);
-  EXPECT_GT(pRsp->connId, 1);
-  EXPECT_EQ(pRsp->superAuth, 1);
-  EXPECT_EQ(pRsp->readAuth, 1);
-  EXPECT_EQ(pRsp->writeAuth, 1);
-
-  EXPECT_EQ(pRsp->epSet.inUse, 0);
-  EXPECT_EQ(pRsp->epSet.numOfEps, 1);
-  EXPECT_EQ(pRsp->epSet.port[0], 9527);
-  EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
-}

source/dnode/mgmt/impl/test/util/dndTestDeploy.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <gtest/gtest.h>
+#include "os.h"
+
+#include "dnode.h"
+#include "taosmsg.h"
+#include "tconfig.h"
+#include "tglobal.h"
+#include "tnote.h"
+#include "trpc.h"
+#include "tthread.h"
+#include "ulog.h"
+
+typedef struct {
+  SDnode*    pDnode;
+  pthread_t* threadId;
+} SServer;
+
+typedef struct {
+  void*    clientRpc;
+  SRpcMsg* pRsp;
+  tsem_t   sem;
+} SClient;
+
+SServer* createServer(char* path);
+void     dropServer(SServer* pServer);
+SClient* createClient(char *user, char *pass);
+void     dropClient(SClient* pClient);
+void     sendMsg(SClient* pClient, SRpcMsg* pMsg);

source/dnode/mnode/impl/inc/mndUser.h

@@ -24,7 +24,7 @@ extern "C" {
 
 int32_t   mndInitUser(SMnode *pMnode);
 void      mndCleanupUser(SMnode *pMnode);
-SUserObj *mndAcquireUser(SMnode *pMnode, const char *userName);
+SUserObj *mndAcquireUser(SMnode *pMnode, char *userName);
 void      mndReleaseUser(SMnode *pMnode, SUserObj *pUser);
 
 #ifdef __cplusplus

source/dnode/mnode/impl/src/mndProfile.c

@@ -78,7 +78,7 @@ int32_t mndInitProfile(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_HEARTBEAT, mndProcessHeartBeatMsg);
   mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CONNECT, mndProcessConnectMsg);
   mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_QUERY, mndProcessKillQueryMsg);
-  mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CONNECT, mndProcessKillStreamMsg);
+  mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_STREAM, mndProcessKillStreamMsg);
   mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_CONN, mndProcessKillConnectionMsg);
 
   mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_CONNS, mndGetConnsMeta);
@@ -237,7 +237,6 @@ static int32_t mndProcessConnectMsg(SMnodeMsg *pMsg) {
     mndReleaseUser(pMnode, pUser);
   }
 
-  pRsp->acctId = htonl(pUser->acctId);
   pRsp->clusterId = htonl(pMnode->clusterId);
   pRsp->connId = htonl(pConn->connId);
   mndGetMnodeEpSet(pMnode, &pRsp->epSet);
@@ -330,7 +329,7 @@ static int32_t mndProcessHeartBeatMsg(SMnodeMsg *pMsg) {
   }
 
   pRsp->connId = htonl(pConn->connId);
-  pRsp->totalDnodes = htnol(1);
+  pRsp->totalDnodes = htonl(1);
   pRsp->onlineDnodes = htonl(1);
   mndGetMnodeEpSet(pMnode, &pRsp->epSet);
   mndReleaseConn(pMnode, pConn);

source/dnode/mnode/impl/src/mndUser.c

@@ -167,9 +167,9 @@ static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pSrcUser, SUserObj *pDs
   return 0;
 }
 
-SUserObj *mndAcquireUser(SMnode *pMnode, const char *userName) {
+SUserObj *mndAcquireUser(SMnode *pMnode, char *userName) {
   SSdb *pSdb = pMnode->pSdb;
-  return sdbAcquire(pSdb, SDB_USER, &userName);
+  return sdbAcquire(pSdb, SDB_USER, userName);
 }
 
 void mndReleaseUser(SMnode *pMnode, SUserObj *pUser) {

source/dnode/mnode/impl/src/mnode.c

@@ -206,6 +206,7 @@ static int32_t mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
   pMnode->sver = pOption->sver;
   pMnode->statusInterval = pOption->statusInterval;
   pMnode->mnodeEqualVnodeNum = pOption->mnodeEqualVnodeNum;
+  pMnode->shellActivityTimer = pOption->shellActivityTimer;
   pMnode->timezone = strdup(pOption->timezone);
   pMnode->locale = strdup(pOption->locale);
   pMnode->charset = strdup(pOption->charset);

source/libs/transport/src/rpcMain.c

@@ -507,7 +507,6 @@ void rpcSendRedirectRsp(void *thandle, const SEpSet *pEpSet) {
 }
 
 int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
-#if 0
   SRpcConn  *pConn = (SRpcConn *)thandle;
   if (pConn->user[0] == 0) return -1;
 
@@ -516,9 +515,6 @@ int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
   // pInfo->serverIp = pConn->destIp;
   
   tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user));
-#else 
-  strcpy(pInfo->user, "root");
-#endif
   return 0;
 }