Merge pull request #1606 from taosdata/feature/mpeer

Feature/mpeer

cmake/define.inc

@@ -11,9 +11,9 @@ ENDIF ()
 
 IF (TD_VPEER)
   ADD_DEFINITIONS(-D_VPEER)
-  ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=3)
+  #ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=3)
 ELSE ()
-  ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=1)  
+  #ADD_DEFINITIONS(-DTSDB_REPLICA_MAX_NUM=1)  
 ENDIF ()
 
 IF (TD_ACCOUNT)

src/dnode/inc/dnodeMClient.h

@@ -24,6 +24,7 @@ int32_t  dnodeInitMClient();
 void     dnodeCleanupMClient();
 void     dnodeSendMsgToMnode(SRpcMsg *rpcMsg);
 uint32_t dnodeGetMnodeMasteIp();
+void *   dnodeGetMpeerInfos();
 
 #ifdef __cplusplus
 }

src/dnode/src/dnodeMClient.c

@@ -15,33 +15,45 @@
 
 #define _DEFAULT_SOURCE
 #include "os.h"
+#include "cJSON.h"
 #include "taosmsg.h"
 #include "tlog.h"
 #include "trpc.h"
 #include "tutil.h"
+#include "tsync.h"
 #include "dnode.h"
 #include "dnodeMClient.h"
 #include "dnodeModule.h"
 #include "dnodeMgmt.h"
 
+#define MPEER_CONTENT_LEN 2000
+
 static bool   dnodeReadMnodeIpList();
 static void   dnodeSaveMnodeIpList();
 static void   dnodeProcessRspFromMnode(SRpcMsg *pMsg);
 static void   dnodeProcessStatusRsp(SRpcMsg *pMsg);
 static void (*tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void  *tsDnodeMClientRpc = NULL;
-static SRpcIpSet tsDnodeMnodeIpList  = {0};
+static SRpcIpSet    tsMnodeIpList  = {0};
+static SDMNodeInfos tsMnodeInfos = {0};
 
 int32_t dnodeInitMClient() {
   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);
+    memset(&tsMnodeIpList, 0, sizeof(SRpcIpSet));
+    memset(&tsMnodeInfos, 0, sizeof(SDMNodeInfos));
+    tsMnodeIpList.port = tsMnodeDnodePort;
+    tsMnodeIpList.numOfIps = 1;
+    tsMnodeIpList.ip[0] = inet_addr(tsMasterIp);
     if (tsSecondIp[0]) {
-      tsDnodeMnodeIpList.numOfIps = 2;
-      tsDnodeMnodeIpList.ip[1] = inet_addr(tsSecondIp);
+      tsMnodeIpList.numOfIps = 2;
+      tsMnodeIpList.ip[1] = inet_addr(tsSecondIp);
+    }
+  } else {
+    tsMnodeIpList.inUse = tsMnodeInfos.inUse;
+    tsMnodeIpList.numOfIps = tsMnodeInfos.nodeNum;
+    tsMnodeIpList.port = tsMnodeInfos.nodeInfos[0].nodePort;
+    for (int32_t i = 0; i < tsMnodeInfos.nodeNum; i++) {
+      tsMnodeIpList.ip[i] = tsMnodeInfos.nodeInfos[i].nodeIp;
     }
   }
 
@@ -96,23 +108,31 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
   }
 
   SDMStatusRsp *pStatusRsp = pMsg->pCont;
-  if (pStatusRsp->ipList.numOfIps <= 0) {
-    dError("status msg is invalid, num of ips is %d", pStatusRsp->ipList.numOfIps);
+  SDMNodeInfos *mpeers = &pStatusRsp->mpeers;
+  if (mpeers->nodeNum <= 0) {
+    dError("status msg is invalid, num of ips is %d", mpeers->nodeNum);
     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]);
+  SRpcIpSet mgmtIpSet = {0};
+  mgmtIpSet.inUse = mpeers->inUse;
+  mgmtIpSet.numOfIps = mpeers->nodeNum;
+  mgmtIpSet.port = htons(mpeers->nodeInfos[0].nodePort);
+  for (int32_t i = 0; i < mpeers->nodeNum; i++) {
+    mgmtIpSet.ip[i] = htonl(mpeers->nodeInfos[i].nodeIp);
   }
 
-  //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]));
+  if (memcmp(&mgmtIpSet, &tsMnodeIpList, sizeof(SRpcIpSet)) != 0) {
+    memcpy(&tsMnodeIpList, &mgmtIpSet, sizeof(SRpcIpSet));  
+    memcpy(&tsMnodeInfos, mpeers, sizeof(SDMNodeInfos));
+    dPrint("mnode ip list is changed, numOfIps:%d inUse:%d", tsMnodeInfos.nodeNum, tsMnodeInfos.inUse);
+    for (int32_t i = 0; i < mpeers->nodeNum; i++) {
+      tsMnodeInfos.nodeInfos[i].nodeId = htonl(mpeers->nodeInfos[i].nodeId);
+      tsMnodeInfos.nodeInfos[i].nodeIp = htonl(mpeers->nodeInfos[i].nodeIp);
+      tsMnodeInfos.nodeInfos[i].nodePort = htons(mpeers->nodeInfos[i].nodePort);
+      dPrint("mnode:%d, ip:%s:%u name:%s", tsMnodeInfos.nodeInfos[i].nodeId,
+             taosIpStr(tsMnodeInfos.nodeInfos[i].nodeId), tsMnodeInfos.nodeInfos[i].nodePort,
+             tsMnodeInfos.nodeInfos[i].nodeName);
     }
     dnodeSaveMnodeIpList();
   }
@@ -129,70 +149,149 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
 
 void dnodeSendMsgToMnode(SRpcMsg *rpcMsg) {
   if (tsDnodeMClientRpc) {
-    rpcSendRequest(tsDnodeMClientRpc, &tsDnodeMnodeIpList, rpcMsg);
+    rpcSendRequest(tsDnodeMClientRpc, &tsMnodeIpList, rpcMsg);
   }
 }
 
 static bool dnodeReadMnodeIpList() {
   char ipFile[TSDB_FILENAME_LEN] = {0};
-  sprintf(ipFile, "%s/iplist", tsDnodeDir);
-
+  sprintf(ipFile, "%s/mgmtIpList.json", 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;;
+  if (!fp) {
+    dTrace("failed to read mnode mgmtIpList.json, file not exist");
+    return false;
+  }
 
-  num = fscanf(fp, "%s %d", option, &value);
-  if (num != 2) return false;
-  if (strcmp(option, "numOfIps") != 0) return false;
-  tsDnodeMnodeIpList.numOfIps = (int8_t)value;
+  bool  ret = false;
+  int   maxLen = 2000;
+  char *content = calloc(1, maxLen + 1);
+  int   len = fread(content, 1, maxLen, fp);
+  if (len <= 0) {
+    free(content);
+    fclose(fp);
+    dError("failed to read mnode mgmtIpList.json, content is null");
+    return false;
+  }
 
-  num = fscanf(fp, "%s %d", option, &value);
-  if (num != 2) return false;
-  if (strcmp(option, "port") != 0) return false;
-  tsDnodeMnodeIpList.port = (uint16_t)value;
+  cJSON* root = cJSON_Parse(content);
+  if (root == NULL) {
+    dError("failed to read mnode mgmtIpList.json, invalid json format");
+    goto PARSE_OVER;
+  }
 
-  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;
+  cJSON* inUse = cJSON_GetObjectItem(root, "inUse");
+  if (!inUse || inUse->type != cJSON_Number) {
+    dError("failed to read mnode mgmtIpList.json, inUse not found");
+    goto PARSE_OVER;
   }
+  tsMnodeInfos.inUse = inUse->valueint;
 
-  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]));
-  } 
+  cJSON* nodeNum = cJSON_GetObjectItem(root, "nodeNum");
+  if (!nodeNum || nodeNum->type != cJSON_Number) {
+    dError("failed to read mnode mgmtIpList.json, nodeNum not found");
+    goto PARSE_OVER;
+  }
+  tsMnodeInfos.nodeNum = nodeNum->valueint;
+
+  cJSON* nodeInfos = cJSON_GetObjectItem(root, "nodeInfos");
+  if (!nodeInfos || nodeInfos->type != cJSON_Array) {
+    dError("failed to read mnode mgmtIpList.json, nodeInfos not found");
+    goto PARSE_OVER;
+  }
+
+  int size = cJSON_GetArraySize(nodeInfos);
+  if (size != tsMnodeInfos.nodeNum) {
+    dError("failed to read mnode mgmtIpList.json, nodeInfos size not matched");
+    goto PARSE_OVER;
+  }
+
+  for (int i = 0; i < size; ++i) {
+    cJSON* nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
+    if (nodeInfo == NULL) continue;
+
+    cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId");
+    if (!nodeId || nodeId->type != cJSON_Number) {
+      dError("failed to read mnode mgmtIpList.json, nodeId not found");
+      goto PARSE_OVER;
+    }
+    tsMnodeInfos.nodeInfos[i].nodeId = nodeId->valueint;
 
-  return true;
+    cJSON *nodeIp = cJSON_GetObjectItem(nodeInfo, "nodeIp");
+    if (!nodeIp || nodeIp->type != cJSON_String || nodeIp->valuestring == NULL) {
+      dError("failed to read mnode mgmtIpList.json, nodeIp not found");
+      goto PARSE_OVER;
+    }
+    tsMnodeInfos.nodeInfos[i].nodeIp = inet_addr(nodeIp->valuestring);
+
+    cJSON *nodePort = cJSON_GetObjectItem(nodeInfo, "nodePort");
+    if (!nodePort || nodePort->type != cJSON_Number) {
+      dError("failed to read mnode mgmtIpList.json, nodePort not found");
+      goto PARSE_OVER;
+    }
+    tsMnodeInfos.nodeInfos[i].nodePort = (uint16_t)nodePort->valueint;
+
+    cJSON *nodeName = cJSON_GetObjectItem(nodeInfo, "nodeName");
+    if (!nodeIp || nodeName->type != cJSON_String || nodeName->valuestring == NULL) {
+      dError("failed to read mnode mgmtIpList.json, nodeName not found");
+      goto PARSE_OVER;
+    }
+    strncpy(tsMnodeInfos.nodeInfos[i].nodeName, nodeName->valuestring, TSDB_NODE_NAME_LEN);
+  }
+
+  ret = true;
+
+  dPrint("read mnode iplist successed, numOfIps:%d inUse:%d", tsMnodeInfos.nodeNum, tsMnodeInfos.inUse);
+  for (int32_t i = 0; i < tsMnodeInfos.nodeNum; i++) {
+    dPrint("mnode:%d, ip:%s:%u name:%s", tsMnodeInfos.nodeInfos[i].nodeId,
+            taosIpStr(tsMnodeInfos.nodeInfos[i].nodeId), tsMnodeInfos.nodeInfos[i].nodePort,
+            tsMnodeInfos.nodeInfos[i].nodeName);
+  }
+
+PARSE_OVER:
+  free(content);
+  cJSON_Delete(root);
+  fclose(fp);
+  return ret;
 }
 
 static void dnodeSaveMnodeIpList() {
   char ipFile[TSDB_FILENAME_LEN] = {0};
-  sprintf(ipFile, "%s/iplist", tsDnodeDir);
-
+  sprintf(ipFile, "%s/mgmtIpList.json", 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]);
+  int32_t len = 0;
+  int32_t maxLen = 2000;
+  char *  content = calloc(1, maxLen + 1);
+
+  len += snprintf(content + len, maxLen - len, "{\n");
+  len += snprintf(content + len, maxLen - len, "  \"inUse\": %d,\n", tsMnodeInfos.inUse);
+  len += snprintf(content + len, maxLen - len, "  \"nodeNum\": %d,\n", tsMnodeInfos.nodeNum);
+  len += snprintf(content + len, maxLen - len, "  \"nodeInfos\": [{\n");
+  for (int32_t i = 0; i < tsMnodeInfos.nodeNum; i++) {
+    len += snprintf(content + len, maxLen - len, "    \"nodeId\": %d,\n", tsMnodeInfos.nodeInfos[i].nodeId);
+    len += snprintf(content + len, maxLen - len, "    \"nodeIp\": \"%s\",\n", taosIpStr(tsMnodeInfos.nodeInfos[i].nodeIp));
+    len += snprintf(content + len, maxLen - len, "    \"nodePort\": %u,\n", tsMnodeInfos.nodeInfos[i].nodePort);
+    len += snprintf(content + len, maxLen - len, "    \"nodeName\": \"%s\"\n",  tsMnodeInfos.nodeInfos[i].nodeName);
+    if (i < tsMnodeInfos.nodeNum -1) {
+      len += snprintf(content + len, maxLen - len, "  },{\n");  
+    } else {
+      len += snprintf(content + len, maxLen - len, "  }]\n");  
+    }
   }
