[td-32] merge code

src/dnode/inc/dnodeMClient.h

@@ -20,8 +20,10 @@
 extern "C" {
 #endif
 
-int32_t dnodeInitMClient();
-void    dnodeCleanupMClient();
+int32_t  dnodeInitMClient();
+void     dnodeCleanupMClient();
+void     dnodeSendMsgToMnode(SRpcMsg *rpcMsg);
+uint32_t dnodeGetMnodeMasteIp();
 
 #ifdef __cplusplus
 }

src/dnode/inc/dnodeMgmt.h

@@ -23,6 +23,7 @@ extern "C" {
 int32_t dnodeInitMgmt();
 void    dnodeCleanupMgmt();
 void    dnodeMgmt(SRpcMsg *rpcMsg);
+void    dnodeUpdateDnodeId(int32_t dnodeId);
 
 void*   dnodeGetVnode(int32_t vgId);
 int32_t dnodeGetVnodeStatus(void *pVnode);

src/dnode/inc/dnodeModule.h

@@ -23,6 +23,7 @@ extern "C" {
 int32_t dnodeInitModules();
 void dnodeCleanUpModules();
 void dnodeStartModules();
+void dnodeProcessModuleStatus(uint32_t moduleStatus);
 
 #ifdef __cplusplus
 }

src/dnode/src/dnodeMClient.c

@@ -18,16 +18,35 @@
 #include "taosmsg.h"
 #include "tlog.h"
 #include "trpc.h"
+#include "tutil.h"
 #include "dnode.h"
+#include "dnodeMClient.h"
+#include "dnodeModule.h"
+#include "dnodeMgmt.h"
 
-static void (*dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
+static bool   dnodeReadMnodeIpList();
+static void   dnodeSaveMnodeIpList();
 static void   dnodeProcessRspFromMnode(SRpcMsg *pMsg);
 static void   dnodeProcessStatusRsp(SRpcMsg *pMsg);
-static void  *tsDnodeMClientRpc;
+static void (*tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
+static void  *tsDnodeMClientRpc = NULL;
+static SRpcIpSet tsDnodeMnodeIpList  = {0};
 
 int32_t dnodeInitMClient() {
-  dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
+  if (!dnodeReadMnodeIpList()) {
+    dTrace("failed to read mnode iplist, set it from cfg file");
+    memset(&tsDnodeMnodeIpList, 0, sizeof(SRpcIpSet));
+    tsDnodeMnodeIpList.port = tsMnodeDnodePort;
+    tsDnodeMnodeIpList.numOfIps = 1;
+    tsDnodeMnodeIpList.ip[0] = inet_addr(tsMasterIp);
+    if (tsSecondIp[0]) {
+      tsDnodeMnodeIpList.numOfIps = 2;
+      tsDnodeMnodeIpList.ip[1] = inet_addr(tsSecondIp);
+    }
+  }
 
+  tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
+  
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localIp      = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
@@ -35,17 +54,20 @@ int32_t dnodeInitMClient() {
   rpcInit.label        = "DND-MC";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = dnodeProcessRspFromMnode;
-  rpcInit.sessions     = TSDB_SESSIONS_PER_DNODE;
+  rpcInit.sessions     = 100;
   rpcInit.connType     = TAOS_CONN_CLIENT;
-  rpcInit.idleTime     = tsShellActivityTimer * 1000;
+  rpcInit.idleTime     = tsShellActivityTimer * 2000;
+  rpcInit.user         = "t";
+  rpcInit.ckey         = "key";
+  rpcInit.secret       = "secret";
 
   tsDnodeMClientRpc = rpcOpen(&rpcInit);
   if (tsDnodeMClientRpc == NULL) {
-    dError("failed to init connection from mgmt");
+    dError("failed to init mnode rpc client");
     return -1;
   }
 
-  dPrint("client connection to mgmt is opened");
+  dPrint("mnode rpc client is opened");
   return 0;
 }
 
@@ -53,18 +75,122 @@ void dnodeCleanupMClient() {
   if (tsDnodeMClientRpc) {
     rpcClose(tsDnodeMClientRpc);
     tsDnodeMClientRpc = NULL;
+    dPrint("mnode rpc client is closed");
   }
 }
 
 static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) {
-  if (dnodeProcessMgmtRspFp[pMsg->msgType]) {
-    (*dnodeProcessMgmtRspFp[pMsg->msgType])(pMsg);
+  if (tsDnodeProcessMgmtRspFp[pMsg->msgType]) {
+    (*tsDnodeProcessMgmtRspFp[pMsg->msgType])(pMsg);
   } else {
-    dError("%s is not processed in mclient", taosMsg[pMsg->msgType]);
+    dError("%s is not processed in mnode rpc client", taosMsg[pMsg->msgType]);
   }
 
   rpcFreeCont(pMsg->pCont);
 }
 
 static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
+  if (pMsg->code != TSDB_CODE_SUCCESS) {
+    dError("status rsp is received, error:%s", tstrerror(pMsg->code));
+    return;
+  }
+
+  SDMStatusRsp *pStatusRsp = pMsg->pCont;
+  if (pStatusRsp->ipList.numOfIps <= 0) {
+    dError("status msg is invalid, num of ips is %d", pStatusRsp->ipList.numOfIps);
+    return;
+  }
+
+  pStatusRsp->ipList.port = htons(pStatusRsp->ipList.port);
+  for (int32_t i = 0; i < pStatusRsp->ipList.numOfIps; ++i) {
+    pStatusRsp->ipList.ip[i] = htonl(pStatusRsp->ipList.ip[i]);
+  }
+
+  //dTrace("status msg is received, result:%s", tstrerror(pMsg->code));
+
+  if (memcmp(&(pStatusRsp->ipList), &tsDnodeMnodeIpList, sizeof(SRpcIpSet)) != 0) {
+    dPrint("mnode ip list is changed, numOfIps:%d inUse:%d", pStatusRsp->ipList.numOfIps, pStatusRsp->ipList.inUse);
+    memcpy(&tsDnodeMnodeIpList, &pStatusRsp->ipList, sizeof(SRpcIpSet));
+    for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; ++i) {
+      dPrint("mnode index:%d ip:%s", i, taosIpStr(tsDnodeMnodeIpList.ip[i]));
+    }
+    dnodeSaveMnodeIpList();
+  }
+
+  SDnodeState *pState  = &pStatusRsp->dnodeState;
+  pState->numOfVnodes  = htonl(pState->numOfVnodes);
+  pState->moduleStatus = htonl(pState->moduleStatus);
+  pState->createdTime  = htonl(pState->createdTime);
+  pState->dnodeId      = htonl(pState->dnodeId);
+  
+  dnodeProcessModuleStatus(pState->moduleStatus);
+  dnodeUpdateDnodeId(pState->dnodeId);
 }
+
+void dnodeSendMsgToMnode(SRpcMsg *rpcMsg) {
+  rpcSendRequest(tsDnodeMClientRpc, &tsDnodeMnodeIpList, rpcMsg);
+}
+
+static bool dnodeReadMnodeIpList() {
+  char ipFile[TSDB_FILENAME_LEN] = {0};
+  sprintf(ipFile, "%s/iplist", tsDnodeDir);
+
+  FILE *fp = fopen(ipFile, "r");
+  if (!fp) return false;
+  
+  char option[32] = {0};
+  int32_t value = 0;
+  int32_t num = 0;
+  
+  num = fscanf(fp, "%s %d", option, &value);
+  if (num != 2) return false;
+  if (strcmp(option, "inUse") != 0) return false;
+  tsDnodeMnodeIpList.inUse = (int8_t)value;;
+
+  num = fscanf(fp, "%s %d", option, &value);
+  if (num != 2) return false;
+  if (strcmp(option, "numOfIps") != 0) return false;
+  tsDnodeMnodeIpList.numOfIps = (int8_t)value;
+
+  num = fscanf(fp, "%s %d", option, &value);
+  if (num != 2) return false;
+  if (strcmp(option, "port") != 0) return false;
+  tsDnodeMnodeIpList.port = (uint16_t)value;
+
+  for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
+    num = fscanf(fp, "%s %d", option, &value);
+    if (num != 2) return false;
+    if (strncmp(option, "ip", 2) != 0) return false;
+    tsDnodeMnodeIpList.ip[i] = (uint32_t)value;
+  }
+
+  fclose(fp);
+  dPrint("read mnode iplist successed");
+  for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
+    dPrint("mnode index:%d ip:%s", i, taosIpStr(tsDnodeMnodeIpList.ip[i]));
+  } 
+
+  return true;
+}
+
+static void dnodeSaveMnodeIpList() {
+  char ipFile[TSDB_FILENAME_LEN] = {0};
+  sprintf(ipFile, "%s/iplist", tsDnodeDir);
+
+  FILE *fp = fopen(ipFile, "w");
+  if (!fp) return;
+
+  fprintf(fp, "inUse %d\n", tsDnodeMnodeIpList.inUse);
+  fprintf(fp, "numOfIps %d\n", tsDnodeMnodeIpList.numOfIps);
+  fprintf(fp, "port %u\n", tsDnodeMnodeIpList.port);
+  for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
+    fprintf(fp, "ip%d %u\n", i, tsDnodeMnodeIpList.ip[i]);
+  }
+  
+  fclose(fp);
+  dPrint("save mnode iplist successed");
+}
+
+uint32_t dnodeGetMnodeMasteIp() {
+  return tsDnodeMnodeIpList.ip[0];
+}
\ No newline at end of file

src/dnode/src/dnodeMain.c

@@ -175,9 +175,9 @@ static int32_t dnodeInitSystem() {
   if (dnodeInitModules() != 0) return -1;
   if (dnodeInitRead() != 0) return -1;
   if (dnodeInitWrite() != 0) return -1;
-  if (dnodeInitMgmt() != 0) return -1;
-  if (dnodeInitMnode() != 0) return -1;
   if (dnodeInitMClient() != 0) return -1;
+  if (dnodeInitMnode() != 0) return -1;
+  if (dnodeInitMgmt() != 0) return -1;
   if (dnodeInitShell() != 0) return -1;
 
   dnodeStartModules();

src/dnode/src/dnodeMgmt.c

@@ -22,16 +22,18 @@
 #include "trpc.h"
 #include "tstatus.h"
 #include "tsdb.h"
+#include "ttime.h"
 #include "ttimer.h"
+#include "dnodeMClient.h"
 #include "dnodeMgmt.h"
 #include "dnodeRead.h"
 #include "dnodeWrite.h"
 
 typedef struct {
   int32_t vgId;     // global vnode group ID
-  int32_t vnode;
-  int32_t status;   // status: master, slave, notready, deleting
   int32_t refCount; // reference count
+  int8_t  dirty;
+  int8_t  status;   // status: master, slave, notready, deleting
   int64_t version;
   void   *wworker;
   void   *rworker;
@@ -48,6 +50,7 @@ static int32_t  dnodeOpenVnode(int32_t vnode, char *rootDir);
 static void     dnodeCleanupVnode(SVnodeObj *pVnode);
 static int32_t  dnodeCreateVnode(SMDCreateVnodeMsg *cfg);
 static void     dnodeDropVnode(SVnodeObj *pVnode);
+static void     dnodeDoDropVnode(SVnodeObj *pVnode);
 static void     dnodeProcessCreateVnodeMsg(SRpcMsg *pMsg);
 static void     dnodeProcessDropVnodeMsg(SRpcMsg *pMsg);
 static void     dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
@@ -55,17 +58,23 @@ static void     dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
 static void     dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
 static void   (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
 static void     dnodeSendStatusMsg(void *handle, void *tmrId);
+static void     dnodeReadDnodeId();
 
-static void * tsDnodeVnodesHash = NULL;
-static void  *tsDnodeTmr = NULL;
-static void  *tsStatusTimer = NULL;
+static void    *tsDnodeVnodesHash = NULL;
+static void    *tsDnodeTmr = NULL;
+static void    *tsStatusTimer = NULL;
+static uint32_t tsRebootTime;
+static int32_t  tsDnodeId = 0;
+static char     tsDnodeName[TSDB_DNODE_NAME_LEN];
 
 int32_t dnodeInitMgmt() {
+  dnodeReadDnodeId();
+
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeProcessDropVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE]  = dnodeProcessAlterVnodeMsg;
-  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM]  = dnodeProcessAlterStreamMsg;
-  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE]  = dnodeProcessConfigDnodeMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
 
   tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj), taosHashInt);
   if (tsDnodeVnodesHash == NULL) {
@@ -73,6 +82,8 @@ int32_t dnodeInitMgmt() {
     return -1;
   }
 
+  tsRebootTime = taosGetTimestampSec();
+
   tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
   if (tsDnodeTmr == NULL) {
     dError("failed to init dnode timer");
@@ -80,13 +91,6 @@ int32_t dnodeInitMgmt() {
   }
   taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
 
-  SMDCreateVnodeMsg cfg;
-  cfg.cfg.vgId = 1;
-  cfg.cfg.precision = 0;
-  cfg.vnode = 1;
-  cfg.cfg.maxSessions = 1000;
-  cfg.cfg.daysPerFile = 10;
-
   return dnodeOpenVnodes();
 }
 
@@ -96,8 +100,16 @@ void dnodeCleanupMgmt() {
     tsStatusTimer = NULL;
   }
 
+  if (tsDnodeTmr != NULL) {
+    taosTmrCleanUp(tsDnodeTmr);
+    tsDnodeTmr = NULL;
+  }
+
   dnodeCleanupVnodes();
-  taosCleanUpIntHash(tsDnodeVnodesHash);
+  if (tsDnodeVnodesHash == NULL) {
+    taosCleanUpIntHash(tsDnodeVnodesHash);
+    tsDnodeVnodesHash = NULL;
+  }
 }
 
 void dnodeMgmt(SRpcMsg *pMsg) {
@@ -113,7 +125,7 @@ void dnodeMgmt(SRpcMsg *pMsg) {
     rpcSendResponse(&rsp);
   }
 
-  rpcFreeCont(pMsg->pCont);  // free the received message
+  rpcFreeCont(pMsg->pCont);
 }
 
 void *dnodeGetVnode(int32_t vgId) {
@@ -152,8 +164,13 @@ void *dnodeGetVnodeTsdb(void *pVnode) {
   return ((SVnodeObj *)pVnode)->tsdb;
 }
 
-void dnodeReleaseVnode(void *pVnode) {
-  atomic_sub_fetch_32(&((SVnodeObj *) pVnode)->refCount, 1);
+void dnodeReleaseVnode(void *pVnodeRaw) {
+  SVnodeObj *pVnode = pVnodeRaw;
+  int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
+
+  if (count == 0 && pVnode->dirty) {
+    dnodeDoDropVnode(pVnode);
+  }
 }
 
 static int32_t dnodeOpenVnodes() {
@@ -171,8 +188,8 @@ static int32_t dnodeOpenVnodes() {
       int32_t vnode = atoi(de->d_name + 5);
       if (vnode == 0) continue;
 
-      char vnodeDir[TSDB_FILENAME_LEN];
-      sprintf(vnodeDir, "%s/%s", tsVnodeDir, de->d_name);
+      char vnodeDir[TSDB_FILENAME_LEN * 3];
+      snprintf(vnodeDir, TSDB_FILENAME_LEN * 3, "%s/%s", tsVnodeDir, de->d_name);
       int32_t code = dnodeOpenVnode(vnode, vnodeDir);
       if (code == 0) {
         numOfVnodes++;
@@ -181,7 +198,7 @@ static int32_t dnodeOpenVnodes() {
   }
   closedir(dir);
 
-  dPrint("all vnodes is opened, num:%d", numOfVnodes);
+  dPrint("dnode mgmt is opened, vnodes:%d", numOfVnodes);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -189,7 +206,7 @@ typedef void (*CleanupFp)(char *);
 static void dnodeCleanupVnodes() {
   int32_t num = taosGetIntHashSize(tsDnodeVnodesHash);
   taosCleanUpIntHashWithFp(tsDnodeVnodesHash, (CleanupFp)dnodeCleanupVnode);
-  dPrint("all vnodes is opened, num:%d", num);
+  dPrint("dnode mgmt is closed, vnodes:%d", num);
 }
 
 static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
@@ -203,9 +220,8 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
 
   //STsdbRepoInfo *tsdbInfo = tsdbGetStatus(pTsdb);
 
-  SVnodeObj vnodeObj;
-  vnodeObj.vgId     = vnode; //tsdbInfo->tsdbCfg.vgId;
-  vnodeObj.vnode    = vnode; //tsdbInfo->tsdbCfg.tsdbId;
+  SVnodeObj vnodeObj = {0};
+  vnodeObj.vgId     = vnode;//tsdbInfo->tsdbCfg.tsdbId;
   vnodeObj.status   = TSDB_VN_STATUS_NOT_READY;
   vnodeObj.refCount = 1;
   vnodeObj.version  = 0;
@@ -219,7 +235,7 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
 
   taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
 
-  dTrace("open vnode:%d in %s", vnodeObj.vnode, rootDir);
+  dTrace("open vnode:%d in %s", vnodeObj.vgId, rootDir);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -248,14 +264,13 @@ static void dnodeCleanupVnode(SVnodeObj *pVnode) {
     pVnode->tsdb = NULL;
   }
 
-  dTrace("cleanup vnode:%d", pVnode->vnode);
+  dTrace("cleanup vnode:%d", pVnode->vgId);
 }
 
 static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
   STsdbCfg tsdbCfg = {0};
-  tsdbCfg.vgId                = pVnodeCfg->cfg.vgId;
   tsdbCfg.precision           = pVnodeCfg->cfg.precision;
-  tsdbCfg.tsdbId              = pVnodeCfg->vnode;
+  tsdbCfg.tsdbId              = pVnodeCfg->cfg.vgId;
   tsdbCfg.maxTables           = pVnodeCfg->cfg.maxSessions;
   tsdbCfg.daysPerFile         = pVnodeCfg->cfg.daysPerFile;
   tsdbCfg.minRowsPerFileBlock = -1;
@@ -290,13 +305,12 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
 
   void *pTsdb = tsdbCreateRepo(rootDir, &tsdbCfg, NULL);
   if (pTsdb == NULL) {
-    dError("vgroup:%d, failed to create tsdb in vnode:%d, reason:%s", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode, tstrerror(terrno));
+    dError("vgroup:%d, failed to create tsdb in vnode, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(terrno));
     return terrno;
   }
 
-  SVnodeObj vnodeObj;
+  SVnodeObj vnodeObj = {0};
   vnodeObj.vgId     = pVnodeCfg->cfg.vgId;
-  vnodeObj.vnode    = pVnodeCfg->vnode;
   vnodeObj.status   = TSDB_VN_STATUS_NOT_READY;
   vnodeObj.refCount = 1;
   vnodeObj.version  = 0;
@@ -310,18 +324,11 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
 
   taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
 
-  dPrint("vgroup:%d, vnode:%d is created", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode);
+  dPrint("vgroup:%d, vnode:%d is created", vnodeObj.vgId, vnodeObj.vgId);
   return TSDB_CODE_SUCCESS;
 }
 
-static void dnodeDropVnode(SVnodeObj *pVnode) {
-  pVnode->status = TSDB_VN_STATUS_NOT_READY;
-
-  int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
-  if (count > 0) {
-    // wait refcount
-  }
-
+static void dnodeDoDropVnode(SVnodeObj *pVnode) {
   if (pVnode->tsdb) {
     tsdbDropRepo(pVnode->tsdb);
     pVnode->tsdb = NULL;
@@ -331,21 +338,33 @@ static void dnodeDropVnode(SVnodeObj *pVnode) {
   taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
 }
 
+static void dnodeDropVnode(SVnodeObj *pVnode) {
+  pVnode->status = TSDB_VN_STATUS_NOT_READY;
+  pVnode->dirty  = true;
+
+  int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
+  if (count > 0) {
+    dTrace("vgroup:%d, vnode will be dropped until refcount:%d is 0", pVnode->vgId, count);
+    return;
+  }
+
+  dnodeDoDropVnode(pVnode);
+}
+
 static void dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
   SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
-  pCreate->vnode           = htonl(pCreate->vnode);
   pCreate->cfg.vgId        = htonl(pCreate->cfg.vgId);
   pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
   pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
 
-  dTrace("vgroup:%d, start to create vnode:%d", pCreate->cfg.vgId, pCreate->vnode);
+  dTrace("vgroup:%d, start to create vnode:%d in dnode", pCreate->cfg.vgId, pCreate->cfg.vgId);
 
   SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
   if (pVnodeObj != NULL) {
     rpcRsp.code = TSDB_CODE_SUCCESS;
-    dPrint("vgroup:%d, vnode:%d is already exist", pCreate->cfg.vgId, pCreate->vnode);
+    dPrint("vgroup:%d, vnode is already exist", pCreate->cfg.vgId);
   } else {
     rpcRsp.code = dnodeCreateVnode(pCreate);
   }
@@ -359,6 +378,8 @@ static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
   SMDDropVnodeMsg *pDrop = rpcMsg->pCont;
   pDrop->vgId = htonl(pDrop->vgId);
 
+  dTrace("vgroup:%d, start to drop vnode in dnode", pDrop->vgId);
+
   SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pDrop->vgId);
   if (pVnodeObj != NULL) {
     dnodeDropVnode(pVnodeObj);
@@ -374,11 +395,12 @@ static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
   SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
-  pCreate->vnode           = htonl(pCreate->vnode);
   pCreate->cfg.vgId        = htonl(pCreate->cfg.vgId);
   pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
   pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
 
+  dTrace("vgroup:%d, start to alter vnode in dnode", pCreate->cfg.vgId);
+
   SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
   if (pVnodeObj != NULL) {
     rpcRsp.code = TSDB_CODE_SUCCESS;
@@ -401,66 +423,109 @@ static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
 }
 
 static void dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
-//  SCfgDnodeMsg *pCfg = (SCfgDnodeMsg *)pCont;
-//
-//  int32_t code = tsCfgDynamicOptions(pCfg->config);
-//  dnodeSendRspToMnode(pConn, msgType + 1, code, NULL, 0);
+  SMDCfgDnodeMsg *pCfg = (SMDCfgDnodeMsg *)pMsg->pCont;
+  int32_t code = tsCfgDynamicOptions(pCfg->config);
+
+  SRpcMsg rpcRsp = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0, .code = code, .msgType = 0};
+  rpcSendResponse(&rpcRsp);
+}
+
+static void dnodeBuildVloadMsg(char *pNode, void * param) {
+  SVnodeObj *pVnode = (SVnodeObj *) pNode;
+  if (pVnode->dirty) return;
+  
+  SDMStatusMsg *pStatus = param;
+  if (pStatus->openVnodes >= TSDB_MAX_VNODES) return;
+  
+  SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++];
+  pLoad->vgId = htonl(pVnode->vgId);
+  pLoad->vnode = htonl(pVnode->vgId);
+  pLoad->status = pVnode->status;
 }
 
 static void dnodeSendStatusMsg(void *handle, void *tmrId) {
-  taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
+  if (tsDnodeTmr == NULL) {
+     dError("dnode timer is already released");
+    return;
+  }
+
   if (tsStatusTimer == NULL) {
+    taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
     dError("failed to start status timer");
     return;
   }
 
-//  int32_t contLen = sizeof(SDMStatusMsg) + dnodeGetVnodesNum() * sizeof(SVnodeLoad);
-//  SDMStatusMsg *pStatus = rpcMallocCont(contLen);
-//  if (pStatus == NULL) {
-//    dError("Failed to malloc status message");
-//    return;
-//  }
-//
-//  int32_t totalVnodes = dnodeGetVnodesNum();
-//
-//  pStatus->version          = htonl(tsVersion);
-//  pStatus->privateIp        = htonl(inet_addr(tsPrivateIp));
-//  pStatus->publicIp         = htonl(inet_addr(tsPublicIp));
-//  pStatus->lastReboot       = htonl(tsRebootTime);
-//  pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
-//  pStatus->openVnodes       = htons((uint16_t) totalVnodes);
-//  pStatus->numOfCores       = htons((uint16_t) tsNumOfCores);
-//  pStatus->diskAvailable    = tsAvailDataDirGB;
-//  pStatus->alternativeRole  = (uint8_t) tsAlternativeRole;
-//
-//  SVnodeLoad *pLoad = (SVnodeLoad *)pStatus->load;
-
-  //TODO loop all vnodes
-  //  for (int32_t vnode = 0, count = 0; vnode <= totalVnodes; ++vnode) {
-  //    if (vnodeList[vnode].cfg.maxSessions <= 0) continue;
-  //
-  //    SVnodeObj *pVnode = vnodeList + vnode;
-  //    pLoad->vnode = htonl(vnode);
-  //    pLoad->vgId = htonl(pVnode->cfg.vgId);
-  //    pLoad->status = (uint8_t)vnodeList[vnode].vnodeStatus;
-  //    pLoad->syncStatus =(uint8_t)vnodeList[vnode].syncStatus;
-  //    pLoad->accessState = (uint8_t)(pVnode->accessState);
-  //    pLoad->totalStorage = htobe64(pVnode->vnodeStatistic.totalStorage);
-  //    pLoad->compStorage = htobe64(pVnode->vnodeStatistic.compStorage);
-  //    if (pVnode->vnodeStatus == TSDB_VN_STATUS_MASTER) {
-  //      pLoad->pointsWritten = htobe64(pVnode->vnodeStatistic.pointsWritten);
-  //    } else {
-  //      pLoad->pointsWritten = htobe64(0);
-  //    }
-  //    pLoad++;
-  //
-  //    if (++count >= tsOpenVnodes) {
-  //      break;
-  //    }
-  //  }
-
-//  dnodeSendMsgToMnode(TSDB_MSG_TYPE_STATUS, pStatus, contLen);
+  int32_t contLen = sizeof(SDMStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
+  SDMStatusMsg *pStatus = rpcMallocCont(contLen);
+  if (pStatus == NULL) {
+    taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
+    dError("failed to malloc status message");
+    return;
+  }
+
+  strcpy(pStatus->dnodeName, tsDnodeName);
+  pStatus->version          = htonl(tsVersion);
+  pStatus->dnodeId          = htonl(tsDnodeId);
+  pStatus->privateIp        = htonl(inet_addr(tsPrivateIp));
+  pStatus->publicIp         = htonl(inet_addr(tsPublicIp));
+  pStatus->lastReboot       = htonl(tsRebootTime);
+  pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
+  pStatus->numOfCores       = htons((uint16_t) tsNumOfCores);
+  pStatus->diskAvailable    = tsAvailDataDirGB;
+  pStatus->alternativeRole  = (uint8_t) tsAlternativeRole;
+
+  taosVisitIntHashWithFp(tsDnodeVnodesHash, dnodeBuildVloadMsg, pStatus);
+  contLen = sizeof(SDMStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
+  pStatus->openVnodes = htons(pStatus->openVnodes);
+  
+  SRpcMsg rpcMsg = {
+    .pCont   = pStatus,
+    .contLen = contLen,
+    .msgType = TSDB_MSG_TYPE_DM_STATUS
+  };
+
+  dnodeSendMsgToMnode(&rpcMsg);
+  taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
+}
+
+static void dnodeReadDnodeId() {
+  char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
+  sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
+
+  FILE *fp = fopen(dnodeIdFile, "r");
+  if (!fp) return;
+  
+  char option[32] = {0};
+  int32_t value = 0;
+  int32_t num = 0;
+  
+  num = fscanf(fp, "%s %d", option, &value);
+  if (num != 2) return;
+  if (strcmp(option, "dnodeId") != 0) return;
+  tsDnodeId = value;;
+
+  fclose(fp);
+  dPrint("read dnodeId:%d successed", tsDnodeId);
 }
 
+static void dnodeSaveDnodeId() {
+  char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
+  sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
 
+  FILE *fp = fopen(dnodeIdFile, "w");
+  if (!fp) return;
+
+  fprintf(fp, "dnodeId %d\n", tsDnodeId);
+
+  fclose(fp);
+  dPrint("save dnodeId successed");
+}
+
+void dnodeUpdateDnodeId(int32_t dnodeId) {
+  if (tsDnodeId == 0) {
+    dPrint("dnodeId is set to %d", dnodeId);  
+    tsDnodeId = dnodeId;
+    dnodeSaveDnodeId();
+  }
+}
 

src/dnode/src/dnodeMnode.c

@@ -40,41 +40,42 @@ int32_t dnodeInitMnode() {
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localIp      = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
   rpcInit.localPort    = tsDnodeMnodePort;
-  rpcInit.label        = "DND-mgmt";
+  rpcInit.label        = "DND-MS";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = dnodeProcessMsgFromMnode;
-  rpcInit.sessions     = TSDB_SESSIONS_PER_DNODE;
+  rpcInit.sessions     = 100;
   rpcInit.connType     = TAOS_CONN_SERVER;
-  rpcInit.idleTime     = tsShellActivityTimer * 1500;
+  rpcInit.idleTime     = tsShellActivityTimer * 2000;
 
   tsDnodeMnodeRpc = rpcOpen(&rpcInit);
   if (tsDnodeMnodeRpc == NULL) {
-    dError("failed to init connection from mgmt");
+    dError("failed to init mnode rpc server");
     return -1;
   }
 
-  dPrint("connection to mgmt is opened");
+  dPrint("mnode rpc server is opened");
   return 0;
 }
 
 void dnodeCleanupMnode() {
   if (tsDnodeMnodeRpc) {
     rpcClose(tsDnodeMnodeRpc);
+    tsDnodeMnodeRpc = NULL;
+    dPrint("mnode rpc server is closed");
   }
 }
 
 static void dnodeProcessMsgFromMnode(SRpcMsg *pMsg) {
   SRpcMsg rspMsg;
-
-  rspMsg.handle = pMsg->handle;
-  rspMsg.pCont = NULL;
+  rspMsg.handle  = pMsg->handle;
+  rspMsg.pCont   = NULL;
   rspMsg.contLen = 0;
 
   if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) {
     rspMsg.code = TSDB_CODE_NOT_READY;
     rpcSendResponse(&rspMsg);
     rpcFreeCont(pMsg->pCont);
-    dTrace("conn:%p, query msg is ignored since dnode not running", pMsg->handle);
+    dTrace("thandle:%p, query msg is ignored since dnode not running", pMsg->handle);
     return;
   }
 

src/dnode/src/dnodeModule.c

@@ -74,7 +74,7 @@ int32_t dnodeInitModules() {
   for (int mod = 0; mod < TSDB_MOD_MAX; ++mod) {
     if (tsModule[mod].num != 0 && tsModule[mod].initFp) {
       if ((*tsModule[mod].initFp)() != 0) {
-        dError("TDengine initialization failed");
+        dError("failed to init modules");
         return -1;
       }
     }
@@ -84,11 +84,44 @@ int32_t dnodeInitModules() {
 }
 
 void dnodeStartModules() {
-  for (int mod = 1; mod < TSDB_MOD_MAX; ++mod) {
-    if (tsModule[mod].num != 0 && tsModule[mod].startFp) {
-      if ((*tsModule[mod].startFp)() != 0) {
-        dError("failed to start module:%d", mod);
+  // for (int mod = 1; mod < TSDB_MOD_MAX; ++mod) {
+  //   if (tsModule[mod].num != 0 && tsModule[mod].startFp) {
+  //     if ((*tsModule[mod].startFp)() != 0) {
+  //       dError("failed to start module:%d", mod);
+  //     }
+  //   }
+  // }
+}
+
+void dnodeProcessModuleStatus(uint32_t moduleStatus) {
+  if (moduleStatus == tsModuleStatus) return;
+
+  dPrint("module status is received, old:%d, new:%d", tsModuleStatus, moduleStatus);
+
+  int news = moduleStatus;
+  int olds = tsModuleStatus;
+
+  for (int moduleType = 0; moduleType < TSDB_MOD_MAX; ++moduleType) {
+    int newStatus = news & (1 << moduleType);
+    int oldStatus = olds & (1 << moduleType);
+
+    if (oldStatus > 0) {
+      if (newStatus == 0) {
+        if (tsModule[moduleType].stopFp) {
+          dPrint("module:%s is stopped on this node", tsModule[moduleType].name);
+          (*tsModule[moduleType].stopFp)();
+        }
       }
+    } else if (oldStatus == 0) {
+      if (newStatus > 0) {
+        if (tsModule[moduleType].startFp) {
+          dPrint("module:%s is started on this node", tsModule[moduleType].name);
+          (*tsModule[moduleType].startFp)();
+        }
+      }
+    } else {
     }
   }
+
+  tsModuleStatus = moduleStatus;
 }

src/dnode/src/dnodeShell.c

@@ -53,15 +53,15 @@ int32_t dnodeInitShell() {
   rpcInit.sessions     = TSDB_SESSIONS_PER_DNODE;
   rpcInit.connType     = TAOS_CONN_SERVER;
   rpcInit.idleTime     = tsShellActivityTimer * 1500;
-  rpcInit.afp          =
+  rpcInit.afp          = dnodeRetrieveUserAuthInfo;
 
   tsDnodeShellRpc = rpcOpen(&rpcInit);
   if (tsDnodeShellRpc == NULL) {
-    dError("failed to init connection from shell");
+    dError("failed to init shell rpc server");
     return -1;
   }
 
-  dPrint("connection to shell is opened");
+  dPrint("shell rpc server is opened");
   return 0;
 }
 

src/dnode/src/dnodeWrite.c

@@ -94,7 +94,7 @@ void dnodeWrite(SRpcMsg *pMsg) {
   SRpcContext *pRpcContext = NULL;
 
   if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT || pMsg->msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
-    SMsgDesc *pDesc = pCont;
+    SMsgDesc *pDesc = (SMsgDesc *)pCont;
     pDesc->numOfVnodes = htonl(pDesc->numOfVnodes);
     pCont += sizeof(SMsgDesc);
     if (pDesc->numOfVnodes > 1) {

src/inc/mnode.h

@@ -40,47 +40,62 @@ extern "C" {
 #include "tutil.h"
 
 typedef struct {
+  int32_t  mnodeId;
+  uint32_t privateIp;
+  uint32_t publicIp;
+  int64_t  createdTime;
+  int64_t  lostTime;
+  uint64_t dbVersion;
+  uint32_t rack;
+  uint16_t idc;
+  uint16_t slot;
+  int8_t   role;
+  int8_t   status;
+  int8_t   numOfMnodes;
+  int32_t  numOfDnodes;
+  char     mnodeName[TSDB_DNODE_NAME_LEN + 1];
+  char     reserved[7];
+  char     updateEnd[1];
+  int      syncFd;
+  void    *hbTimer;
+  void    *pSync;
+} SMnodeObj;
+
+typedef struct {
+  int32_t    dnodeId;
   uint32_t   privateIp;
-  int32_t    sid;
+  uint32_t   publicIp;
   uint32_t   moduleStatus;
-  int32_t    openVnodes;
-  int32_t    numOfVnodes;
-  int32_t    numOfFreeVnodes;
   int64_t    createdTime;
-  uint32_t   publicIp;
-  int32_t    status;
   uint32_t   lastAccess;
-  uint32_t   rebootTimes;
-  uint32_t   lastReboot;       // time stamp for last reboot
+  int32_t    openVnodes;
+  int32_t    numOfTotalVnodes; // from dnode status msg, config information
+  uint32_t   rack;
+  uint16_t   idc;
+  uint16_t   slot;
   uint16_t   numOfCores;       // from dnode status msg
-  uint8_t    alternativeRole;  // from dnode status msg, 0-any, 1-mgmt, 2-dnode
-  uint8_t    reserveStatus;
-  uint16_t   numOfTotalVnodes; // from dnode status msg, config information
-  uint16_t   unused;
+  int8_t     alternativeRole;  // from dnode status msg, 0-any, 1-mgmt, 2-dnode
+  int8_t     lbStatus;         // set in balance function
+  float      lbScore;          // calc in balance function
+  int32_t    customScore;      // config by user
+  char       dnodeName[TSDB_DNODE_NAME_LEN + 1];
+  char       reserved[7];
+  char       updateEnd[1];
+  SVnodeLoad vload[TSDB_MAX_VNODES];
+  int32_t    status;
+  uint32_t   lastReboot;       // time stamp for last reboot
   float      diskAvailable;    // from dnode status msg
-  int32_t    bandwidthMb;      // config by user
+  int16_t    diskAvgUsage;     // calc from sys.disk
   int16_t    cpuAvgUsage;      // calc from sys.cpu
   int16_t    memoryAvgUsage;   // calc from sys.mem
-  int16_t    diskAvgUsage;     // calc from sys.disk
   int16_t    bandwidthUsage;   // calc from sys.band
-  uint32_t   rack;
-  uint16_t   idc;
-  uint16_t   slot;
-  int32_t    customScore;     // config by user
-  float      lbScore;         // calc in balance function
-  int16_t    lbStatus;         // set in balance function
-  int16_t    lastAllocVnode;  // increase while create vnode
-  SVnodeLoad vload[TSDB_MAX_VNODES];
-  char       reserved[16];
-  char       updateEnd[1];
-  void *     thandle;
 } SDnodeObj;
 
 typedef struct {
   int32_t  dnodeId;
-  uint32_t ip;
-  uint32_t publicIp;
   int32_t  vnode;
+  uint32_t privateIp;
+  uint32_t publicIp;
 } SVnodeGid;
 
 typedef struct {
@@ -149,15 +164,13 @@ typedef struct _vg_obj {
   uint32_t        vgId;
   char            dbName[TSDB_DB_NAME_LEN + 1];
   int64_t         createdTime;
-  uint64_t        lastCreate;
-  uint64_t        lastRemove;
-  int32_t         numOfVnodes;
   SVnodeGid       vnodeGid[TSDB_VNODES_SUPPORT];
+  int32_t         numOfVnodes;
   int32_t         numOfTables;
   int32_t         lbIp;
   int32_t         lbTime;
   int8_t          lbStatus;
-  int8_t          reserved[16];
+  int8_t          reserved[14];
   int8_t          updateEnd[1];
   struct _vg_obj *prev, *next;
   void *          idPool;
@@ -169,8 +182,7 @@ typedef struct _db_obj {
   int8_t  dirty;
   int64_t createdTime;
   SDbCfg  cfg;
-  int8_t  dropStatus;
-  char    reserved[16];
+  char    reserved[15];
   char    updateEnd[1];
   struct _db_obj *prev, *next;
   int32_t numOfVgroups;

src/inc/taosdef.h

@@ -176,6 +176,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_MAX_COLUMNS          256
 #define TSDB_MIN_COLUMNS          2       //PRIMARY COLUMN(timestamp) + other columns
 
+#define TSDB_DNODE_NAME_LEN       63
 #define TSDB_TABLE_NAME_LEN       192
 #define TSDB_DB_NAME_LEN          32
 #define TSDB_COL_NAME_LEN         64

src/inc/taosmsg.h

@@ -25,7 +25,6 @@ extern "C" {
 
 #include "taosdef.h"
 #include "taoserror.h"
-#include "taosdef.h"
 #include "trpc.h"
 
 // message type
@@ -523,21 +522,21 @@ typedef struct {
 } SRetrieveTableRsp;
 
 typedef struct {
-  uint32_t vnode;
-  uint32_t vgId;
-  uint8_t  status;
-  uint8_t  dropStatus;
-  uint8_t  accessState;
-  int64_t  totalStorage;
-  int64_t  compStorage;
-  int64_t  pointsWritten;
-  uint8_t  syncStatus;
-  uint8_t  reserved[15];
+  int32_t vgId;
+  int32_t vnode;
+  int64_t totalStorage;
+  int64_t compStorage;
+  int64_t pointsWritten;
+  uint8_t status;
+  uint8_t syncStatus;
+  uint8_t accessState;
+  uint8_t reserved[5];
 } SVnodeLoad;
 
 typedef struct {
   uint32_t vnode;
-  char     accessState;
+  uint8_t  accessState;
+  uint8_t  reserved[3];
 } SVnodeAccess;
 
 /*
@@ -585,14 +584,16 @@ typedef struct {
 } SVnodeStatisticInfo;
 
 typedef struct {
+  int32_t  dnodeId;
   uint32_t moduleStatus;
   uint32_t createdTime;
   uint32_t numOfVnodes;
-  uint32_t reserved;
 } SDnodeState;
 
 typedef struct {
   uint32_t   version;
+  int32_t    dnodeId;
+  char       dnodeName[TSDB_DNODE_NAME_LEN];
   uint32_t   privateIp;
   uint32_t   publicIp;
   uint32_t   lastReboot;       // time stamp for last reboot
@@ -606,14 +607,12 @@ typedef struct {
 } SDMStatusMsg;
 
 typedef struct {
-  int32_t      code;
-  SDnodeState  dnodeState;
   SRpcIpSet    ipList;
+  SDnodeState  dnodeState;
   SVnodeAccess vnodeAccess[];
 } SDMStatusRsp;
 
 typedef struct {
-  int32_t    vnode;
   SVnodeCfg  cfg;
   SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
 } SMDCreateVnodeMsg;

src/mnode/inc/mgmtBalance.h

@@ -23,6 +23,7 @@ extern "C" {
 
 int32_t mgmtInitBalance();
 void    mgmtCleanupBalance();
+void    mgmtStartBalanceTimer(int32_t afterMs) ;
 int32_t mgmtAllocVnodes(SVgObj *pVgroup);
 
 #ifdef __cplusplus

src/mnode/inc/mgmtDServer.h

@@ -24,21 +24,6 @@ int32_t mgmtInitDServer();
 void    mgmtCleanupDServer();
 void    mgmtAddDServerMsgHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg));
 
-
-//extern void *mgmtStatusTimer;
-//
-//void mgmtSendCreateTableMsg(SMDCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle);
-//void mgmtSendDropTableMsg(SMDDropTableMsg *pRemove, SRpcIpSet *ipSet, void *ahandle);
-//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle);
-//void mgmtSendDropVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
-//
-//int32_t mgmtInitDnodeInt();
-//void    mgmtCleanUpDnodeInt();
-//
-//void mgmtSendMsgToDnode(SRpcIpSet *ipSet, int8_t msgType, void *pCont, int32_t contLen, void *ahandle);
-//void mgmtSendRspToDnode(void *pConn, int8_t msgType, int32_t code, void *pCont, int32_t contLen);
-//void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
-
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mgmtDnode.h

@@ -25,7 +25,8 @@ int32_t mgmtInitDnodes();
 void    mgmtCleanUpDnodes();
 int32_t mgmtGetDnodesNum();
 int32_t mgmtUpdateDnode(SDnodeObj *pDnode);
-SDnodeObj* mgmtGetDnode(uint32_t ip);
+SDnodeObj* mgmtGetDnode(int32_t dnodeId);
+SDnodeObj* mgmtGetDnodeByIp(uint32_t ip);
 
 bool mgmtCheckDnodeInRemoveState(SDnodeObj *pDnode);
 bool mgmtCheckDnodeInOfflineState(SDnodeObj *pDnode);

src/mnode/inc/mgmtMnode.h

@@ -22,6 +22,11 @@ extern "C" {
 
 bool mgmtCheckRedirect(void *handle);
 
+void mgmtGetMnodeIpList(SRpcIpSet *ipSet);
+
+int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp);
+int32_t mgmtRemoveMnode(uint32_t privateIp);
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/inc/mgmtVgroup.h

@@ -32,15 +32,15 @@ SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
 void    mgmtCreateVgroup(SQueuedMsg *pMsg);
 void    mgmtDropVgroup(SVgObj *pVgroup, void *ahandle);
 void    mgmtUpdateVgroup(SVgObj *pVgroup);
+void    mgmtUpdateVgroupIp(SDnodeObj *pDnode);
 
 void    mgmtSetVgroupIdPool();
 SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb);
 
 void    mgmtAddTableIntoVgroup(SVgObj *pVgroup, STableInfo *pTable);
 void    mgmtRemoveTableFromVgroup(SVgObj *pVgroup, STableInfo *pTable);
-
-SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode);
-void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
+void    mgmtSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle);
+void    mgmtSendDropVnodeMsg(int32_t vgId, SRpcIpSet *ipSet, void *ahandle);
 
 SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup);
 SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip);

src/mnode/src/mgmtAcct.c

@@ -24,7 +24,7 @@ void      (*mgmtCleanUpAcctsFp)() = NULL;
 SAcctObj *(*mgmtGetAcctFp)(char *acctName) = NULL;
 int32_t   (*mgmtCheckUserLimitFp)(SAcctObj *pAcct) = NULL;
 int32_t   (*mgmtCheckDbLimitFp)(SAcctObj *pAcct) = NULL;
-int32_t   (*mgmtCheckTimeSeriesLimitFp)(SAcctObj *pAcct, int32_t numOfTimeSeries) = NULL;
+int32_t   (*mgmtCheckTableLimitFp)(SAcctObj *pAcct, int32_t numOfTimeSeries) = NULL;
 
 int32_t mgmtAddDbIntoAcct(SAcctObj *pAcct, SDbObj *pDb) {
   pthread_mutex_lock(&pAcct->mutex);
@@ -137,8 +137,8 @@ int32_t mgmtCheckDbLimit(SAcctObj *pAcct) {
 }
 
 int32_t mgmtCheckTableLimit(SAcctObj *pAcct, int32_t numOfTimeSeries) {
-  if (mgmtCheckTimeSeriesLimitFp) {
-    return (*mgmtCheckTimeSeriesLimitFp)(pAcct, numOfTimeSeries);
+  if (mgmtCheckTableLimitFp) {
+    return (*mgmtCheckTableLimitFp)(pAcct, numOfTimeSeries);
   } else {
     return 0;
   }

src/mnode/src/mgmtBalance.c

@@ -20,6 +20,7 @@
 
 int32_t (*mgmtInitBalanceFp)() = NULL;
 void    (*mgmtCleanupBalanceFp)() = NULL;
+void    (*mgmtStartBalanceTimerFp)(int32_t afterMs) = NULL;
 int32_t (*mgmtAllocVnodesFp)(SVgObj *pVgroup) = NULL;
 
 int32_t mgmtInitBalance() {
@@ -36,33 +37,28 @@ void mgmtCleanupBalance() {
   }
 }
 
+void mgmtStartBalanceTimer(int32_t afterMs) {
+  if (mgmtStartBalanceTimerFp) {
+    (*mgmtStartBalanceTimerFp)(afterMs);
+  }
+}
+
 int32_t mgmtAllocVnodes(SVgObj *pVgroup) {
   if (mgmtAllocVnodesFp) {
-    return mgmtAllocVnodesFp(pVgroup);
+    return (*mgmtAllocVnodesFp)(pVgroup);
   }
 
-  SDnodeObj *pDnode = mgmtGetDnode(0);
+  SDnodeObj *pDnode = mgmtGetDnode(1);
   if (pDnode == NULL) return TSDB_CODE_OTHERS;
 
-  int32_t selectedVnode = -1;
-  int32_t lastAllocVode = pDnode->lastAllocVnode;
-
-  for (int32_t i = 0; i < pDnode->numOfVnodes; i++) {
-    int32_t vnode = (i + lastAllocVode) % pDnode->numOfVnodes;
-    if (pDnode->vload[vnode].vgId == 0 && pDnode->vload[vnode].status == TSDB_VN_STATUS_OFFLINE) {
-      selectedVnode = vnode;
-      break;
-    }
-  }
-
-  if (selectedVnode == -1) {
-    mError("alloc vnode failed, free vnodes:%d", pDnode->numOfFreeVnodes);
-    return -1;
+  if (pDnode->openVnodes < pDnode->numOfTotalVnodes) {
+    pVgroup->vnodeGid[0].dnodeId   = pDnode->dnodeId;
+    pVgroup->vnodeGid[0].privateIp = pDnode->privateIp;
+    pVgroup->vnodeGid[0].publicIp  = pDnode->publicIp;
+    mTrace("dnode:%d, alloc one vnode to vgroup", pDnode->dnodeId);
+    return TSDB_CODE_SUCCESS;
   } else {
-    mTrace("allocate vnode:%d, last allocated vnode:%d", selectedVnode, lastAllocVode);
-    pVgroup->vnodeGid[0].vnode = selectedVnode;
-    pDnode->lastAllocVnode     = selectedVnode + 1;
-    if (pDnode->lastAllocVnode >= pDnode->numOfVnodes) pDnode->lastAllocVnode = 0;
-    return 0;
+    mError("dnode:%d, failed to alloc vnode to vgroup", pDnode->dnodeId);
+    return TSDB_CODE_NO_ENOUGH_DNODES;
   }
 }

src/mnode/src/mgmtChildTable.c

@@ -33,7 +33,7 @@
 #include "mgmtTable.h"
 #include "mgmtVgroup.h"
 
-void *tsChildTableSdb;
+void   *tsChildTableSdb;
 int32_t tsChildTableUpdateSize;
 void *(*mgmtChildTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
 
@@ -50,12 +50,12 @@ static void mgmtDestroyChildTable(SChildTableObj *pTable) {
 }
 
 static void mgmtChildTableActionInit() {
-  mgmtChildTableActionFp[SDB_TYPE_INSERT] = mgmtChildTableActionInsert;
-  mgmtChildTableActionFp[SDB_TYPE_DELETE] = mgmtChildTableActionDelete;
-  mgmtChildTableActionFp[SDB_TYPE_UPDATE] = mgmtChildTableActionUpdate;
-  mgmtChildTableActionFp[SDB_TYPE_ENCODE] = mgmtChildTableActionEncode;
-  mgmtChildTableActionFp[SDB_TYPE_DECODE] = mgmtChildTableActionDecode;
-  mgmtChildTableActionFp[SDB_TYPE_RESET] = mgmtChildTableActionReset;
+  mgmtChildTableActionFp[SDB_TYPE_INSERT]  = mgmtChildTableActionInsert;
+  mgmtChildTableActionFp[SDB_TYPE_DELETE]  = mgmtChildTableActionDelete;
+  mgmtChildTableActionFp[SDB_TYPE_UPDATE]  = mgmtChildTableActionUpdate;
+  mgmtChildTableActionFp[SDB_TYPE_ENCODE]  = mgmtChildTableActionEncode;
+  mgmtChildTableActionFp[SDB_TYPE_DECODE]  = mgmtChildTableActionDecode;
+  mgmtChildTableActionFp[SDB_TYPE_RESET]   = mgmtChildTableActionReset;
   mgmtChildTableActionFp[SDB_TYPE_DESTROY] = mgmtChildTableActionDestroy;
 }
 
@@ -76,26 +76,26 @@ void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ss
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
-    mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
+    mError("ctable:%s, not in vgroup:%d", pTable->tableId, pTable->vgId);
     return NULL;
   }
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
-    mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
+    mError("ctable:%s, vgroup:%d not in db:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
     return NULL;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
-    mError("account not exists");
+    mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
     return NULL;
   }
 
   if (!sdbMaster) {
     int32_t sid = taosAllocateId(pVgroup->idPool);
     if (sid != pTable->sid) {
-      mError("sid:%d is not matched from the master:%d", sid, pTable->sid);
+      mError("ctable:%s, sid:%d is not matched from the master:%d", pTable->tableId, sid, pTable->sid);
       return NULL;
     }
   }
@@ -127,13 +127,13 @@ void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ss
 
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
-    mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
+    mError("ctable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
     return NULL;
   }
 
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
   if (pAcct == NULL) {
-    mError("account not exists");
+    mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
     return NULL;
   }
 
@@ -312,7 +312,7 @@ void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTab
 void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
   int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
   if (numOfTables >= tsMaxTables) {
-    mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
+    mError("ctable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
     terrno = TSDB_CODE_TOO_MANY_TABLES;
     return NULL;
   }
@@ -320,14 +320,14 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
   char *pTagData = (char *) pCreate->schema;  // it is a tag key
   SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData);
   if (pSuperTable == NULL) {
-    mError("table:%s, corresponding super table does not exist", pCreate->tableId);
+    mError("ctable:%s, corresponding super table does not exist", pCreate->tableId);
     terrno = TSDB_CODE_INVALID_TABLE;
     return NULL;
   }
 
   SChildTableObj *pTable = (SChildTableObj *) calloc(sizeof(SChildTableObj), 1);
   if (pTable == NULL) {
-    mError("table:%s, failed to alloc memory", pCreate->tableId);
+    mError("ctable:%s, failed to alloc memory", pCreate->tableId);
     terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
     return NULL;
   }
@@ -344,25 +344,25 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
 
   if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
     free(pTable);
-    mError("table:%s, update sdb error", pCreate->tableId);
+    mError("ctable:%s, update sdb error", pCreate->tableId);
     terrno = TSDB_CODE_SDB_ERROR;
     return NULL;
   }
 
-  mTrace("table:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
+  mTrace("ctable:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
   return pTable;
 }
 
 int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable) {
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
-    mError("table:%s, failed to drop child table, vgroup not exist", pTable->tableId);
+    mError("ctable:%s, failed to drop child table, vgroup not exist", pTable->tableId);
     return TSDB_CODE_OTHERS;
   }
 
   SMDDropTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropTableMsg));
   if (pDrop == NULL) {
-    mError("table:%s, failed to drop child table, no enough memory", pTable->tableId);
+    mError("ctable:%s, failed to drop child table, no enough memory", pTable->tableId);
     return TSDB_CODE_SERV_OUT_OF_MEMORY;
   }
 
@@ -374,7 +374,7 @@ int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable) {
 
   SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
 
-  mTrace("table:%s, send drop table msg", pDrop->tableId);
+  mTrace("ctable:%s, send drop table msg", pDrop->tableId);
   SRpcMsg rpcMsg = {
     .handle  = newMsg,
     .pCont   = pDrop,
@@ -394,6 +394,7 @@ void* mgmtGetChildTable(char *tableId) {
 }
 
 int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent) {
+// TODO: send message to dnode  
 //  int32_t col = mgmtFindSuperTableTagIndex(pTable->superTable, tagName);
 //  if (col < 0 || col > pTable->superTable->numOfTags) {
 //    return TSDB_CODE_APP_ERROR;
@@ -461,7 +462,7 @@ int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMetaMsg
     if (usePublicIp) {
       pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].publicIp;
     } else {
-      pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].ip;
+      pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].privateIp;
     }
     pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
   }

src/mnode/src/mgmtDClient.c

@@ -83,90 +83,3 @@ static void mgmtProcessRspFromDnode(SRpcMsg *rpcMsg) {
 
   rpcFreeCont(rpcMsg->pCont);
 }
-
-//static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
-//  mTrace("drop table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg) {
-//  mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//
-//static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
-//  mTrace("drop vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//static void mgmtProcessAlterVnodeRsp(SRpcMsg *rpcMsg) {
-//  mTrace("alter vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
-//  mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//static void mgmtProcessAlterStreamRsp(SRpcMsg *rpcMsg) {
-//  mTrace("alter stream rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-//static void mgmtProcessConfigDnodeRsp(SRpcMsg *rpcMsg) {
-//  mTrace("config dnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
-//}
-//
-
-
-//
-//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) {
-//  mTrace("table:%s, send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle);
-//}
-//
-//void mgmtSendDropVnodeMsg(int32_t vgId, int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
-//  mTrace("vnode:%d send free vnode msg, ahandle:%p", vnode, ahandle);
-//  SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
-//  SRpcMsg rpcMsg = {
-//      .handle  = ahandle,
-//      .pCont   = pDrop,
-//      .contLen = pDrop ? sizeof(SMDDropVnodeMsg) : 0,
-//      .code    = 0,
-//      .msgType = TSDB_MSG_TYPE_MD_DROP_VNODE
-//  };
-//  rpcSendRequest(tsMgmtDClientRpc, ipSet, &rpcMsg);
-//}
-//
-
-////
-////int32_t mgmtCfgDynamicOptions(SDnodeObj *pDnode, char *msg) {
-////  char *option, *value;
-////  int32_t   olen, valen;
-////
-////  paGetToken(msg, &option, &olen);
-////  if (strncasecmp(option, "unremove", 8) == 0) {
-////    mgmtSetDnodeUnRemove(pDnode);
-////    return TSDB_CODE_SUCCESS;
-////  } else if (strncasecmp(option, "score", 5) == 0) {
-////    paGetToken(option + olen + 1, &value, &valen);
-////    if (valen > 0) {
-////      int32_t score = atoi(value);
-////      mTrace("dnode:%s, custom score set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->customScore, score);
-////      pDnode->customScore = score;
-////      mgmtUpdateDnode(pDnode);
-////      //mgmtStartBalanceTimer(15);
-////    }
-////    return TSDB_CODE_INVALID_SQL;
-////  } else if (strncasecmp(option, "bandwidth", 9) == 0) {
-////    paGetToken(msg, &value, &valen);
-////    if (valen > 0) {
-////      int32_t bandwidthMb = atoi(value);
-////      if (bandwidthMb >= 0 && bandwidthMb < 10000000) {
-////        mTrace("dnode:%s, bandwidth(Mb) set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->bandwidthMb, bandwidthMb);
-////        pDnode->bandwidthMb = bandwidthMb;
-////        mgmtUpdateDnode(pDnode);
-////        return TSDB_CODE_SUCCESS;
-////      }
-////    }
-////    return TSDB_CODE_INVALID_SQL;
-////  }
-////
-////  return -1;
-////}
-////

src/mnode/src/mgmtDServer.c

@@ -85,276 +85,3 @@ static void mgmtProcessMsgFromDnode(SRpcMsg *rpcMsg) {
 static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
   return TSDB_CODE_SUCCESS;
 }
-
-//
-//
-//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
-//  SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
-//  pCfg->dnode = htonl(pCfg->dnode);
-//  pCfg->vnode = htonl(pCfg->vnode);
-//  pCfg->sid   = htonl(pCfg->sid);
-//  mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//
-//  if (!sdbMaster) {
-//    mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
-//    return;
-//  }
-//
-//  STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
-//  if (pTable == NULL) {
-//    mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
-//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
-//    return;
-//  }
-//
-//  mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
-//
-//  //TODO
-//  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
-//  mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
-//}
-//
-//static void mgmtProcessVnodeCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *pConn) {
-//  if (!sdbMaster) {
-//    mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
-//    return;
-//  }
-//
-//  SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) pCont;
-//  pCfg->dnode = htonl(pCfg->dnode);
-//  pCfg->vnode = htonl(pCfg->vnode);
-//
-//  SVgObj *pVgroup = mgmtGetVgroupByVnode(pCfg->dnode, pCfg->vnode);
-//  if (pVgroup == NULL) {
-//    mTrace("dnode:%s, vnode:%d, no vgroup info", taosIpStr(pCfg->dnode), pCfg->vnode);
-//    mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_NOT_ACTIVE_VNODE, NULL, 0);
-//    return;
-//  }
-//
-//  mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
-//
-//  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
-//  mgmtSendCreateVnodeMsg(pVgroup, pCfg->vnode, &ipSet, NULL);
-//}
-//
-//static void mgmtProcessCreateTableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
-//  mTrace("create table rsp received, thandle:%p code:%d", thandle, code);
-//  if (thandle == NULL) return;
-//
-//  SProcessInfo *info = thandle;
-//  assert(info->type == TSDB_PROCESS_CREATE_TABLE || info->type == TSDB_PROCESS_CREATE_TABLE_GET_META);
-//  STableInfo *pTable = info->ahandle;
-//
-//  if (code != TSDB_CODE_SUCCESS) {
-//    mError("table:%s, failed to create in dnode, code:%d, set it dirty", pTable->tableId);
-//    mgmtSetTableDirty(pTable, true);
-//  } else {
-//    mTrace("table:%s, created in dnode", pTable->tableId);
-//    mgmtSetTableDirty(pTable, false);
-//  }
-//
-//  if (code != TSDB_CODE_SUCCESS) {
-//    SRpcMsg rpcMsg = {0};
-//    rpcMsg.code = code;
-//    rpcMsg.handle = info->thandle;
-//    rpcSendResponse(&rpcMsg);
-//  } else {
-//    if (info->type == TSDB_PROCESS_CREATE_TABLE_GET_META) {
-//      mTrace("table:%s, start to process get meta", pTable->tableId);
-//      mgmtProcessGetTableMeta(pTable, thandle);
-//    } else {
-//      SRpcMsg rpcMsg = {0};
-//      rpcMsg.code = code;
-//      rpcMsg.handle = info->thandle;
-//      rpcSendResponse(&rpcMsg);
-//    }
-//  }
-//
-//  free(info);
-//}
-//
-
-//static void mgmtProcessRemoveTableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
-//  mTrace("remove table rsp received, thandle:%p code:%d", thandle, code);
-//}
-//
-
-//
-//static void mgmtProcessDropVnodeRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
-//  mTrace("free vnode rsp received, thandle:%p code:%d", thandle, code);
-//}
-//
-//static void mgmtProcessDropStableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
-//  mTrace("drop stable rsp received, thandle:%p code:%d", thandle, code);
-//}
-//
-//static void mgmtProcessCreateVnodeRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
-//  mTrace("create vnode rsp received, thandle:%p code:%d", thandle, code);
-//  if (thandle == NULL) return;
-//
-//  SProcessInfo *info = thandle;
-//  assert(info->type == TSDB_PROCESS_CREATE_VGROUP || info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META);
-//  info->received++;
-//  SVgObj *pVgroup = info->ahandle;
-//
-//  bool isGetMeta = false;
-//  if (info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META) {
-//    isGetMeta = true;
-//  }
-//
-//  mTrace("vgroup:%d, received:%d numOfVnodes:%d", pVgroup->vgId, info->received, pVgroup->numOfVnodes);
-//  if (info->received == pVgroup->numOfVnodes) {
-//    mgmtProcessCreateTable(pVgroup, info->cont, info->contLen, info->thandle, isGetMeta);
-//    free(info);
-//  }
-//}
-//
-//void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
-//  if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) {
-//    mError("invalid msg type:%d", msgType);
-//    return;
-//  }
-//
-//  mTrace("msg:%d:%s is received from dnode, pConn:%p", msgType, taosMsg[(int8_t)msgType], pConn);
-//
-//  if (msgType == TSDB_MSG_TYPE_DM_CONFIG_TABLE) {
-//    mgmtProcessTableCfgMsg(msgType, pCont, contLen, pConn);
-//  } else if (msgType == TSDB_MSG_TYPE_DM_CONFIG_VNODE) {
-//    mgmtProcessVnodeCfgMsg(msgType, pCont, contLen, pConn);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP) {
-//    mgmtProcessCreateTableRsp(msgType, pCont, contLen, pConn, code);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_DROP_TABLE_RSP) {
-//    mgmtProcessRemoveTableRsp(msgType, pCont, contLen, pConn, code);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP) {
-//    mgmtProcessCreateVnodeRsp(msgType, pCont, contLen, pConn, code);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_DROP_VNODE_RSP) {
-//    mgmtProcessDropVnodeRsp(msgType, pCont, contLen, pConn, code);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
-//    mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code);
-//  } else if (msgType == TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP) {
-//  } else if (msgType == TSDB_MSG_TYPE_CM_ALTER_STREAM_RSP) {
-//  } else if (msgType == TSDB_MSG_TYPE_DM_STATUS) {
-//    mgmtProcessDnodeStatus(msgType, pCont, contLen, pConn, code);
-//  }  else {
-//    mError("%s from dnode is not processed", taosMsg[(int8_t)msgType]);
-//  }
-//
-//  //rpcFreeCont(pCont);
-//}
-//
-//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) {
-//  mTrace("table:%s, sid:%d send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle);
-//}
-//
-//void mgmtSendDropVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
-//  mTrace("vnode:%d send free vnode msg, ahandle:%p", vnode, ahandle);
-//
-//  SMDDropVnodeMsg *pFreeVnode = rpcMallocCont(sizeof(SMDDropVnodeMsg));
-//  if (pFreeVnode != NULL) {
-//    pFreeVnode->vnode = htonl(vnode);
-//    mgmtSendMsgToDnode(ipSet, TSDB_MSG_TYPE_MD_DROP_VNODE, pFreeVnode, sizeof(SMDDropVnodeMsg), ahandle);
-//  }
-//}
-//
-
-//int32_t mgmtCfgDynamicOptions(SDnodeObj *pDnode, char *msg) {
-//  char *option, *value;
-//  int32_t   olen, valen;
-//
-//  paGetToken(msg, &option, &olen);
-//  if (strncasecmp(option, "unremove", 8) == 0) {
-//    mgmtSetDnodeUnRemove(pDnode);
-//    return TSDB_CODE_SUCCESS;
-//  } else if (strncasecmp(option, "score", 5) == 0) {
-//    paGetToken(option + olen + 1, &value, &valen);
-//    if (valen > 0) {
-//      int32_t score = atoi(value);
-//      mTrace("dnode:%s, custom score set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->customScore, score);
-//      pDnode->customScore = score;
-//      mgmtUpdateDnode(pDnode);
-//      //mgmtStartBalanceTimer(15);
-//    }
-//    return TSDB_CODE_INVALID_SQL;
-//  } else if (strncasecmp(option, "bandwidth", 9) == 0) {
-//    paGetToken(msg, &value, &valen);
-//    if (valen > 0) {
-//      int32_t bandwidthMb = atoi(value);
-//      if (bandwidthMb >= 0 && bandwidthMb < 10000000) {
-//        mTrace("dnode:%s, bandwidth(Mb) set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->bandwidthMb, bandwidthMb);
-//        pDnode->bandwidthMb = bandwidthMb;
-//        mgmtUpdateDnode(pDnode);
-//        return TSDB_CODE_SUCCESS;
-//      }
-//    }
-//    return TSDB_CODE_INVALID_SQL;
-//  }
-//
-//  return -1;
-//}
-//
-//
-//void mgmtCleanUpDnodeInt() {
-//  if (mgmtCleanUpDnodeIntFp) {
-//    mgmtCleanUpDnodeIntFp();
-//  }
-//}
-//
-//void mgmtProcessDnodeStatus(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
-//  SDMStatusMsg *pStatus = (SDMStatusMsg *)pCont;
-//
-//  SDnodeObj *pObj = mgmtGetDnode(htonl(pStatus->privateIp));
-//  if (pObj == NULL) {
-//    mError("dnode:%s not exist", taosIpStr(pObj->privateIp));
-//    mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_DNODE_NOT_EXIST, NULL, 0);
-//    return;
-//  }
-//
-//  pObj->lastReboot       = htonl(pStatus->lastReboot);
-//  pObj->numOfTotalVnodes = htons(pStatus->numOfTotalVnodes);
-//  pObj->openVnodes       = htons(pStatus->openVnodes);
-//  pObj->numOfCores       = htons(pStatus->numOfCores);
-//  pObj->diskAvailable = pStatus->diskAvailable;
-//  pObj->alternativeRole  = pStatus->alternativeRole;
-////
-////  if (mgmtProcessDnodeStatusFp) {
-////    mgmtProcessDnodeStatusFp(pStatus, pObj, pConn);
-////    return;
-////  }
-//
-//  pObj->status = TSDB_DN_STATUS_READY;
-//
-////  // wait vnode dropped
-////  for (int32_t vnode = 0; vnode < pObj->numOfVnodes; ++vnode) {
-////    SVnodeLoad *pVload = &(pObj->vload[vnode]);
-////    if (pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) {
-////      bool existInDnode = false;
-////      for (int32_t j = 0; j < pObj->openVnodes; ++j) {
-////        if (htonl(pStatus->load[j].vnode) == vnode) {
-////          existInDnode = true;
-////          break;
-////        }
-////      }
-////
-////      if (!existInDnode) {
-////        pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
-////        pVload->status = TSDB_VN_STATUS_OFFLINE;
-////        mgmtUpdateDnode(pObj);
-////        mPrint("dnode:%s, vid:%d, drop finished", taosIpStr(pObj->privateIp), vnode);
-////        taosTmrStart(mgmtMonitorDbDrop, 10000, NULL, tsMgmtTmr);
-////      }
-////    } else if (pVload->vgId == 0) {
-////      /*
-////       * In some cases, vnode information may be reported abnormally, recover it
-////       */
-////      if (pVload->dropStatus != TSDB_VN_DROP_STATUS_READY || pVload->status != TSDB_VN_STATUS_OFFLINE) {
-////        mPrint("dnode:%s, vid:%d, vgroup:%d status:%s dropStatus:%s, set it to avail status",
-////               taosIpStr(pObj->privateIp), vnode, pVload->vgId, taosGetVnodeStatusStr(pVload->status),
-////               taosGetVnodeDropStatusStr(pVload->dropStatus));
-////        pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
-////        pVload->status = TSDB_VN_STATUS_OFFLINE;
-////        mgmtUpdateDnode(pObj);
-////      }
-////    }
-////  }
-//}

src/mnode/src/mgmtDb.c

@@ -36,7 +36,7 @@
 #include "tutil.h"
 #include "name.h"
 
-static void *tsDbSdb = NULL;
+static void   *tsDbSdb = NULL;
 static int32_t tsDbUpdateSize;
 
 static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate);
@@ -681,7 +681,7 @@ static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    strcpy(pWrite, pDb->dropStatus != TSDB_DB_STATUS_READY ? "dropping" : "ready");
+    strcpy(pWrite, pDb->dirty != TSDB_DB_STATUS_READY ? "dropping" : "ready");
     cols++;
 
     numOfRows++;

src/mnode/src/mgmtDnode.c

@@ -22,12 +22,14 @@
 #include "mgmtDClient.h"
 #include "mgmtMnode.h"
 #include "mgmtShell.h"
+#include "mgmtDServer.h"
 #include "mgmtUser.h"
 #include "mgmtVgroup.h"
 
 int32_t    (*mgmtInitDnodesFp)() = NULL;
 void       (*mgmtCleanUpDnodesFp)() = NULL;
 SDnodeObj *(*mgmtGetDnodeFp)(uint32_t ip) = NULL;
+SDnodeObj *(*mgmtGetDnodeByIpFp)(int32_t dnodeId) = NULL;
 int32_t    (*mgmtGetDnodesNumFp)() = NULL;
 int32_t    (*mgmtUpdateDnodeFp)(SDnodeObj *pDnode) = NULL;
 void *     (*mgmtGetNextDnodeFp)(SShowObj *pShow, SDnodeObj **pDnode) = NULL;
@@ -43,80 +45,29 @@ static int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, vo
 static int32_t mgmtGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static void    mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg);
+static void    mgmtProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) ;
+static void    mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg);
 
 void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) {
   int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore;
+
   maxVnodes = maxVnodes > TSDB_MAX_VNODES ? TSDB_MAX_VNODES : maxVnodes;
   maxVnodes = maxVnodes < TSDB_MIN_VNODES ? TSDB_MIN_VNODES : maxVnodes;
-  if (pDnode->numOfTotalVnodes != 0) {
-    maxVnodes = pDnode->numOfTotalVnodes;
+  
+  if (pDnode->numOfTotalVnodes == 0) {
+    pDnode->numOfTotalVnodes = maxVnodes;
   }
+
   if (pDnode->alternativeRole == TSDB_DNODE_ROLE_MGMT) {
-    maxVnodes = 0;
+    pDnode->numOfTotalVnodes = 0;
   }
 
-  pDnode->numOfVnodes = maxVnodes;
-  pDnode->numOfFreeVnodes = maxVnodes;
   pDnode->openVnodes = 0;
   pDnode->status = TSDB_DN_STATUS_OFFLINE;
-}
-
-void mgmtCalcNumOfFreeVnodes(SDnodeObj *pDnode) {
-  int32_t totalVnodes = 0;
-
-  mTrace("dnode:%s, begin calc free vnodes", taosIpStr(pDnode->privateIp));
-  for (int32_t i = 0; i < pDnode->numOfVnodes; ++i) {
-    SVnodeLoad *pVload = pDnode->vload + i;
-    if (pVload->vgId != 0) {
-      mTrace("%d-dnode:%s, calc free vnodes, exist vnode:%d, vgroup:%d, state:%d %s, dropstate:%d %s, syncstatus:%d %s",
-             totalVnodes, taosIpStr(pDnode->privateIp), i, pVload->vgId,
-             pVload->status, taosGetVnodeStatusStr(pVload->status),
-             pVload->dropStatus, taosGetVnodeDropStatusStr(pVload->dropStatus),
-             pVload->syncStatus, taosGetVnodeSyncStatusStr(pVload->syncStatus));
-      totalVnodes++;
-    }
-  }
-
-  pDnode->numOfFreeVnodes = pDnode->numOfVnodes - totalVnodes;
-  mTrace("dnode:%s, numOfVnodes:%d, numOfFreeVnodes:%d, totalVnodes:%d",
-          taosIpStr(pDnode->privateIp), pDnode->numOfVnodes, pDnode->numOfFreeVnodes, totalVnodes);
-}
 
-void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId) {
-  SDnodeObj *pDnode;
-
-  for (int32_t i = 0; i < numOfVnodes; ++i) {
-    pDnode = mgmtGetDnode(vnodeGid[i].ip);
-    if (pDnode) {
-      SVnodeLoad *pVload = pDnode->vload + vnodeGid[i].vnode;
-      memset(pVload, 0, sizeof(SVnodeLoad));
-      pVload->vnode = vnodeGid[i].vnode;
-      pVload->vgId = vgId;
-      mTrace("dnode:%s, vnode:%d add to vgroup:%d", taosIpStr(pDnode->privateIp), vnodeGid[i].vnode, pVload->vgId);
-      mgmtCalcNumOfFreeVnodes(pDnode);
-    } else {
-      mError("dnode:%s, not in dnode DB!!!", taosIpStr(vnodeGid[i].ip));
-    }
-  }
-}
-
-void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes) {
-  SDnodeObj *pDnode;
-
-  for (int32_t i = 0; i < numOfVnodes; ++i) {
-    pDnode = mgmtGetDnode(vnodeGid[i].ip);
-    if (pDnode) {
-      SVnodeLoad *pVload = pDnode->vload + vnodeGid[i].vnode;
-      mTrace("dnode:%s, vnode:%d remove from vgroup:%d", taosIpStr(vnodeGid[i].ip), vnodeGid[i].vnode, pVload->vgId);
-      memset(pVload, 0, sizeof(SVnodeLoad));
-      mgmtCalcNumOfFreeVnodes(pDnode);
-    } else {
-      mError("dnode:%s not in dnode DB!!!", taosIpStr(vnodeGid[i].ip));
-    }
-  }
+  mgmtUpdateDnode(pDnode);
 }
 
-
 bool mgmtCheckModuleInDnode(SDnodeObj *pDnode, int32_t moduleType) {
   uint32_t status = pDnode->moduleStatus & (1 << moduleType);
   return status > 0;
@@ -317,12 +268,6 @@ static int32_t mgmtGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
-  pShow->bytes[cols] = 4;
-  pSchema[cols].type = TSDB_DATA_TYPE_INT;
-  strcpy(pSchema[cols].name, "vgid");
-  pSchema[cols].bytes = htons(pShow->bytes[cols]);
-  cols++;
-
   pShow->bytes[cols] = 12;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
   strcpy(pSchema[cols].name, "status");
@@ -341,11 +286,10 @@ static int32_t mgmtGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
   pShow->offset[0] = 0;
   for (int32_t i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
 
-  // TODO: if other thread drop dnode ????
   SDnodeObj *pDnode = NULL;
   if (pShow->payloadLen > 0 ) {
     uint32_t ip = ip2uint(pShow->payload);
-    pDnode = mgmtGetDnode(ip);
+    pDnode = mgmtGetDnodeByIp(ip);
     if (NULL == pDnode) {
       return TSDB_CODE_NODE_OFFLINE;
     }
@@ -398,10 +342,6 @@ static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, voi
         
         cols = 0;
         
-        pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-        *(uint32_t *)pWrite = pVnode->vnode;
-        cols++;
-        
         pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
         *(uint32_t *)pWrite = pVnode->vgId;
         cols++;
@@ -434,19 +374,22 @@ int32_t mgmtInitDnodes() {
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VNODES, mgmtGetVnodeMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VNODES, mgmtRetrieveVnodes);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONFIG_DNODE, mgmtProcessCfgDnodeMsg);
+  mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP, mgmtProcessCfgDnodeMsgRsp);
+  mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_STATUS, mgmtProcessDnodeStatusMsg);
 
   if (mgmtInitDnodesFp) {
     return mgmtInitDnodesFp();
   } else {
-    tsDnodeObj.privateIp        = inet_addr(tsPrivateIp);;
+    tsDnodeObj.dnodeId          = 1;
+    tsDnodeObj.privateIp        = inet_addr(tsPrivateIp);
+    tsDnodeObj.publicIp         = inet_addr(tsPublicIp);
     tsDnodeObj.createdTime      = taosGetTimestampMs();
-    tsDnodeObj.lastReboot       = taosGetTimestampSec();
+    tsDnodeObj.numOfTotalVnodes = tsNumOfTotalVnodes;
     tsDnodeObj.numOfCores       = (uint16_t) tsNumOfCores;
-    tsDnodeObj.status           = TSDB_DN_STATUS_READY;
     tsDnodeObj.alternativeRole  = TSDB_DNODE_ROLE_ANY;
-    tsDnodeObj.numOfTotalVnodes = tsNumOfTotalVnodes;
-    tsDnodeObj.thandle          = (void *) (1);  //hack way
-    tsDnodeObj.status           = TSDB_DN_STATUS_READY;
+    tsDnodeObj.status           = TSDB_DN_STATUS_OFFLINE;
+    tsDnodeObj.lastReboot       = taosGetTimestampSec();
+    sprintf(tsDnodeObj.dnodeName, "%d", tsDnodeObj.dnodeId);
     mgmtSetDnodeMaxVnodes(&tsDnodeObj);
 
     tsDnodeObj.moduleStatus |= (1 << TSDB_MOD_MGMT);
@@ -462,21 +405,30 @@ int32_t mgmtInitDnodes() {
 
 void mgmtCleanUpDnodes() {
   if (mgmtCleanUpDnodesFp) {
-    mgmtCleanUpDnodesFp();
+    (*mgmtCleanUpDnodesFp)();
   }
 }
 
-SDnodeObj *mgmtGetDnode(uint32_t ip) {
+SDnodeObj *mgmtGetDnode(int32_t dnodeId) {
   if (mgmtGetDnodeFp) {
-    return mgmtGetDnodeFp(ip);
-  } else {
+    return (*mgmtGetDnodeFp)(dnodeId);
+  } 
+  if (dnodeId == 1) {
     return &tsDnodeObj;
   }
+  return NULL;
+}
+
+SDnodeObj *mgmtGetDnodeByIp(uint32_t ip) {
+  if (mgmtGetDnodeByIpFp) {
+    return (*mgmtGetDnodeByIpFp)(ip);
+  } 
+  return &tsDnodeObj;
 }
 
 int32_t mgmtGetDnodesNum() {
   if (mgmtGetDnodesNumFp) {
-    return mgmtGetDnodesNumFp();
+    return (*mgmtGetDnodesNumFp)();
   } else {
     return 1;
   }
@@ -484,7 +436,7 @@ int32_t mgmtGetDnodesNum() {
 
 int32_t mgmtUpdateDnode(SDnodeObj *pDnode) {
   if (mgmtUpdateDnodeFp) {
-    return mgmtUpdateDnodeFp(pDnode);
+    return (*mgmtUpdateDnodeFp)(pDnode);
   } else {
     return 0;
   }
@@ -492,7 +444,7 @@ int32_t mgmtUpdateDnode(SDnodeObj *pDnode) {
 
 void *mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode) {
   if (mgmtGetNextDnodeFp) {
-    return mgmtGetNextDnodeFp(pShow, pDnode);
+    return (*mgmtGetNextDnodeFp)(pShow, pDnode);
   } else {
     if (*pDnode == NULL) {
       *pDnode = &tsDnodeObj;
@@ -506,14 +458,12 @@ void *mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode) {
 
 void mgmtSetDnodeUnRemove(SDnodeObj *pDnode) {
   if (mgmtSetDnodeUnRemoveFp) {
-    mgmtSetDnodeUnRemoveFp(pDnode);
+    (*mgmtSetDnodeUnRemoveFp)(pDnode);
   }
 }
 
 bool mgmtCheckConfigShow(SGlobalConfig *cfg) {
-  if (cfg->cfgType & TSDB_CFG_CTYPE_B_CLUSTER)
-    return false;
-  if (cfg->cfgType & TSDB_CFG_CTYPE_B_NOT_PRINT)
+  if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_SHOW))
     return false;
   return true;
 }
@@ -531,6 +481,11 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
   if (mgmtCheckRedirect(pMsg->thandle)) return;
 
   SCMCfgDnodeMsg *pCmCfgDnode = pMsg->pCont;
+  if (pCmCfgDnode->ip[0] == 0) {
+    strcpy(pCmCfgDnode->ip, tsPrivateIp);
+  } else {
+    strcpy(pCmCfgDnode->ip, pCmCfgDnode->ip);
+  }
   uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip);
 
   if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
@@ -557,3 +512,107 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
 
   rpcSendResponse(&rpcRsp);
 }
+
+static void mgmtProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) {
+  mTrace("cfg vnode rsp is received");
+}
+
+void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
+  if (mgmtCheckRedirect(rpcMsg->handle)) return;
+
+  SDMStatusMsg *pStatus = rpcMsg->pCont;
+  pStatus->dnodeId = htonl(pStatus->dnodeId);
+
+  SDnodeObj *pDnode = NULL;
+  if (pStatus->dnodeId == 0) {
+    pDnode = mgmtGetDnodeByIp(htonl(pStatus->privateIp));
+    if (pDnode == NULL) {
+      mTrace("dnode not created, privateIp:%s", taosIpStr(htonl(pStatus->privateIp)));
+      mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_DNODE_NOT_EXIST);
+      return;
+    }
+  } else {
+    pDnode = mgmtGetDnode(pStatus->dnodeId);
+    if (pDnode == NULL) {
+      mError("dnode:%d, not exist, privateIp:%s", taosIpStr(pStatus->dnodeId), pStatus->dnodeName);
+      mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_DNODE_NOT_EXIST);
+      return;
+    }
+  }
+
+  uint32_t version = htonl(pStatus->version);
+  if (version != tsVersion) {
+    mError("dnode:%d, status msg version:%d not equal with mnode:%d", pDnode->dnodeId, version, tsVersion);
+    mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_INVALID_MSG_VERSION);
+    return ;
+  }
+  
+  uint32_t lastPrivateIp = pDnode->privateIp;
+  uint32_t lastPublicIp  = pDnode->publicIp;
+
+  pDnode->privateIp        = htonl(pStatus->privateIp);
+  pDnode->publicIp         = htonl(pStatus->publicIp);
+  pDnode->lastReboot       = htonl(pStatus->lastReboot);
+  pDnode->numOfTotalVnodes = htons(pStatus->numOfTotalVnodes);
+  pDnode->numOfCores       = htons(pStatus->numOfCores);
+  pDnode->diskAvailable    = pStatus->diskAvailable;
+  pDnode->alternativeRole  = pStatus->alternativeRole;
+
+  if (pStatus->dnodeId == 0) {
+    mTrace("dnode:%d, first access, privateIp:%s, name:%s, ", pDnode->dnodeId, taosIpStr(pDnode->privateIp), pDnode->dnodeName);
+    mgmtSetDnodeMaxVnodes(pDnode);
+  }
+ 
+  if (lastPrivateIp != pDnode->privateIp || lastPublicIp != pDnode->publicIp) {
+    mgmtUpdateVgroupIp(pDnode);
+    //mgmtUpdateMnodeIp();
+  }
+
+  int32_t openVnodes = htons(pStatus->openVnodes);
+  for (int32_t j = 0; j < openVnodes; ++j) {
+    pDnode->vload[j].vgId          = htonl(pStatus->load[j].vgId);
+    pDnode->vload[j].totalStorage  = htobe64(pStatus->load[j].totalStorage);
+    pDnode->vload[j].compStorage   = htobe64(pStatus->load[j].compStorage);
+    pDnode->vload[j].pointsWritten = htobe64(pStatus->load[j].pointsWritten);
+    
+    SVgObj *pVgroup = mgmtGetVgroup(pDnode->vload[j].vgId);
+    if (pVgroup == NULL) {
+      SRpcIpSet ipSet = mgmtGetIpSetFromIp(pDnode->privateIp);
+      mPrint("dnode:%d, vnode:%d not exist in mnode, drop it", pDnode->dnodeId, pDnode->vload[j].vgId);
+      mgmtSendDropVnodeMsg(pDnode->vload[j].vgId, &ipSet, NULL);
+    }
+  }
+
+  if (pDnode->status != TSDB_DN_STATUS_READY) {
+    mTrace("dnode:%d, from offline to online", pDnode->dnodeId);
+    pDnode->status = TSDB_DN_STATUS_READY;
+    mgmtStartBalanceTimer(200);
+  }
+
+  int32_t contLen = sizeof(SDMStatusRsp) + TSDB_MAX_VNODES * sizeof(SVnodeAccess);
+  SDMStatusRsp *pRsp = rpcMallocCont(contLen);
+  if (pRsp == NULL) {
+    mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_SERV_OUT_OF_MEMORY);
+    return;
+  }
+
+  mgmtGetMnodeIpList(&pRsp->ipList);
+
+  pRsp->dnodeState.dnodeId = htonl(pDnode->dnodeId);
+  pRsp->dnodeState.moduleStatus = htonl(pDnode->moduleStatus);
+  pRsp->dnodeState.createdTime  = htonl(pDnode->createdTime / 1000);
+  pRsp->dnodeState.numOfVnodes = 0;
+  
+  contLen = sizeof(SDMStatusRsp);
+
+  //TODO: set vnode access
+  
+  SRpcMsg rpcRsp = {
+    .handle  = rpcMsg->handle,
+    .code    = TSDB_CODE_SUCCESS,
+    .pCont   = pRsp,
+    .contLen = contLen
+  };
+
+  rpcSendResponse(&rpcRsp);
+}

src/mnode/src/mgmtMnode.c

@@ -15,8 +15,162 @@
 
 #define _DEFAULT_SOURCE
 #include "os.h"
+#include "trpc.h"
+#include "tschemautil.h"
 #include "mgmtMnode.h"
+#include "mgmtUser.h"
+
+int32_t (*mgmtAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
+int32_t (*mgmtRemoveMnodeFp)(uint32_t privateIp) = NULL;
+int32_t (*mgmtGetMnodesNumFp)() = NULL;
+void *  (*mgmtGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
+
+static SMnodeObj tsMnodeObj = {0};
+static int32_t mgmtGetMnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
+static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
 bool mgmtCheckRedirect(void *handle) {
   return false;
+}
+
+int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp) {
+  if (mgmtAddMnodeFp) {
+    return (*mgmtAddMnodeFp)(privateIp, publicIp);
+  } else {
+    return 0;
+  }
+}
+
+int32_t mgmtRemoveMnode(uint32_t privateIp) {
+  if (mgmtRemoveMnodeFp) {
+    return (*mgmtRemoveMnodeFp)(privateIp);
+  } else {
+    return 0;
+  }
+}
+
+static int32_t mgmtGetMnodesNum() {
+  if (mgmtGetMnodesNumFp) {
+    return (*mgmtGetMnodesNumFp)();
+  } else {
+    return 1;
+  }
+}
+
+static void *mgmtGetNextMnode(SShowObj *pShow, SMnodeObj **pMnode) {
+  if (mgmtGetNextMnodeFp) {
+    return (*mgmtGetNextMnodeFp)(pShow, pMnode);
+  } else {
+    if (*pMnode == NULL) {
+      *pMnode = &tsMnodeObj;
+    } else {
+      *pMnode = NULL;
+    }
+  }
+
+  return *pMnode;
+}
+
+static int32_t mgmtGetMnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
+  int32_t cols = 0;
+
+  SUserObj *pUser = mgmtGetUserFromConn(pConn);
+  if (pUser == NULL) return 0;
+
+  if (strcmp(pUser->user, "root") != 0) return TSDB_CODE_NO_RIGHTS;
+
+  SSchema *pSchema = tsGetSchema(pMeta);
+
+  pShow->bytes[cols] = 16;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "private ip");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 8;
+  pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
+  strcpy(pSchema[cols].name, "create time");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 10;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "status");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 10;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "role");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 16;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "public ip");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pMeta->numOfColumns = htons(cols);
+  pShow->numOfColumns = cols;
+
+  pShow->offset[0] = 0;
+  for (int32_t i = 1; i < cols; ++i) {
+    pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
+  }
+
+  pShow->numOfRows = mgmtGetMnodesNum();
+  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
+  pShow->pNode = NULL;
+
+  return 0;
+}
+
+static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
+  int32_t  numOfRows = 0;
+  int32_t  cols      = 0;
+  SMnodeObj *pMnode   = NULL;
+  char     *pWrite;
+  char     ipstr[32];
+
+  while (numOfRows < rows) {
+    pShow->pNode = mgmtGetNextMnode(pShow, (SMnodeObj **)&pMnode);
+    if (pMnode == NULL) break;
+
+    cols = 0;
+
+    tinet_ntoa(ipstr, pMnode->privateIp);
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    strcpy(pWrite, ipstr);
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int64_t *)pWrite = pMnode->createdTime;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    strcpy(pWrite, sdbStatusStr[(uint8_t)pMnode->status]);
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    strcpy(pWrite, sdbRoleStr[(uint8_t)pMnode->role]);
+    cols++;
+
+    tinet_ntoa(ipstr, pMnode->publicIp);
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    strcpy(pWrite, ipstr);
+    cols++;
+
+    numOfRows++;
+  }
+
+  pShow->numOfReads += numOfRows;
+  return numOfRows;
+}
+
+void mgmtGetMnodeIpList(SRpcIpSet *ipSet) {
+  ipSet->inUse = 0;
+  ipSet->port = htons(tsMnodeDnodePort);
+  ipSet->numOfIps = 1;
+  ipSet->ip[0] = htonl(inet_addr(tsMasterIp));
 }
\ No newline at end of file

src/mnode/src/mgmtNormalTable.c

@@ -542,7 +542,7 @@ int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMetaM
     if (usePublicIp) {
       pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].publicIp;
     } else {
-      pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].ip;
+      pMeta->vpeerDesc[i].ip    = pVgroup->vnodeGid[i].privateIp;
     }
     
     pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);

src/mnode/src/mgmtShell.c

@@ -392,7 +392,7 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
   }
 
   if (pConnectMsg->db[0]) {
-    char dbName[TSDB_TABLE_ID_LEN] = {0};
+    char dbName[TSDB_TABLE_ID_LEN * 3] = {0};
     sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db);
     SDbObj *pDb = mgmtGetDb(dbName);
     if (pDb == NULL) {

src/mnode/src/mgmtSuperTable.c

@@ -61,8 +61,6 @@ static void mgmtSuperTableActionInit() {
   mgmtSuperTableActionFp[SDB_TYPE_DECODE]  = mgmtSuperTableActionDecode;
   mgmtSuperTableActionFp[SDB_TYPE_RESET]   = mgmtSuperTableActionReset;
   mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy;
-  mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mgmtGetShowSuperTableMeta);
-  mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mgmtRetrieveShowSuperTables);
 }
 
 void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {

src/mnode/src/mgmtTable.c

@@ -55,6 +55,9 @@ static void mgmtProcessSuperTableMetaMsg(SQueuedMsg *queueMsg);
 static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg);
 static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg);
 static int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