-  
+  len += snprintf(content + len, maxLen - len, "}\n"); 
+
+  fwrite(content, 1, len, fp);
   fclose(fp);
+  free(content);
+  
   dPrint("save mnode iplist successed");
 }
 
 uint32_t dnodeGetMnodeMasteIp() {
-  return tsDnodeMnodeIpList.ip[0];
+  return tsMnodeIpList.ip[tsMnodeIpList.inUse];
+}
+
+void* dnodeGetMpeerInfos() {
+  return &tsMnodeInfos;
 }
\ No newline at end of file

src/dnode/src/dnodeMgmt.c

@@ -45,7 +45,7 @@ static void    *tsDnodeTmr = NULL;
 static void    *tsStatusTimer = NULL;
 static uint32_t tsRebootTime;
 static int32_t  tsDnodeId = 0;
-static char     tsDnodeName[TSDB_DNODE_NAME_LEN];
+static char     tsDnodeName[TSDB_NODE_NAME_LEN];
 
 int32_t dnodeInitMgmt() {
   dnodeReadDnodeId();

src/inc/mnode.h

@@ -41,28 +41,21 @@ struct _vg_obj;
 struct _db_obj;
 struct _acct_obj;
 struct _user_obj;
+struct _mnode_obj;
 
-typedef struct {
+typedef struct _mnode_obj {
   int32_t  mnodeId;
-  uint32_t privateIp;
-  uint32_t publicIp;
+  int32_t  dnodeId;
   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];
-  int8_t   reserved[15];
+  int8_t   reserved[14];
   int8_t   updateEnd[1];
   int32_t  refCount;
-  int      syncFd;
-  void    *hbTimer;
-  void    *pSync;
+  int8_t   role;
+  int8_t   status;
+  uint16_t port;
+  uint32_t privateIp;
+  uint32_t publicIp;  
+  char     mnodeName[TSDB_NODE_NAME_LEN + 1];
 } SMnodeObj;
 
 typedef struct _dnode_obj {
@@ -81,7 +74,7 @@ typedef struct _dnode_obj {
   int8_t     alternativeRole;  // from dnode status msg, 0-any, 1-mgmt, 2-dnode
   int8_t     status;           // set in balance function
   int32_t    customScore;      // config by user
-  char       dnodeName[TSDB_DNODE_NAME_LEN + 1];
+  char       dnodeName[TSDB_NODE_NAME_LEN + 1];
   int8_t     reserved[15];
   int8_t     updateEnd[1];
   int32_t    refCount;

src/mnode/inc/mgmtMnode.h → src/inc/mpeer.h

@@ -13,35 +13,37 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_MGMT_MNODE_H
-#define TDENGINE_MGMT_MNODE_H
+#ifndef TDENGINE_MPEER_H
+#define TDENGINE_MPEER_H
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-enum _TSDB_MN_STATUS {
-  TSDB_MN_STATUS_OFFLINE,
-  TSDB_MN_STATUS_UNSYNCED,
-  TSDB_MN_STATUS_SYNCING,
-  TSDB_MN_STATUS_SERVING
-};
+struct _mnode_obj;
 
-enum _TSDB_MN_ROLE {
-  TSDB_MN_ROLE_UNDECIDED,
-  TSDB_MN_ROLE_SLAVE,
-  TSDB_MN_ROLE_MASTER
+enum _TAOS_MN_STATUS {
+  TAOS_MN_STATUS_OFFLINE,
+  TAOS_MN_STATUS_DROPPING,
+  TAOS_MN_STATUS_READY
 };
 
-int32_t mgmtInitMnodes();
-void    mgmtCleanupMnodes();
+int32_t mpeerInit();
+void    mpeerCleanup();
+int32_t mpeerGetMnodesNum();
+void *  mpeerGetNextMnode(void *pNode, struct _mnode_obj **pMnode);
+void    mpeerReleaseMnode(struct _mnode_obj *pMnode);
+
+bool    mpeerInServerStatus();   
+bool    mpeerIsMaster();
+bool    mpeerCheckRedirect();
 
-bool    mgmtInServerStatus();   
-bool    mgmtIsMaster();
+void    mpeerGetPrivateIpList(SRpcIpSet *ipSet);
+void    mpeerGetPublicIpList(SRpcIpSet *ipSet);
+void    mpeerGetMpeerInfos(void *mpeers);
 
-bool    mgmtCheckRedirect(void *handle);
-void    mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet);
-void    mgmtGetMnodePublicIpList(SRpcIpSet *ipSet);
+char *  mpeerGetMnodeStatusStr(int32_t status);
+char *  mpeerGetMnodeRoleStr(int32_t role);
 
 #ifdef __cplusplus
 }

src/inc/taosdef.h

@@ -176,7 +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       64
+#define TSDB_NODE_NAME_LEN        64
 #define TSDB_TABLE_NAME_LEN       192
 #define TSDB_DB_NAME_LEN          32
 #define TSDB_COL_NAME_LEN         64
@@ -233,7 +233,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 /*
  * this is defined in CMakeList.txt
  */
-//#define TSDB_REPLICA_MAX_NUM      3
+#define TSDB_REPLICA_MAX_NUM      3
 
 #define TSDB_TBNAME_COLUMN_INDEX       (-1)
 #define TSDB_MULTI_METERMETA_MAX_NUM    100000  // maximum batch size allowed to load metermeta

src/inc/taosmsg.h

@@ -577,7 +577,7 @@ typedef struct {
 typedef struct {
   uint32_t   version;
   int32_t    dnodeId;
-  char       dnodeName[TSDB_DNODE_NAME_LEN];
+  char       dnodeName[TSDB_NODE_NAME_LEN + 1];
   uint32_t   privateIp;
   uint32_t   publicIp;
   uint32_t   lastReboot;        // time stamp for last reboot
@@ -591,7 +591,20 @@ typedef struct {
 } SDMStatusMsg;
 
 typedef struct {
-  SRpcIpSet    ipList;
+  int32_t   nodeId;
+  uint32_t  nodeIp;
+  uint16_t  nodePort;
+  char      nodeName[TSDB_NODE_NAME_LEN + 1];
+} SDMNodeInfo;
+
+typedef struct {
+  int8_t       inUse;
+  int8_t       nodeNum;
+  SDMNodeInfo  nodeInfos[TSDB_MAX_MPEERS];
+} SDMNodeInfos;
+
+typedef struct {
+  SDMNodeInfos mpeers;
   SDnodeState  dnodeState;
   SVnodeAccess vnodeAccess[];
 } SDMStatusRsp;

src/mnode/CMakeLists.txt

@@ -13,5 +13,5 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   AUX_SOURCE_DIRECTORY(src SRC)
 
   ADD_LIBRARY(mnode ${SRC})
-  TARGET_LINK_LIBRARIES(mnode trpc tutil pthread)
+  #TARGET_LINK_LIBRARIES(mnode trpc tutil pthread)
 ENDIF ()
\ No newline at end of file

src/mnode/inc/mgmtSdb.h

@@ -21,21 +21,31 @@ extern "C" {
 #endif
 
 typedef enum {
-  SDB_KEY_TYPE_STRING, 
-  SDB_KEY_TYPE_AUTO
+  SDB_TABLE_MNODE   = 0,
+  SDB_TABLE_DNODE   = 1,
+  SDB_TABLE_ACCOUNT = 2,
+  SDB_TABLE_USER    = 3,
+  SDB_TABLE_DB      = 4,
+  SDB_TABLE_VGROUP  = 5,
+  SDB_TABLE_STABLE  = 6,
+  SDB_TABLE_CTABLE  = 7,
+  SDB_TABLE_MAX     = 8
+} ESdbTable;
+
+typedef enum {
+  SDB_KEY_STRING, 
+  SDB_KEY_AUTO
 } ESdbKeyType;
 
 typedef enum {
-  SDB_OPER_TYPE_GLOBAL,
-  SDB_OPER_TYPE_LOCAL
+  SDB_OPER_GLOBAL,
+  SDB_OPER_LOCAL
 } ESdbOperType;
 
 typedef struct {
   ESdbOperType type;
   void *  table;
   void *  pObj;
-  int64_t version;
-  int32_t maxRowSize;
   int32_t rowSize;
   void *  rowData;
 } SSdbOperDesc;
@@ -45,6 +55,7 @@ typedef struct {
   int32_t hashSessions;
   int32_t maxRowSize;
   int32_t refCountPos;
+  ESdbTable   tableId;
   ESdbKeyType keyType;
   int32_t (*insertFp)(SSdbOperDesc *pOper);
   int32_t (*deleteFp)(SSdbOperDesc *pOper);
@@ -52,8 +63,12 @@ typedef struct {
   int32_t (*encodeFp)(SSdbOperDesc *pOper);
   int32_t (*decodeFp)(SSdbOperDesc *pDesc);  
   int32_t (*destroyFp)(SSdbOperDesc *pDesc);
+  int32_t (*updateAllFp)();
 } SSdbTableDesc;
 
+int32_t sdbInit();
+void    sdbCleanUp();
+
 void *  sdbOpenTable(SSdbTableDesc *desc);
 void    sdbCloseTable(void *handle);
 

src/mnode/src/mgmtDb.c

@@ -24,8 +24,8 @@
 #include "mgmtDb.h"
 #include "tcluster.h"
 #include "tgrant.h"
+#include "mpeer.h"
 #include "mgmtShell.h"
-#include "mgmtMnode.h"
 #include "mgmtProfile.h"
 #include "mgmtSdb.h"
 #include "mgmtTable.h"
@@ -88,14 +88,9 @@ static int32_t mgmtDbActionUpdate(SSdbOperDesc *pOper) {
 
 static int32_t mgmtDbActionEncode(SSdbOperDesc *pOper) {
   SDbObj *pDb = pOper->pObj;
-
-  if (pOper->maxRowSize < tsDbUpdateSize) {
-    return -1;
-  } else {
-    memcpy(pOper->rowData, pDb, tsDbUpdateSize);
-    pOper->rowSize = tsDbUpdateSize;
-    return TSDB_CODE_SUCCESS;
-  }
+  memcpy(pOper->rowData, pDb, tsDbUpdateSize);
+  pOper->rowSize = tsDbUpdateSize;
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t mgmtDbActionDecode(SSdbOperDesc *pOper) {
@@ -107,22 +102,28 @@ static int32_t mgmtDbActionDecode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t mgmtDbActionUpdateAll() {
+  return 0;
+}
+
 int32_t mgmtInitDbs() {
   SDbObj tObj;
   tsDbUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
 
   SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_DB,
     .tableName    = "dbs",
     .hashSessions = TSDB_MAX_DBS,
     .maxRowSize   = tsDbUpdateSize,
     .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
-    .keyType      = SDB_KEY_TYPE_STRING,
+    .keyType      = SDB_KEY_STRING,
     .insertFp     = mgmtDbActionInsert,
     .deleteFp     = mgmtDbActionDelete,
     .updateFp     = mgmtDbActionUpdate,
     .encodeFp     = mgmtDbActionEncode,
     .decodeFp     = mgmtDbActionDecode,
     .destroyFp    = mgmtDbActionDestroy,
+    .updateAllFp  = mgmtDbActionUpdateAll
   };
 
   tsDbSdb = sdbOpenTable(&tableDesc);
@@ -136,7 +137,7 @@ int32_t mgmtInitDbs() {
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_DB, mgmtProcessDropDbMsg);
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_DB, mgmtGetDbMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mgmtRetrieveDbs);
-
+  
   mTrace("db data is initialized");
   return 0;
 }
@@ -310,7 +311,7 @@ static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
   pDb->cfg = *pCreate;
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb,
     .rowSize = sizeof(SDbObj)
@@ -663,7 +664,7 @@ static int32_t mgmtSetDbDropping(SDbObj *pDb) {
 
   pDb->status = true;
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb,
     .rowSize = tsDbUpdateSize
@@ -678,8 +679,6 @@ static int32_t mgmtSetDbDropping(SDbObj *pDb) {
 }
 
 static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg) {
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
   SCMCreateDbMsg *pCreate = pMsg->pCont;
   pCreate->maxSessions     = htonl(pCreate->maxSessions);
   pCreate->cacheBlockSize  = htonl(pCreate->cacheBlockSize);
@@ -750,7 +749,7 @@ static int32_t mgmtAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
   if (memcmp(&newCfg, &pDb->cfg, sizeof(SDbCfg)) != 0) {
     pDb->cfg = newCfg;
     SSdbOperDesc oper = {
-      .type = SDB_OPER_TYPE_GLOBAL,
+      .type = SDB_OPER_GLOBAL,
       .table = tsDbSdb,
       .pObj = pDb,
       .rowSize = tsDbUpdateSize
@@ -808,7 +807,7 @@ static void mgmtDropDb(SQueuedMsg *pMsg) {
   mPrint("db:%s, drop db from sdb", pDb->name);
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb
   };

src/mnode/src/mgmtDnode.c

@@ -19,8 +19,8 @@
 #include "tbalance.h"
 #include "tcluster.h"
 #include "mnode.h"
+#include "mpeer.h"
 #include "mgmtDClient.h"
-#include "mgmtMnode.h"
 #include "mgmtShell.h"
 #include "mgmtDServer.h"
 #include "mgmtUser.h"
@@ -141,8 +141,6 @@ static void clusterProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) {
 }
 
 void clusterProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
-  if (mgmtCheckRedirect(rpcMsg->handle)) return;
-
   SDMStatusMsg *pStatus = rpcMsg->pCont;
   pStatus->dnodeId = htonl(pStatus->dnodeId);
   pStatus->privateIp = htonl(pStatus->privateIp);
@@ -221,7 +219,7 @@ void clusterProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  mgmtGetMnodePrivateIpList(&pRsp->ipList);
+  mpeerGetMpeerInfos(&pRsp->mpeers);
 
   pRsp->dnodeState.dnodeId = htonl(pDnode->dnodeId);
   pRsp->dnodeState.moduleStatus = htonl(pDnode->moduleStatus);

src/mnode/src/mgmtMain.c

@@ -22,11 +22,11 @@
 #include "taccount.h"
 #include "tbalance.h"
 #include "tcluster.h"
+#include "tgrant.h"
+#include "mpeer.h"
 #include "mgmtDb.h"
 #include "mgmtDClient.h"
 #include "mgmtDServer.h"
-#include "tgrant.h"
-#include "mgmtMnode.h"
 #include "mgmtSdb.h"
 #include "mgmtVgroup.h"
 #include "mgmtUser.h"
@@ -105,7 +105,12 @@ int32_t mgmtStartSystem() {
   }
 
   if (mgmtInitTables() < 0) {
-    mError("failed to init meters");
+    mError("failed to init tables");
+    return -1;
+  }
+
+  if (sdbInit() < 0) {
+    mError("failed to init sdb");
     return -1;
   }
 
@@ -117,8 +122,8 @@ int32_t mgmtStartSystem() {
     return -1;
   }
 
-  if (mgmtInitMnodes() < 0) {
-    mError("failed to init mnodes");
+  if (mpeerInit() < 0) {
+    mError("failed to init mpeers");
     return -1;
   }
 
@@ -135,7 +140,7 @@ int32_t mgmtStartSystem() {
 
 
 void mgmtStopSystem() {
-  if (mgmtIsMaster()) {
+  if (mpeerIsMaster()) {
     mTrace("it is a master mgmt node, it could not be stopped");
     return;
   }
@@ -147,7 +152,7 @@ void mgmtStopSystem() {
 void mgmtCleanUpSystem() {
   mPrint("starting to clean up mgmt");
   grantCleanUp();
-  mgmtCleanupMnodes();
+  mpeerCleanup();
   balanceCleanUp();
   mgmtCleanUpShell();
   mgmtCleanupDClient();
@@ -158,6 +163,7 @@ void mgmtCleanUpSystem() {
   clusterCleanUp();
   mgmtCleanUpUsers();
   acctCleanUp();
+  sdbCleanUp();
   taosTmrCleanUp(tsMgmtTmr);
   mPrint("mgmt is cleaned up");
 }

src/mnode/src/mgmtMnode.c

@@ -17,43 +17,36 @@
 #include "os.h"
 #include "taoserror.h"
 #include "trpc.h"
-#include "mgmtMnode.h"
+#include "tsync.h"
+#include "mpeer.h"
 #include "mgmtSdb.h"
 #include "mgmtShell.h"
 #include "mgmtUser.h"
 
-#ifndef _MPEER
-
-static SMnodeObj tsMnodeObj = {0};
+extern int32_t mpeerInitMnodes();
+extern void    mpeerCleanupMnodes();
 static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
-int32_t mgmtInitMnodes() {
-  mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mgmtGetMnodeMeta);
-  mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mgmtRetrieveMnodes);
+#ifndef _MPEER
 
+static SMnodeObj tsMnodeObj = {0};
+
+int32_t mpeerInitMnodes() {
   tsMnodeObj.mnodeId     = 1;
+  tsMnodeObj.dnodeId     = 1;
   tsMnodeObj.privateIp   = inet_addr(tsPrivateIp);
   tsMnodeObj.publicIp    = inet_addr(tsPublicIp);
   tsMnodeObj.createdTime = taosGetTimestampMs();
-  tsMnodeObj.role        = TSDB_MN_ROLE_MASTER;
-  tsMnodeObj.status      = TSDB_MN_STATUS_SERVING;
-  tsMnodeObj.numOfMnodes = 1;
-  sprintf(tsMnodeObj.mnodeName, "%d", tsMnodeObj.mnodeId);
-
+  tsMnodeObj.role        = TAOS_SYNC_ROLE_MASTER;
+  tsMnodeObj.status      = TAOS_MN_STATUS_READY;
+  tsMnodeObj.port        = tsMnodeDnodePort;
+  sprintf(tsMnodeObj.mnodeName, "m%d", tsMnodeObj.mnodeId);
+  
   return TSDB_CODE_SUCCESS;
 }
 
-void mgmtCleanupMnodes() {}
-bool mgmtInServerStatus() { return tsMnodeObj.status == TSDB_MN_STATUS_SERVING; }
-bool mgmtIsMaster() { return tsMnodeObj.role == TSDB_MN_ROLE_MASTER; }
-bool mgmtCheckRedirect(void *thandle) { return false; }
-
-static int32_t mgmtGetMnodesNum() {
-  return 1;
-}
-
-static void *mgmtGetNextMnode(void *pNode, SMnodeObj **pMnode) {
+void *mpeerGetNextMnode(void *pNode, SMnodeObj **pMnode) {
   if (*pMnode == NULL) {
     *pMnode = &tsMnodeObj;
   } else {
@@ -63,22 +56,74 @@ static void *mgmtGetNextMnode(void *pNode, SMnodeObj **pMnode) {
   return *pMnode;
 }
 
-char *taosGetMnodeStatusStr(int32_t mnodeStatus) {
-  switch (mnodeStatus) {
-    case TSDB_MN_STATUS_OFFLINE:   return "offline";
-    case TSDB_MN_STATUS_UNSYNCED:  return "unsynced";
-    case TSDB_MN_STATUS_SYNCING:   return "syncing";
-    case TSDB_MN_STATUS_SERVING:   return "serving";
-    default:                       return "undefined";
+void mpeerGetPrivateIpList(SRpcIpSet *ipSet) {
+  ipSet->inUse = 0;
+  ipSet->port = htons(tsMnodeDnodePort);
+  ipSet->numOfIps = 1;
+  ipSet->ip[0] = htonl(tsMnodeObj.privateIp);
+}
+
+void mpeerGetPublicIpList(SRpcIpSet *ipSet) {
+  ipSet->inUse = 0;
+  ipSet->port = htons(tsMnodeDnodePort);
+  ipSet->numOfIps = 1;
+  ipSet->ip[0] = htonl(tsMnodeObj.publicIp);
+}
+
+void mpeerGetMpeerInfos(void *param) {
+  SDMNodeInfos *mpeers = param;
+  mpeers->nodeNum = 1;
+  mpeers->nodeInfos[0].nodeId = htonl(tsMnodeObj.mnodeId);
+  mpeers->nodeInfos[0].nodeIp = htonl(tsMnodeObj.privateIp);
+  mpeers->nodeInfos[0].nodePort = htons(tsMnodeObj.port);
+  strcpy(mpeers->nodeInfos[0].nodeName, tsMnodeObj.mnodeName);
+}
+
+void    mpeerCleanupDnodes() {}
+int32_t mpeerGetMnodesNum() { return 1; }
+void    mpeerReleaseMnode(struct _mnode_obj *pMnode) {}
+bool    mpeerInServerStatus() { return tsMnodeObj.status == TAOS_MN_STATUS_READY; }
+bool    mpeerIsMaster() { return tsMnodeObj.role == TAOS_SYNC_ROLE_MASTER; }
+bool    mpeerCheckRedirect() { return false; }
+
+#endif
+
+int32_t mpeerInit() {
+  mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mgmtGetMnodeMeta);
+  mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mgmtRetrieveMnodes);
+
+  return mpeerInitMnodes();
+}
+
+void mpeerCleanup() {
+  mpeerCleanupDnodes();
+}
+
+char *mpeerGetMnodeStatusStr(int32_t status) {
+  switch (status) {
+    case TAOS_MN_STATUS_OFFLINE:
+      return "offline";
+    case TAOS_MN_STATUS_DROPPING:
+      return "dropping";
+    case TAOS_MN_STATUS_READY:
+      return "ready";
+    default:
+      return "undefined";
   }
 }
 
-char *taosGetMnodeRoleStr(int32_t mnodeRole) {
-  switch (mnodeRole) {
-    case TSDB_MN_ROLE_UNDECIDED: return "undicided";
-    case TSDB_MN_ROLE_SLAVE:     return "slave";
-    case TSDB_MN_ROLE_MASTER:    return "master";
-    default:                     return "undefined";
+char *mpeerGetMnodeRoleStr(int32_t role) {
+  switch (role) {
+    case TAOS_SYNC_ROLE_OFFLINE:
+      return "offline";
+    case TAOS_SYNC_ROLE_UNSYNCED:
+      return "unsynced";
+    case TAOS_SYNC_ROLE_SLAVE:
+      return "slave";
+    case TAOS_SYNC_ROLE_MASTER:
+      return "master";
+    default:
+      return "undefined";
   }
 }
 
@@ -135,7 +180,7 @@ static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
     pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
   }
 
-  pShow->numOfRows = mgmtGetMnodesNum();
+  pShow->numOfRows = mpeerGetMnodesNum();
   pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
   pShow->pNode = NULL;
   mgmtReleaseUser(pUser);
@@ -151,7 +196,7 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
   char       ipstr[32];
 
   while (numOfRows < rows) {
-    pShow->pNode = mgmtGetNextMnode(pShow->pNode, (SMnodeObj **)&pMnode);
+    pShow->pNode = mpeerGetNextMnode(pShow->pNode, &pMnode);
     if (pMnode == NULL) break;
 
     cols = 0;
@@ -175,11 +220,11 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    strcpy(pWrite, taosGetMnodeStatusStr(pMnode->status));
+    strcpy(pWrite, mpeerGetMnodeStatusStr(pMnode->status));
     cols++;
 
      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    strcpy(pWrite, taosGetMnodeRoleStr(pMnode->role));
+    strcpy(pWrite, mpeerGetMnodeRoleStr(pMnode->role));
     cols++;
 
     numOfRows++;
@@ -188,20 +233,4 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
   pShow->numOfReads += numOfRows;
 
   return numOfRows;
-}
-
-void mgmtGetMnodePrivateIpList(SRpcIpSet *ipSet) {
-  ipSet->inUse = 0;
-  ipSet->port = htons(tsMnodeDnodePort);
-  ipSet->numOfIps = 1;
-  ipSet->ip[0] = htonl(tsMnodeObj.privateIp);
-}
-
-void mgmtGetMnodePublicIpList(SRpcIpSet *ipSet) {
-  ipSet->inUse = 0;
-  ipSet->port = htons(tsMnodeDnodePort);
-  ipSet->numOfIps = 1;
-  ipSet->ip[0] = htonl(tsMnodeObj.publicIp);
-}
-
-#endif
\ No newline at end of file
+}
\ No newline at end of file

src/mnode/src/mgmtProfile.c

@@ -19,7 +19,7 @@
 #include "taccount.h"
 #include "tcluster.h"
 #include "mgmtDb.h"
-#include "mgmtMnode.h"
+#include "mpeer.h"
 #include "mgmtProfile.h"
 #include "mgmtShell.h"
 #include "mgmtTable.h"
@@ -681,8 +681,7 @@ int32_t mgmtRetrieveConns(SShowObj *pShow, char *data, int32_t rows, void *pConn
 
 void mgmtProcessKillQueryMsg(SQueuedMsg *pMsg) {
   SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
+  
   SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle, NULL);
   if (pUser == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_USER;
@@ -705,8 +704,7 @@ void mgmtProcessKillQueryMsg(SQueuedMsg *pMsg) {
 
 void mgmtProcessKillStreamMsg(SQueuedMsg *pMsg) {
   SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
+  
   SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle, NULL);
   if (pUser == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_USER;
@@ -729,8 +727,7 @@ void mgmtProcessKillStreamMsg(SQueuedMsg *pMsg) {
 
 void mgmtProcessKillConnectionMsg(SQueuedMsg *pMsg) {
   SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
+  
   SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle, NULL);
   if (pUser == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_USER;

src/mnode/src/mgmtShell.c

@@ -27,7 +27,7 @@
 #include "mgmtDb.h"
 #include "tcluster.h"
 #include "tgrant.h"
-#include "mgmtMnode.h"
+#include "mpeer.h"
 #include "mgmtProfile.h"
 #include "mgmtSdb.h"
 #include "mgmtShell.h"
@@ -142,14 +142,14 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
     return;
   }
 
-  if (mgmtCheckRedirect(rpcMsg->handle)) {
+  if (mpeerCheckRedirect()) {
     // rpcSendRedirectRsp(rpcMsg->handle, mgmtGetMnodeIpListForRedirect());
     mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_NO_MASTER);
     rpcFreeCont(rpcMsg->pCont);
     return;
   }
 
-  if (!mgmtInServerStatus()) {
+  if (!mpeerInServerStatus()) {
     mgmtProcessMsgWhileNotReady(rpcMsg);
     rpcFreeCont(rpcMsg->pCont);
     return;
@@ -337,9 +337,9 @@ static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
   }
 
   if (pMsg->usePublicIp) {
-    mgmtGetMnodePublicIpList(&pHBRsp->ipList);
+    mpeerGetPublicIpList(&pHBRsp->ipList);
   } else {
-    mgmtGetMnodePrivateIpList(&pHBRsp->ipList);
+    mpeerGetPrivateIpList(&pHBRsp->ipList);
   }
 
   /*
@@ -423,9 +423,9 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
   pConnectRsp->superAuth = pUser->superAuth;
 
   if (pMsg->usePublicIp) {
-    mgmtGetMnodePublicIpList(&pConnectRsp->ipList);
+    mpeerGetPublicIpList(&pConnectRsp->ipList);
   } else {
-    mgmtGetMnodePrivateIpList(&pConnectRsp->ipList);
+    mpeerGetPrivateIpList(&pConnectRsp->ipList);
   }
 
 connect_over:

src/mnode/src/mgmtTable.c

@@ -35,7 +35,7 @@
 #include "tcluster.h"
 #include "mgmtDServer.h"
 #include "tgrant.h"
-#include "mgmtMnode.h"
+#include "mpeer.h"
 #include "mgmtProfile.h"
 #include "mgmtSdb.h"
 #include "mgmtShell.h"
@@ -174,6 +174,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOperDesc *pOper) {
 }
 
 static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
+  const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS;
   SChildTableObj *pTable = pOper->pObj;
   assert(pTable != NULL && pOper->rowData != NULL);
 
@@ -182,7 +183,7 @@ static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
     pOper->rowSize = tsChildTableUpdateSize;
   } else {
     int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
-    if (pOper->maxRowSize < tsChildTableUpdateSize + schemaSize) {
+    if (maxRowSize < tsChildTableUpdateSize + schemaSize) {
       return TSDB_CODE_INVALID_MSG_LEN;
     }
     memcpy(pOper->rowData, pTable, tsChildTableUpdateSize);
@@ -224,35 +225,11 @@ static int32_t mgmtChildTableActionDecode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtInitChildTables() {
+static int32_t mgmtChildTableActionUpdateAll() {
   void *pNode = NULL;
   void *pLastNode = NULL;
   SChildTableObj *pTable = NULL;
 
-  SChildTableObj tObj;
-  tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
-
-  SSdbTableDesc tableDesc = {
-    .tableName    = "ctables",
-    .hashSessions = tsMaxTables,
-    .maxRowSize   = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
-    .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
-    .keyType      = SDB_KEY_TYPE_STRING,
-    .insertFp     = mgmtChildTableActionInsert,
-    .deleteFp     = mgmtChildTableActionDelete,
-    .updateFp     = mgmtChildTableActionUpdate,
-    .encodeFp     = mgmtChildTableActionEncode,
-    .decodeFp     = mgmtChildTableActionDecode,
-    .destroyFp    = mgmtChildTableActionDestroy,
-  };
-
-  tsChildTableSdb = sdbOpenTable(&tableDesc);
-  if (tsChildTableSdb == NULL) {
-    mError("failed to init child table data");
-    return -1;
-  }
-
-  pNode = NULL;
   while (1) {
     pLastNode = pNode;
     mgmtDecTableRef(pTable);
@@ -263,7 +240,7 @@ static int32_t mgmtInitChildTables() {
     if (pDb == NULL) {
       mError("ctable:%s, failed to get db, discard it", pTable->info.tableId);
       SSdbOperDesc desc = {0};
-      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
       sdbDeleteRow(&desc);
@@ -277,7 +254,7 @@ static int32_t mgmtInitChildTables() {
       mError("ctable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->info.tableId, pTable->vgId, pTable->sid);
       pTable->vgId = 0;
       SSdbOperDesc desc = {0};
-      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
       sdbDeleteRow(&desc);
@@ -291,7 +268,7 @@ static int32_t mgmtInitChildTables() {
              pTable->info.tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
       pTable->vgId = 0;
       SSdbOperDesc desc = {0};
-      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
       sdbDeleteRow(&desc);
@@ -303,7 +280,7 @@ static int32_t mgmtInitChildTables() {
       mError("ctable:%s, vgroup:%d tableList is null", pTable->info.tableId, pTable->vgId);
       pTable->vgId = 0;
       SSdbOperDesc desc = {0};
-      desc.type = SDB_OPER_TYPE_LOCAL;
+      desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
       sdbDeleteRow(&desc);
@@ -317,7 +294,7 @@ static int32_t mgmtInitChildTables() {
         mError("ctable:%s, stable:%s not exist", pTable->info.tableId, pTable->superTableId);
         pTable->vgId = 0;
         SSdbOperDesc desc = {0};
-        desc.type = SDB_OPER_TYPE_LOCAL;
+        desc.type = SDB_OPER_LOCAL;
         desc.pObj = pTable;
         desc.table = tsChildTableSdb;
         sdbDeleteRow(&desc);
@@ -328,6 +305,35 @@ static int32_t mgmtInitChildTables() {
     }
   }
 
+  return 0;
+}
+
+static int32_t mgmtInitChildTables() {
+  SChildTableObj tObj;
+  tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_CTABLE,
+    .tableName    = "ctables",
+    .hashSessions = tsMaxTables,
+    .maxRowSize   = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS,
+    .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
+    .keyType      = SDB_KEY_STRING,
+    .insertFp     = mgmtChildTableActionInsert,
+    .deleteFp     = mgmtChildTableActionDelete,
+    .updateFp     = mgmtChildTableActionUpdate,
+    .encodeFp     = mgmtChildTableActionEncode,
+    .decodeFp     = mgmtChildTableActionDecode,
+    .destroyFp    = mgmtChildTableActionDestroy,
+    .updateAllFp  = mgmtChildTableActionUpdateAll
+  };
+
+  tsChildTableSdb = sdbOpenTable(&tableDesc);
+  if (tsChildTableSdb == NULL) {
+    mError("failed to init child table data");
+    return -1;
+  }
+
   mTrace("child table is initialized");
   return 0;
 }
@@ -374,12 +380,14 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOperDesc *pOper) {
 }
 
 static int32_t mgmtSuperTableActionEncode(SSdbOperDesc *pOper) {
+  const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS;
+
   SSuperTableObj *pStable = pOper->pObj;
   assert(pOper->pObj != NULL && pOper->rowData != NULL);
 
   int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
 
-  if (pOper->maxRowSize < tsSuperTableUpdateSize + schemaSize) {
+  if (maxRowSize < tsSuperTableUpdateSize + schemaSize) {
     return TSDB_CODE_INVALID_MSG_LEN;
   }
 
@@ -411,22 +419,28 @@ static int32_t mgmtSuperTableActionDecode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t mgmtSuperTableActionUpdateAll() {
+  return 0;
+}
+
 static int32_t mgmtInitSuperTables() {
   SSuperTableObj tObj;
   tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
 
   SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_STABLE,
     .tableName    = "stables",
     .hashSessions = TSDB_MAX_SUPER_TABLES,
     .maxRowSize   = tsSuperTableUpdateSize + sizeof(SSchema) * TSDB_MAX_COLUMNS,
     .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
-    .keyType      = SDB_KEY_TYPE_STRING,
+    .keyType      = SDB_KEY_STRING,
     .insertFp     = mgmtSuperTableActionInsert,
     .deleteFp     = mgmtSuperTableActionDelete,
     .updateFp     = mgmtSuperTableActionUpdate,
     .encodeFp     = mgmtSuperTableActionEncode,
     .decodeFp     = mgmtSuperTableActionDecode,
     .destroyFp    = mgmtSuperTableActionDestroy,
+    .updateAllFp  = mgmtSuperTableActionUpdateAll
   };
 
   tsSuperTableSdb = sdbOpenTable(&tableDesc);
@@ -662,7 +676,7 @@ static void mgmtProcessCreateSuperTableMsg(SQueuedMsg *pMsg) {
   }
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = sizeof(SSuperTableObj) + schemaSize
@@ -686,7 +700,7 @@ static void mgmtProcessDropSuperTableMsg(SQueuedMsg *pMsg) {
     mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_OTHERS);
   } else {
     SSdbOperDesc oper = {
-      .type = SDB_OPER_TYPE_GLOBAL,
+      .type = SDB_OPER_GLOBAL,
       .table = tsSuperTableSdb,
       .pObj = pStable
     };
@@ -737,7 +751,7 @@ static int32_t mgmtAddSuperTableTag(SSuperTableObj *pStable, SSchema schema[], i
   pStable->sversion++;
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = tsSuperTableUpdateSize
@@ -768,7 +782,7 @@ static int32_t mgmtDropSuperTableTag(SSuperTableObj *pStable, char *tagName) {
   pStable->schema = realloc(pStable->schema, schemaSize);
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = tsSuperTableUpdateSize
@@ -803,7 +817,7 @@ static int32_t mgmtModifySuperTableTagName(SSuperTableObj *pStable, char *oldTag
   strncpy(schema->name, newTagName, TSDB_COL_NAME_LEN);
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = tsSuperTableUpdateSize
@@ -862,7 +876,7 @@ static int32_t mgmtAddSuperTableColumn(SDbObj *pDb, SSuperTableObj *pStable, SSc
   }
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = tsSuperTableUpdateSize
@@ -899,7 +913,7 @@ static int32_t mgmtDropSuperTableColumn(SDbObj *pDb, SSuperTableObj *pStable, ch
   }
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
     .rowSize = tsSuperTableUpdateSize
@@ -1047,7 +1061,7 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
 
     if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_LOCAL,
+        .type = SDB_OPER_LOCAL,
         .table = tsSuperTableSdb,
         .pObj = pTable,
       };
@@ -1262,7 +1276,7 @@ static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj
   }
   
   SSdbOperDesc desc = {0};
-  desc.type = SDB_OPER_TYPE_GLOBAL;
+  desc.type = SDB_OPER_GLOBAL;
   desc.pObj = pTable;
   desc.table = tsChildTableSdb;
   
@@ -1416,7 +1430,7 @@ static int32_t mgmtAddNormalTableColumn(SDbObj *pDb, SChildTableObj *pTable, SSc
   }
  
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable,
     .rowSize = tsChildTableUpdateSize
@@ -1450,7 +1464,7 @@ static int32_t mgmtDropNormalTableColumn(SDbObj *pDb, SChildTableObj *pTable, ch
   }
  
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable,
     .rowSize = tsChildTableUpdateSize
@@ -1592,7 +1606,7 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
 
     if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_LOCAL,
+        .type = SDB_OPER_LOCAL,
         .table = tsChildTableSdb,
         .pObj = pTable,
       };
@@ -1621,7 +1635,7 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
 
     if (pTable->superTable == pStable) {
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_LOCAL,
+        .type = SDB_OPER_LOCAL,
         .table = tsChildTableSdb,
         .pObj = pTable,
       };
@@ -1650,8 +1664,6 @@ static SChildTableObj* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_
 }
 
 static void mgmtProcessTableCfgMsg(SRpcMsg *rpcMsg) {
-  if (mgmtCheckRedirect(rpcMsg->handle)) return;
-
   SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) rpcMsg->pCont;
   pCfg->dnode = htonl(pCfg->dnode);
   pCfg->vnode = htonl(pCfg->vnode);
@@ -1712,7 +1724,7 @@ static void mgmtProcessDropChildTableRsp(SRpcMsg *rpcMsg) {
   }
   
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable
   };
@@ -1755,7 +1767,7 @@ static void mgmtProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
              queueMsg->thandle, tstrerror(rpcMsg->code));
       
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_GLOBAL,
+        .type = SDB_OPER_GLOBAL,
         .table = tsChildTableSdb,
         .pObj = pTable
       };

src/mnode/src/mgmtUser.c

@@ -20,7 +20,7 @@
 #include "tutil.h"
 #include "taccount.h"
 #include "tgrant.h"
-#include "mgmtMnode.h"
+#include "mpeer.h"
 #include "mgmtSdb.h"
 #include "mgmtShell.h"
 #include "mgmtUser.h"
@@ -70,14 +70,9 @@ static int32_t mgmtUserActionUpdate(SSdbOperDesc *pOper) {
 
 static int32_t mgmtUserActionEncode(SSdbOperDesc *pOper) {
   SUserObj *pUser = pOper->pObj;
-
-  if (pOper->maxRowSize < tsUserUpdateSize) {
-    return -1;
-  } else {
-    memcpy(pOper->rowData, pUser, tsUserUpdateSize);
-    pOper->rowSize = tsUserUpdateSize;
-    return TSDB_CODE_SUCCESS;
-  }
+  memcpy(pOper->rowData, pUser, tsUserUpdateSize);
+  pOper->rowSize = tsUserUpdateSize;
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
@@ -89,22 +84,34 @@ static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t mgmtUserActionUpdateAll() {
+  SAcctObj *pAcct = acctGetAcct("root");
+  mgmtCreateUser(pAcct, "root", "taosdata");
+  mgmtCreateUser(pAcct, "monitor", tsInternalPass);
+  mgmtCreateUser(pAcct, "_root", tsInternalPass);
+  acctReleaseAcct(pAcct);
+
+  return 0;
+}
+
 int32_t mgmtInitUsers() {
   SUserObj tObj;
   tsUserUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
 
   SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_USER,
     .tableName    = "users",
     .hashSessions = TSDB_MAX_USERS,
     .maxRowSize   = tsUserUpdateSize,
     .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
-    .keyType      = SDB_KEY_TYPE_STRING,
+    .keyType      = SDB_KEY_STRING,
     .insertFp     = mgmtUserActionInsert,
     .deleteFp     = mgmtUserActionDelete,
     .updateFp     = mgmtUserActionUpdate,
     .encodeFp     = mgmtUserActionEncode,
     .decodeFp     = mgmtUserActionDecode,
     .destroyFp    = mgmtUserActionDestroy,
+    .updateAllFp  = mgmtUserActionUpdateAll
   };
 
   tsUserSdb = sdbOpenTable(&tableDesc);
@@ -113,12 +120,6 @@ int32_t mgmtInitUsers() {
     return -1;
   }
 
-  SAcctObj *pAcct = acctGetAcct("root");
-  mgmtCreateUser(pAcct, "root", "taosdata");
-  mgmtCreateUser(pAcct, "monitor", tsInternalPass);
-  mgmtCreateUser(pAcct, "_root", tsInternalPass);
-  acctReleaseAcct(pAcct);
-
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_USER, mgmtProcessCreateUserMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_USER, mgmtProcessAlterUserMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_USER, mgmtProcessDropUserMsg);
@@ -143,7 +144,7 @@ void mgmtReleaseUser(SUserObj *pUser) {
 
 static int32_t mgmtUpdateUser(SUserObj *pUser) {
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser,
     .rowSize = tsUserUpdateSize
@@ -191,7 +192,7 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
   }
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser,
     .rowSize = sizeof(SUserObj)
@@ -208,7 +209,7 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
 
 static int32_t mgmtDropUser(SUserObj *pUser) {
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser
   };
@@ -314,8 +315,6 @@ SUserObj *mgmtGetUserFromConn(void *pConn, bool *usePublicIp) {
 }
 
 static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg) {
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
   int32_t code;
   SUserObj *pUser = pMsg->pUser;
   
@@ -333,8 +332,6 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg) {
 }
 
 static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
   int32_t code;
   SUserObj *pOperUser = pMsg->pUser;
   
@@ -427,8 +424,6 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
 }
 
 static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
-  if (mgmtCheckRedirect(pMsg->thandle)) return;
-
   int32_t code;
   SUserObj *pOperUser = pMsg->pUser;
 
@@ -488,7 +483,7 @@ void  mgmtDropAllUsers(SAcctObj *pAcct)  {
 
     if (strncmp(pUser->acct, pAcct->user, acctNameLen) == 0) {
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_LOCAL,
+        .type = SDB_OPER_LOCAL,
         .table = tsUserSdb,
         .pObj = pUser,
       };

src/mnode/src/mgmtVgroup.c

@@ -24,7 +24,7 @@
 #include "mgmtDb.h"
 #include "mgmtDClient.h"
 #include "mgmtDServer.h"
-#include "mgmtMnode.h"
+#include "mpeer.h"
 #include "mgmtProfile.h"
 #include "mgmtSdb.h"
 #include "mgmtShell.h"
@@ -138,13 +138,9 @@ static int32_t mgmtVgroupActionUpdate(SSdbOperDesc *pOper) {
 
 static int32_t mgmtVgroupActionEncode(SSdbOperDesc *pOper) {
   SVgObj *pVgroup = pOper->pObj;
-  if (pOper->maxRowSize < tsVgUpdateSize) {
-    return -1;
-  } else {
-    memcpy(pOper->rowData, pVgroup, tsVgUpdateSize);
-    pOper->rowSize = tsVgUpdateSize;
-    return TSDB_CODE_SUCCESS;
-  }
+  memcpy(pOper->rowData, pVgroup, tsVgUpdateSize);
+  pOper->rowSize = tsVgUpdateSize;
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t mgmtVgroupActionDecode(SSdbOperDesc *pOper) {
@@ -156,22 +152,28 @@ static int32_t mgmtVgroupActionDecode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t mgmtVgroupActionUpdateAll() {
+  return 0;
+}
+
 int32_t mgmtInitVgroups() {
   SVgObj tObj;
   tsVgUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
 
   SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_VGROUP,
     .tableName    = "vgroups",
     .hashSessions = TSDB_MAX_VGROUPS,
     .maxRowSize   = tsVgUpdateSize,
     .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
-    .keyType      = SDB_KEY_TYPE_AUTO,
+    .keyType      = SDB_KEY_AUTO,
     .insertFp     = mgmtVgroupActionInsert,
     .deleteFp     = mgmtVgroupActionDelete,
     .updateFp     = mgmtVgroupActionUpdate,
     .encodeFp     = mgmtVgroupActionEncode,
     .decodeFp     = mgmtVgroupActionDecode,
     .destroyFp    = mgmtVgroupActionDestroy,
+    .updateAllFp  = mgmtVgroupActionUpdateAll,
   };
 
   tsVgroupSdb = sdbOpenTable(&tableDesc);
@@ -187,6 +189,7 @@ int32_t mgmtInitVgroups() {
   mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_CONFIG_VNODE, mgmtProcessVnodeCfgMsg);
 
   mTrace("vgroup is initialized");
+  
   return 0;
 }
 
@@ -200,7 +203,7 @@ SVgObj *mgmtGetVgroup(int32_t vgId) {
 
 void mgmtUpdateVgroup(SVgObj *pVgroup) {
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup,
     .rowSize = tsVgUpdateSize
@@ -244,7 +247,7 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg, SDbObj *pDb) {
   }
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup,
     .rowSize = sizeof(SVgObj)
@@ -276,7 +279,7 @@ void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle) {
     mTrace("vgroup:%d, replica:%d is deleting from sdb", pVgroup->vgId, pVgroup->numOfVnodes);
     mgmtSendDropVgroupMsg(pVgroup, NULL);
     SSdbOperDesc oper = {
-      .type = SDB_OPER_TYPE_GLOBAL,
+      .type = SDB_OPER_GLOBAL,
       .table = tsVgroupSdb,
       .pObj = pVgroup
     };
@@ -583,7 +586,7 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
     mgmtAddToShellQueue(newMsg);
   } else {
     SSdbOperDesc oper = {
-      .type = SDB_OPER_TYPE_GLOBAL,
+      .type = SDB_OPER_GLOBAL,
       .table = tsVgroupSdb,
       .pObj = pVgroup
     };
@@ -646,7 +649,7 @@ static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
   if (queueMsg->received != queueMsg->expected) return;
 
   SSdbOperDesc oper = {
-    .type = SDB_OPER_TYPE_GLOBAL,
+    .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup
   };
@@ -663,8 +666,6 @@ static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
 }
 
 static void mgmtProcessVnodeCfgMsg(SRpcMsg *rpcMsg) {
-  if (mgmtCheckRedirect(rpcMsg->handle)) return;
-
   SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) rpcMsg->pCont;
   pCfg->dnodeId = htonl(pCfg->dnodeId);
   pCfg->vgId    = htonl(pCfg->vgId);
@@ -705,7 +706,7 @@ void mgmtDropAllVgroups(SDbObj *pDropDb) {
 
     if (strncmp(pDropDb->name, pVgroup->dbName, dbNameLen) == 0) {
       SSdbOperDesc oper = {
-        .type = SDB_OPER_TYPE_LOCAL,
+        .type = SDB_OPER_LOCAL,
         .table = tsVgroupSdb,
         .pObj = pVgroup,
       };

src/plugins/http/inc/cJSON.h

-/*
-  Copyright (c) 2009-2017 Dave Gamble and cJSON contributors
-
-  Permission is hereby granted, free of charge, to any person obtaining a copy
-  of this software and associated documentation files (the "Software"), to deal
-  in the Software without restriction, including without limitation the rights
-  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-  copies of the Software, and to permit persons to whom the Software is
-  furnished to do so, subject to the following conditions:
-
-  The above copyright notice and this permission notice shall be included in
-  all copies or substantial portions of the Software.
-
-  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-  THE SOFTWARE.
-*/
-
-#ifndef cJSON__h
-#define cJSON__h
-
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
-/* project version */
-#define CJSON_VERSION_MAJOR 1
-#define CJSON_VERSION_MINOR 5
-#define CJSON_VERSION_PATCH 9
-
-#include <stddef.h>
-#include <stdint.h>
-
-/* cJSON Types: */
-#define cJSON_Invalid (0)
-#define cJSON_False  (1 << 0)
-#define cJSON_True   (1 << 1)
-#define cJSON_NULL   (1 << 2)
-#define cJSON_Number (1 << 3)
-#define cJSON_String (1 << 4)
-#define cJSON_Array  (1 << 5)
-#define cJSON_Object (1 << 6)
-#define cJSON_Raw    (1 << 7) /* raw json */
-
-#define cJSON_IsReference 256
-#define cJSON_StringIsConst 512
-
-/* The cJSON structure: */
-typedef struct cJSON
-{
-    /* next/prev allow you to walk array/object chains. Alternatively, use GetArraySize/GetArrayItem/GetObjectItem */
-    struct cJSON *next;
-    struct cJSON *prev;
-    /* An array or object item will have a child pointer pointing to a chain of the items in the array/object. */
-    struct cJSON *child;
-
-    /* The type of the item, as above. */
-    int type;
-
-    /* The item's string, if type==cJSON_String  and type == cJSON_Raw */
-    char *valuestring;
-    /* writing to valueint is DEPRECATED, use cJSON_SetNumberValue instead */
-    int64_t valueint;
-    /* The item's number, if type==cJSON_Number */
-    double valuedouble;
-
-    /* The item's name string, if this item is the child of, or is in the list of subitems of an object. */
-    char *string;
-} cJSON;
-
-typedef struct cJSON_Hooks
-{
-      void *(*malloc_fn)(size_t sz);
-      void (*free_fn)(void *ptr);
-} cJSON_Hooks;
-
-typedef int cJSON_bool;
-
-#if !defined(__WINDOWS__) && (defined(WIN32) || defined(WIN64) || defined(_MSC_VER) || defined(_WIN32))
-#define __WINDOWS__
-#endif
-#ifdef __WINDOWS__
-
-/* When compiling for windows, we specify a specific calling convention to avoid issues where we are being called from a project with a different default calling convention.  For windows you have 2 define options:
-
-CJSON_HIDE_SYMBOLS - Define this in the case where you don't want to ever dllexport symbols
-CJSON_EXPORT_SYMBOLS - Define this on library build when you want to dllexport symbols (default)
-CJSON_IMPORT_SYMBOLS - Define this if you want to dllimport symbol
-
-For *nix builds that support visibility attribute, you can define similar behavior by
-
-setting default visibility to hidden by adding
--fvisibility=hidden (for gcc)
-or
--xldscope=hidden (for sun cc)
-to CFLAGS
-
-then using the CJSON_API_VISIBILITY flag to "export" the same symbols the way CJSON_EXPORT_SYMBOLS does
-
-*/
-
-/* export symbols by default, this is necessary for copy pasting the C and header file */
-#if !defined(CJSON_HIDE_SYMBOLS) && !defined(CJSON_IMPORT_SYMBOLS) && !defined(CJSON_EXPORT_SYMBOLS)
-#define CJSON_EXPORT_SYMBOLS
-#endif
-
-#if defined(CJSON_HIDE_SYMBOLS)
-#define CJSON_PUBLIC(type)   type __stdcall
-#elif defined(CJSON_EXPORT_SYMBOLS)
-#define CJSON_PUBLIC(type)   __declspec(dllexport) type __stdcall
-#elif defined(CJSON_IMPORT_SYMBOLS)
-#define CJSON_PUBLIC(type)   __declspec(dllimport) type __stdcall
-#endif
-#else /* !WIN32 */
-#if (defined(__GNUC__) || defined(__SUNPRO_CC) || defined (__SUNPRO_C)) && defined(CJSON_API_VISIBILITY)
-#define CJSON_PUBLIC(type)   __attribute__((visibility("default"))) type
-#else
-#define CJSON_PUBLIC(type) type
-#endif
-#endif
-
-/* Limits how deeply nested arrays/objects can be before cJSON rejects to parse them.
- * This is to prevent stack overflows. */
-#ifndef CJSON_NESTING_LIMIT
-#define CJSON_NESTING_LIMIT 1000
-#endif
-
-/* returns the version of cJSON as a string */
-CJSON_PUBLIC(const char*) cJSON_Version(void);
-
-/* Supply malloc, realloc and free functions to cJSON */
-CJSON_PUBLIC(void) cJSON_InitHooks(cJSON_Hooks* hooks);
-
-/* Memory Management: the caller is always responsible to free the results from all variants of cJSON_Parse (with cJSON_Delete) and cJSON_Print (with stdlib free, cJSON_Hooks.free_fn, or cJSON_free as appropriate). The exception is cJSON_PrintPreallocated, where the caller has full responsibility of the buffer. */
-/* Supply a block of JSON, and this returns a cJSON object you can interrogate. */
-CJSON_PUBLIC(cJSON *) cJSON_Parse(const char *value);
-/* ParseWithOpts allows you to require (and check) that the JSON is null terminated, and to retrieve the pointer to the final byte parsed. */
-/* If you supply a ptr in return_parse_end and parsing fails, then return_parse_end will contain a pointer to the error so will match cJSON_GetErrorPtr(). */
-CJSON_PUBLIC(cJSON *) cJSON_ParseWithOpts(const char *value, const char **return_parse_end, cJSON_bool require_null_terminated);
-
-/* Render a cJSON entity to text for transfer/storage. */
-CJSON_PUBLIC(char *) cJSON_Print(const cJSON *item);
-/* Render a cJSON entity to text for transfer/storage without any formatting. */
-CJSON_PUBLIC(char *) cJSON_PrintUnformatted(const cJSON *item);
-/* Render a cJSON entity to text using a buffered strategy. prebuffer is a guess at the final size. guessing well reduces reallocation. fmt=0 gives unformatted, =1 gives formatted */
-CJSON_PUBLIC(char *) cJSON_PrintBuffered(const cJSON *item, int prebuffer, cJSON_bool fmt);
-/* Render a cJSON entity to text using a buffer already allocated in memory with given length. Returns 1 on success and 0 on failure. */
-/* NOTE: cJSON is not always 100% accurate in estimating how much memory it will use, so to be safe allocate 5 bytes more than you actually need */
-CJSON_PUBLIC(cJSON_bool) cJSON_PrintPreallocated(cJSON *item, char *buffer, const int length, const cJSON_bool format);
-/* Delete a cJSON entity and all subentities. */
-CJSON_PUBLIC(void) cJSON_Delete(cJSON *c);
-
-/* Returns the number of items in an array (or object). */
-CJSON_PUBLIC(int) cJSON_GetArraySize(const cJSON *array);
-/* Retrieve item number "item" from array "array". Returns NULL if unsuccessful. */
-CJSON_PUBLIC(cJSON *) cJSON_GetArrayItem(const cJSON *array, int index);
-/* Get item "string" from object. Case insensitive. */
-CJSON_PUBLIC(cJSON *) cJSON_GetObjectItem(const cJSON * const object, const char * const string);
-CJSON_PUBLIC(cJSON *) cJSON_GetObjectItemCaseSensitive(const cJSON * const object, const char * const string);
-CJSON_PUBLIC(cJSON_bool) cJSON_HasObjectItem(const cJSON *object, const char *string);
-/* For analysing failed parses. This returns a pointer to the parse error. You'll probably need to look a few chars back to make sense of it. Defined when cJSON_Parse() returns 0. 0 when cJSON_Parse() succeeds. */
-CJSON_PUBLIC(const char *) cJSON_GetErrorPtr(void);
-
-/* These functions check the type of an item */
-CJSON_PUBLIC(cJSON_bool) cJSON_IsInvalid(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsFalse(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsTrue(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsBool(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsNull(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsNumber(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsString(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsArray(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsObject(const cJSON * const item);
-CJSON_PUBLIC(cJSON_bool) cJSON_IsRaw(const cJSON * const item);
-
-/* These calls create a cJSON item of the appropriate type. */
-CJSON_PUBLIC(cJSON *) cJSON_CreateNull(void);
-CJSON_PUBLIC(cJSON *) cJSON_CreateTrue(void);
-CJSON_PUBLIC(cJSON *) cJSON_CreateFalse(void);
-CJSON_PUBLIC(cJSON *) cJSON_CreateBool(cJSON_bool boolean);
-CJSON_PUBLIC(cJSON *) cJSON_CreateNumber(double num);
-CJSON_PUBLIC(cJSON *) cJSON_CreateString(const char *string);
-/* raw json */
-CJSON_PUBLIC(cJSON *) cJSON_CreateRaw(const char *raw);
-CJSON_PUBLIC(cJSON *) cJSON_CreateArray(void);
-CJSON_PUBLIC(cJSON *) cJSON_CreateObject(void);
-
-/* These utilities create an Array of count items. */
-CJSON_PUBLIC(cJSON *) cJSON_CreateIntArray(const int *numbers, int count);
-CJSON_PUBLIC(cJSON *) cJSON_CreateFloatArray(const float *numbers, int count);
-CJSON_PUBLIC(cJSON *) cJSON_CreateDoubleArray(const double *numbers, int count);
-CJSON_PUBLIC(cJSON *) cJSON_CreateStringArray(const char **strings, int count);
-
-/* Append item to the specified array/object. */
-CJSON_PUBLIC(void) cJSON_AddItemToArray(cJSON *array, cJSON *item);
-CJSON_PUBLIC(void) cJSON_AddItemToObject(cJSON *object, const char *string, cJSON *item);
-/* Use this when string is definitely const (i.e. a literal, or as good as), and will definitely survive the cJSON object.
- * WARNING: When this function was used, make sure to always check that (item->type & cJSON_StringIsConst) is zero before
- * writing to `item->string` */
-CJSON_PUBLIC(void) cJSON_AddItemToObjectCS(cJSON *object, const char *string, cJSON *item);
-/* Append reference to item to the specified array/object. Use this when you want to add an existing cJSON to a new cJSON, but don't want to corrupt your existing cJSON. */
-CJSON_PUBLIC(void) cJSON_AddItemReferenceToArray(cJSON *array, cJSON *item);
-CJSON_PUBLIC(void) cJSON_AddItemReferenceToObject(cJSON *object, const char *string, cJSON *item);
-
-/* Remove/Detatch items from Arrays/Objects. */
-CJSON_PUBLIC(cJSON *) cJSON_DetachItemViaPointer(cJSON *parent, cJSON * const item);
-CJSON_PUBLIC(cJSON *) cJSON_DetachItemFromArray(cJSON *array, int which);
-CJSON_PUBLIC(void) cJSON_DeleteItemFromArray(cJSON *array, int which);
-CJSON_PUBLIC(cJSON *) cJSON_DetachItemFromObject(cJSON *object, const char *string);
-CJSON_PUBLIC(cJSON *) cJSON_DetachItemFromObjectCaseSensitive(cJSON *object, const char *string);
-CJSON_PUBLIC(void) cJSON_DeleteItemFromObject(cJSON *object, const char *string);
-CJSON_PUBLIC(void) cJSON_DeleteItemFromObjectCaseSensitive(cJSON *object, const char *string);
-
-/* Update array items. */
-CJSON_PUBLIC(void) cJSON_InsertItemInArray(cJSON *array, int which, cJSON *newitem); /* Shifts pre-existing items to the right. */
-CJSON_PUBLIC(cJSON_bool) cJSON_ReplaceItemViaPointer(cJSON * const parent, cJSON * const item, cJSON * replacement);
-CJSON_PUBLIC(void) cJSON_ReplaceItemInArray(cJSON *array, int which, cJSON *newitem);
-CJSON_PUBLIC(void) cJSON_ReplaceItemInObject(cJSON *object,const char *string,cJSON *newitem);
-CJSON_PUBLIC(void) cJSON_ReplaceItemInObjectCaseSensitive(cJSON *object,const char *string,cJSON *newitem);
-
-/* Duplicate a cJSON item */
-CJSON_PUBLIC(cJSON *) cJSON_Duplicate(const cJSON *item, cJSON_bool recurse);
-/* Duplicate will create a new, identical cJSON item to the one you pass, in new memory that will
-need to be released. With recurse!=0, it will duplicate any children connected to the item.
-The item->next and ->prev pointers are always zero on return from Duplicate. */
-/* Recursively compare two cJSON items for equality. If either a or b is NULL or invalid, they will be considered unequal.
- * case_sensitive determines if object keys are treated case sensitive (1) or case insensitive (0) */
-CJSON_PUBLIC(cJSON_bool) cJSON_Compare(const cJSON * const a, const cJSON * const b, const cJSON_bool case_sensitive);
-
-
-CJSON_PUBLIC(void) cJSON_Minify(char *json);
-
-/* Macros for creating things quickly. */
-#define cJSON_AddNullToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateNull())
-#define cJSON_AddTrueToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateTrue())
-#define cJSON_AddFalseToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateFalse())
-#define cJSON_AddBoolToObject(object,name,b) cJSON_AddItemToObject(object, name, cJSON_CreateBool(b))
-#define cJSON_AddNumberToObject(object,name,n) cJSON_AddItemToObject(object, name, cJSON_CreateNumber(n))
-#define cJSON_AddStringToObject(object,name,s) cJSON_AddItemToObject(object, name, cJSON_CreateString(s))
-#define cJSON_AddRawToObject(object,name,s) cJSON_AddItemToObject(object, name, cJSON_CreateRaw(s))
-
-/* When assigning an integer value, it needs to be propagated to valuedouble too. */
-#define cJSON_SetIntValue(object, number) ((object) ? (object)->valueint = (object)->valuedouble = (number) : (number))
-/* helper for the cJSON_SetNumberValue macro */
-CJSON_PUBLIC(double) cJSON_SetNumberHelper(cJSON *object, double number);
-#define cJSON_SetNumberValue(object, number) ((object != NULL) ? cJSON_SetNumberHelper(object, (double)number) : (number))
-
-/* Macro for iterating over an array or object */
-#define cJSON_ArrayForEach(element, array) for(element = (array != NULL) ? (array)->child : NULL; element != NULL; element = element->next)
-
-/* malloc/free objects using the malloc/free functions that have been set with cJSON_InitHooks */
-CJSON_PUBLIC(void *) cJSON_malloc(size_t size);
-CJSON_PUBLIC(void) cJSON_free(void *object);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif

src/plugins/http/inc/tgHandle.h

@@ -33,6 +33,7 @@
 #define TG_PASS_URL_POS   3
 
 void tgInitHandle(HttpServer *pServer);
+void tgCleanupHandle();
 
 bool tgProcessRquest(struct HttpContext *pContext);
 

src/plugins/http/src/httpSystem.c

@@ -122,12 +122,14 @@ void httpStopSystem() {
   if (httpServer != NULL) {
     httpServer->online = false;
   }
+  tgCleanupHandle();
 }
 
 void httpCleanUpSystem() {
   httpPrint("http service cleanup");
   httpStopSystem();
-#if 0
+
+#if 1
   if (httpServer == NULL) {
     return;
   }

src/plugins/http/src/tgHandle.c

@@ -116,6 +116,7 @@ void tgFreeSchemas() {
     }
     free(tgSchemas.schemas);
     tgSchemas.size = 0;
+    tgSchemas.schemas = NULL;
   }
 }
 
@@ -290,6 +291,10 @@ void tgInitHandle(HttpServer *pServer) {
   httpAddMethod(pServer, &tgDecodeMethod);
 }
 
+void tgCleanupHandle() {
+  tgFreeSchemas();
+}
+
 bool tgGetUserFromUrl(HttpContext *pContext) {
   HttpParser *pParser = &pContext->parser;
   if (pParser->path[TG_USER_URL_POS].len > TSDB_USER_LEN - 1 || pParser->path[TG_USER_URL_POS].len <= 0) {