+static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg);
+static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg);
+static int32_t mgmtGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
 
@@ -86,8 +89,8 @@ int32_t mgmtInitTables() {
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables);
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP, mgmtProcessCreateTableRsp);
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_TABLE_RSP, mgmtProcessDropTableRsp);
-  mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, NULL);
-  mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, NULL);
+  mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, mgmtProcessAlterTableRsp);
+  mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, mgmtProcessDropStableRsp);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -111,17 +114,15 @@ STableInfo* mgmtGetTable(char *tableId) {
   return NULL;
 }
 
-STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid) {
-  SDnodeObj *pObj = mgmtGetDnode(dnodeIp);
-  if (pObj != NULL && vnode >= 0 && vnode < pObj->numOfVnodes) {
-    int32_t vgId = pObj->vload[vnode].vgId;
-    SVgObj *pVgroup = mgmtGetVgroup(vgId);
-    if (pVgroup) {
-      return pVgroup->tableList[sid];
-    }
+STableInfo* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_t sid) {
+  SDnodeObj *pObj = mgmtGetDnode(dnodeId);
+  SVgObj *pVgroup = mgmtGetVgroup(vnode);
+
+  if (pObj == NULL || pVgroup == NULL) {
+    return NULL;
   }
 
-  return NULL;
+  return pVgroup->tableList[sid];
 }
 
 int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMetaMsg *pMeta, bool usePublicIp) {
@@ -589,7 +590,7 @@ void mgmtProcessAlterTableMsg(SQueuedMsg *pMsg) {
 void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) {
   SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
   SDbObj* pDb = mgmtGetDbByTableId(pTable->tableId);
-  if (pDb == NULL || pDb->dropStatus != TSDB_DB_STATUS_READY) {
+  if (pDb == NULL || pDb->dirty) {
     mError("table:%s, failed to get table meta, db not selected", pTable->tableId);
     rpcRsp.code = TSDB_CODE_DB_NOT_SELECTED;
     rpcSendResponse(&rpcRsp);
@@ -786,6 +787,10 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
   free(queueMsg);
 }
 
+static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg) {
+  mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
+}
+
 static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
   if (rpcMsg->handle == NULL) return;
 
@@ -834,3 +839,37 @@ static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
   mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SUCCESS);
   free(queueMsg);
 }
+
+static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
+ mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
+}
+
+//
+//
+//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
+//  SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
+//  pCfg->dnode = htonl(pCfg->dnode);
+//  pCfg->vnode = htonl(pCfg->vnode);
+//  pCfg->sid   = htonl(pCfg->sid);
+//  mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
+//
+//  if (!sdbMaster) {
+//    mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
+//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
+//    return;
+//  }
+//
+//  STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
+//  if (pTable == NULL) {
+//    mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
+//    mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
+//    return;
+//  }
+//
+//  mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
+//
+//  //TODO
+//  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
+//  mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
+//}
+//
\ No newline at end of file

src/mnode/src/mgmtVgroup.c

@@ -110,11 +110,11 @@ int32_t mgmtInitVgroups() {
 
     if (tsIsCluster && pVgroup->vnodeGid[0].publicIp == 0) {
       pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp);
-      pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp);
+      pVgroup->vnodeGid[0].privateIp = inet_addr(tsPrivateIp);
       sdbUpdateRow(tsVgroupSdb, pVgroup, tsVgUpdateSize, 1);
     }
 
-    mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
+    // mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
   }
 
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta);
@@ -130,9 +130,6 @@ SVgObj *mgmtGetVgroup(int32_t vgId) {
   return (SVgObj *)sdbGetRow(tsVgroupSdb, &vgId);
 }
 
-/*
- * TODO: check if there is enough sids
- */
 SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb) {
   return pDb->pHead;
 }
@@ -161,13 +158,13 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg) {
   pVgroup->idPool      = taosInitIdPool(pDb->cfg.maxSessions);
 
   mgmtAddVgroupIntoDb(pDb, pVgroup);
-  mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
+  // mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
 
   sdbInsertRow(tsVgroupSdb, pVgroup, 0);
 
   mPrint("vgroup:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
-    mPrint("vgroup:%d, dnode:%s vnode:%d", pVgroup->vgId, taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode);
+    mPrint("vgroup:%d, dnode:%d vnode:%d", pVgroup->vgId, pVgroup->vnodeGid[i].dnodeId, pVgroup->vnodeGid[i].vnode);
   }
 
   pMsg->ahandle = pVgroup;
@@ -304,9 +301,9 @@ int32_t mgmtGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
 }
 
 char *mgmtGetVnodeStatus(SVgObj *pVgroup, SVnodeGid *pVnode) {
-  SDnodeObj *pDnode = mgmtGetDnode(pVnode->ip);
+  SDnodeObj *pDnode = mgmtGetDnode(pVnode->dnodeId);
   if (pDnode == NULL) {
-    mError("dnode:%s, vgroup:%d, vnode:%d dnode not exist", taosIpStr(pVnode->ip), pVgroup->vgId, pVnode->vnode);
+    mError("vgroup:%d, not exist in dnode:%d", pVgroup->vgId, pDnode->dnodeId);
     return "null";
   }
 
@@ -314,14 +311,13 @@ char *mgmtGetVnodeStatus(SVgObj *pVgroup, SVnodeGid *pVnode) {
     return "offline";
   }
 
-  SVnodeLoad *vload = pDnode->vload + pVnode->vnode;
-  if (vload->vgId != pVgroup->vgId || vload->vnode != pVnode->vnode) {
-    mError("dnode:%s, vgroup:%d, vnode:%d not same with dnode vgroup:%d vnode:%d",
-           taosIpStr(pVnode->ip), pVgroup->vgId, pVnode->vnode, vload->vgId, vload->vnode);
-    return "null";
+  for (int i = 0; i < pDnode->openVnodes; ++i) {
+    if (pDnode->vload[i].vgId == pVgroup->vgId) {
+       return (char*)taosGetVnodeStatusStr(pDnode->vload[i].status);
+    }
   }
-
-  return (char*)taosGetVnodeStatusStr(vload->status);
+  
+  return "null";
 }
 
 int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
@@ -361,7 +357,7 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
     cols++;
 
     for (int32_t i = 0; i < maxReplica; ++i) {
-      tinet_ntoa(ipstr, pVgroup->vnodeGid[i].ip);
+      tinet_ntoa(ipstr, pVgroup->vnodeGid[i].privateIp);
       pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
       strcpy(pWrite, ipstr);
       cols++;
@@ -371,7 +367,7 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
       cols++;
 
       pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-      if (pVgroup->vnodeGid[i].ip != 0) {
+      if (pVgroup->vnodeGid[i].dnodeId != 0) {
         char *vnodeStatus = mgmtGetVnodeStatus(pVgroup, pVgroup->vnodeGid + i);
         strcpy(pWrite, vnodeStatus);
       } else {
@@ -393,6 +389,11 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
 }
 
 static void *mgmtVgroupActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
+  SVgObj *pVgroup = row;
+  for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
+    pVgroup->vnodeGid[i].vnode = pVgroup->vgId;
+  }
+
   return NULL;
 }
 
@@ -404,7 +405,7 @@ static void *mgmtVgroupActionDelete(void *row, char *str, int32_t size, int32_t
     mgmtRemoveVgroupFromDb(pDb, pVgroup);
   }
 
-  mgmtUnSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes);
+  // mgmtUnSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes);
   tfree(pVgroup->tableList);
 
   return NULL;
@@ -488,15 +489,14 @@ void mgmtRemoveTableFromVgroup(SVgObj *pVgroup, STableInfo *pTable) {
   taosFreeId(pVgroup->idPool, pTable->sid);
 }
 
-SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
+SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup) {
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) return NULL;
 
   SMDCreateVnodeMsg *pVnode = rpcMallocCont(sizeof(SMDCreateVnodeMsg));
   if (pVnode == NULL) return NULL;
 
-  pVnode->vnode = htonl(vnode);
-  pVnode->cfg   = pDb->cfg;
+  pVnode->cfg = pDb->cfg;
 
   SVnodeCfg *pCfg = &pVnode->cfg;
   pCfg->vgId                         = htonl(pVgroup->vgId);
@@ -514,8 +514,7 @@ SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
 
   SVnodeDesc *vpeerDesc = pVnode->vpeerDesc;
   for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {
-    vpeerDesc[j].ip    = htonl(pVgroup->vnodeGid[j].ip);
-    vpeerDesc[j].vnode = htonl(pVgroup->vnodeGid[j].vnode);
+    vpeerDesc[j].ip    = htonl(pVgroup->vnodeGid[j].privateIp);
   }
 
   return pVnode;
@@ -542,7 +541,7 @@ SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
     .port = tsDnodeMnodePort
   };
   for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
-    ipSet.ip[i] = pVgroup->vnodeGid[i].ip;
+    ipSet.ip[i] = pVgroup->vnodeGid[i].privateIp;
   }
   return ipSet;
 }
@@ -557,9 +556,9 @@ SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip) {
   return ipSet;
 }
 
-void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
-  mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, vnode, ahandle);
-  SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup, vnode);
+void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
+  mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, pVgroup->vgId, ahandle);
+  SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup);
   SRpcMsg rpcMsg = {
     .handle  = ahandle,
     .pCont   = pCreate,
@@ -573,8 +572,8 @@ void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, vo
 void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
   mTrace("vgroup:%d, send create all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
-    SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
-    mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
+    SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].privateIp);
+    mgmtSendCreateVnodeMsg(pVgroup, &ipSet, ahandle);
   }
 }
 
@@ -613,17 +612,17 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
   free(queueMsg);
 }
 
-static SMDDropVnodeMsg *mgmtBuildDropVnodeMsg(SVgObj *pVgroup) {
+static SMDDropVnodeMsg *mgmtBuildDropVnodeMsg(int32_t vgId) {
   SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
   if (pDrop == NULL) return NULL;
 
-  pDrop->vgId = htonl(pVgroup->vgId);
+  pDrop->vgId = htonl(vgId);
   return pDrop;
 }
 
-static void mgmtSendDropVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
-  mTrace("vgroup:%d, send drop vnode msg, ahandle:%p", pVgroup->vgId, ahandle);
-  SMDDropVnodeMsg *pDrop = mgmtBuildDropVnodeMsg(pVgroup);
+void mgmtSendDropVnodeMsg(int32_t vgId, SRpcIpSet *ipSet, void *ahandle) {
+  mTrace("vgroup:%d, send drop vnode msg, ahandle:%p", vgId, ahandle);
+  SMDDropVnodeMsg *pDrop = mgmtBuildDropVnodeMsg(vgId);
   SRpcMsg rpcMsg = {
       .handle  = ahandle,
       .pCont   = pDrop,
@@ -637,8 +636,8 @@ static void mgmtSendDropVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandl
 static void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle) {
   mTrace("vgroup:%d, send drop all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
-    SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
-    mgmtSendDropVnodeMsg(pVgroup, &ipSet, ahandle);
+    SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].privateIp);
+    mgmtSendDropVnodeMsg(pVgroup->vgId, &ipSet, ahandle);
   }
 }
 
@@ -673,4 +672,49 @@ static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
   mgmtAddToShellQueue(newMsg);
 
   free(queueMsg);
-}
+}
\ No newline at end of file
+
+void mgmtUpdateVgroupIp(SDnodeObj *pDnode) {
+  void *  pNode = NULL;
+  SVgObj *pVgroup = NULL;
+  while (1) {
+    pNode = sdbFetchRow(tsVgroupSdb, pNode, (void **)&pVgroup);
+    if (pVgroup == NULL) break;
+
+    for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
+      SVnodeGid *vnodeGid = pVgroup->vnodeGid + i;
+      if (vnodeGid->dnodeId == pDnode->dnodeId) {
+        mPrint("vgroup:%d, dnode:%d, privateIp:%s change to %s, publicIp:%s change to %s",
+               pVgroup->vgId, vnodeGid->dnodeId, pDnode->privateIp, taosIpStr(vnodeGid->privateIp),
+               pDnode->publicIp, taosIpStr(vnodeGid->publicIp));
+        vnodeGid->publicIp = pDnode->publicIp;
+        vnodeGid->privateIp = pDnode->privateIp;
+        sdbUpdateRow(tsVgroupSdb, pVgroup, tsVgUpdateSize, 1);
+      }
+    }
+  }
+}
+
+//static void mgmtProcessVnodeCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *pConn) {
+//  if (!sdbMaster) {
+//    mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
+//    return;
+//  }
+//
+//  SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) pCont;
+//  pCfg->dnode = htonl(pCfg->dnode);
+//  pCfg->vnode = htonl(pCfg->vnode);
+//
+//  SVgObj *pVgroup = mgmtGetVgroupByVnode(pCfg->dnode, pCfg->vnode);
+//  if (pVgroup == NULL) {
+//    mTrace("dnode:%s, vnode:%d, no vgroup info", taosIpStr(pCfg->dnode), pCfg->vnode);
+//    mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_NOT_ACTIVE_VNODE, NULL, 0);
+//    return;
+//  }
+//
+//  mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
+//
+//  SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
+//  mgmtSendCreateVnodeMsg(pVgroup, pCfg->vnode, &ipSet, NULL);
+//}
+//
\ No newline at end of file

src/util/inc/ihash.h

@@ -36,7 +36,7 @@ int32_t taosHashInt(void *handle, uint64_t key);
 
 void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *));
 
-char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *));
+void taosVisitIntHashWithFp(void *handle, void (*fp)(char *, void *), void *param);
 
 int32_t taosGetIntHashSize(void *handle);
 

src/util/src/ihash.c

@@ -187,7 +187,6 @@ void taosCleanUpIntHash(void *handle) {
   free(pObj);
 }
 
-
 void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
   IHashObj * pObj;
   IHashNode *pNode, *pNext;
@@ -202,7 +201,7 @@ void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
       pNode = pObj->hashList[i];
       while (pNode) {
         pNext = pNode->next;
-        if (fp != NULL) fp(pNode->data);
+        if (fp != NULL) (*fp)(pNode->data);
         free(pNode);
         pNode = pNext;
       }
@@ -219,7 +218,7 @@ void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
   free(pObj);
 }
 
-char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *)) {
+void taosVisitIntHashWithFp(void *handle, int (*fp)(char *, void *), void *param) {
   IHashObj * pObj;
   IHashNode *pNode, *pNext;
   char *     pData = NULL;
@@ -234,21 +233,13 @@ char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *)) {
       pNode = pObj->hashList[i];
       while (pNode) {
         pNext = pNode->next;
-        int flag = fp(pNode->data);
-        if (flag) {
-          pData = pNode->data;
-          goto VisitEnd;
-        }
-
+        (*fp)(pNode->data, param);
         pNode = pNext;
       }
     }
   }
 
-VisitEnd:
-
   pthread_mutex_unlock(&pObj->mutex);
-  return pData;
 }
 
 int32_t taosGetIntHashSize(void *handle) {

src/util/src/tglobalcfg.c

@@ -135,7 +135,7 @@ int tsOfflineThreshold = 864000;   // seconds 10days
 int tsMgmtEqualVnodeNum = 4;
 
 int tsEnableHttpModule = 1;
-int tsEnableMonitorModule = 1;
+int tsEnableMonitorModule = 0;
 int tsRestRowLimit = 10240;
 int tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
 

src/util/src/tstatus.c

@@ -18,13 +18,13 @@
 
 char* taosGetVgroupStatusStr(int32_t vgroupStatus) {
   switch (vgroupStatus) {
-    case TSDB_VG_STATUS_READY:                 return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_IN_PROGRESS:           return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_NO_DISK_PERMISSIONS:   return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_SERVER_NO_PACE:        return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_SERV_OUT_OF_MEMORY:    return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_INIT_FAILED:           return tstrerror(vgroupStatus);
-    case TSDB_VG_STATUS_FULL:                  return tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_READY:                 return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_IN_PROGRESS:           return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_NO_DISK_PERMISSIONS:   return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_SERVER_NO_PACE:        return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_SERV_OUT_OF_MEMORY:    return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_INIT_FAILED:           return (char*)tstrerror(vgroupStatus);
+    case TSDB_VG_STATUS_FULL:                  return (char*)tstrerror(vgroupStatus);
     default:                                   return "undefined";
   }
 }

src/vnode/tsdb/inc/tsdb.h

@@ -37,7 +37,6 @@ extern "C" {
 // --------- TSDB REPOSITORY CONFIGURATION DEFINITION
 typedef struct {
   int8_t  precision;
-  int32_t vgId;
   int32_t tsdbId;
   int32_t maxTables;            // maximum number of tables this repository can have
   int32_t daysPerFile;          // day per file sharding policy