[TD-998]

src/dnode/src/dnodeMgmt.c

@@ -492,6 +492,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
   pCfg->numOfVnodes  = htonl(pCfg->numOfVnodes);
   pCfg->moduleStatus = htonl(pCfg->moduleStatus);
   pCfg->dnodeId      = htonl(pCfg->dnodeId);
+  pCfg->clusterId    = htonl(pCfg->clusterId);
 
   for (int32_t i = 0; i < pMnodes->nodeNum; ++i) {
     SDMMnodeInfo *pMnodeInfo = &pMnodes->nodeInfos[i];
@@ -697,6 +698,7 @@ static void dnodeSendStatusMsg(void *handle, void *tmrId) {
   //strcpy(pStatus->dnodeName, tsDnodeName);
   pStatus->version          = htonl(tsVersion);
   pStatus->dnodeId          = htonl(tsDnodeCfg.dnodeId);
+  pStatus->clusterId        = htonl(tsDnodeCfg.clusterId);
   strcpy(pStatus->dnodeEp, tsLocalEp);
   pStatus->lastReboot       = htonl(tsRebootTime);
   pStatus->numOfCores       = htons((uint16_t) tsNumOfCores);
@@ -767,6 +769,13 @@ static bool dnodeReadDnodeCfg() {
   }
   tsDnodeCfg.dnodeId = dnodeId->valueint;
 
+  cJSON* clusterId = cJSON_GetObjectItem(root, "clusterId");
+  if (!clusterId || clusterId->type != cJSON_Number) {
+    dError("failed to read dnodeCfg.json, clusterId not found");
+    goto PARSE_CFG_OVER;
+  }
+  tsDnodeCfg.clusterId = clusterId->valueint;
+
   ret = true;
 
   dInfo("read numOfVnodes successed, dnodeId:%d", tsDnodeCfg.dnodeId);
@@ -791,6 +800,7 @@ static void dnodeSaveDnodeCfg() {
 
   len += snprintf(content + len, maxLen - len, "{\n");
   len += snprintf(content + len, maxLen - len, "  \"dnodeId\": %d\n", tsDnodeCfg.dnodeId);
+  len += snprintf(content + len, maxLen - len, "  \"clusterId\": %d\n", tsDnodeCfg.clusterId);
   len += snprintf(content + len, maxLen - len, "}\n"); 
 
   fwrite(content, 1, len, fp);
@@ -803,8 +813,9 @@ static void dnodeSaveDnodeCfg() {
 
 void dnodeUpdateDnodeCfg(SDMDnodeCfg *pCfg) {
   if (tsDnodeCfg.dnodeId == 0) {
-    dInfo("dnodeId is set to %d", pCfg->dnodeId);  
+    dInfo("dnodeId is set to %d, clusterId is set to %d", pCfg->dnodeId, pCfg->clusterId);  
     tsDnodeCfg.dnodeId = pCfg->dnodeId;
+    tsDnodeCfg.clusterId = pCfg->clusterId;
     dnodeSaveDnodeCfg();
   }
 }

src/inc/taoserror.h

@@ -131,6 +131,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_BALANCE_ENABLED,          0, 0x0337, "balance al
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_NOT_IN_DNODE,      0, 0x0338, "vgroup not in dnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_ALREADY_IN_DNODE,  0, 0x0339, "vgroup already in dnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_NOT_FREE,           0, 0x033A, "dnode not avaliable")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CLUSTER_ID,       0, 0x033B, "cluster id not match")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST,       0, 0x0340, "mnode accounts already exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT,             0, 0x0341, "mnode invalid account")

src/inc/taosmsg.h

@@ -547,6 +547,7 @@ typedef struct {
 
 typedef struct {
   int32_t  dnodeId;
+  int32_t  clusterId;
   uint32_t moduleStatus;
   uint32_t numOfVnodes;
   uint32_t reserved;
@@ -587,6 +588,7 @@ typedef struct {
   uint16_t    openVnodes;
   uint16_t    numOfCores;
   float       diskAvailable;  // GB
+  int32_t     clusterId;
   uint8_t     alternativeRole;
   uint8_t     reserve2[15];
   SClusterCfg clusterCfg;

src/mnode/src/mnodeCluster.c

@@ -77,6 +77,7 @@ static int32_t mnodeClusterActionRestored() {
     }
   }
 
+  mnodeUpdateClusterId();
   return TSDB_CODE_SUCCESS;
 }
 
@@ -141,7 +142,7 @@ static int32_t mnodeCreateCluster() {
   SClusterObj *pCluster = malloc(sizeof(SClusterObj));
   memset(pCluster, 0, sizeof(SClusterObj));
   pCluster->createdTime = taosGetTimestampMs();
-  pCluster->clusterId = (pCluster->createdTime >> 32) & (pCluster->createdTime) & (*(int32_t*)tsFirst);
+  pCluster->clusterId = abs(((pCluster->createdTime >> 32) & (pCluster->createdTime)) | (*(int32_t*)tsFirst));
   
   SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,

src/mnode/src/mnodeDnode.c

@@ -37,6 +37,7 @@
 #include "mnodeVgroup.h"
 #include "mnodeWrite.h"
 #include "mnodePeer.h"
+#include "mnodeCluster.h"
 
 int32_t tsAccessSquence = 0;
 static void   *tsDnodeSdb = NULL;
@@ -355,6 +356,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
   pStatus->moduleStatus = htonl(pStatus->moduleStatus);
   pStatus->lastReboot   = htonl(pStatus->lastReboot);
   pStatus->numOfCores   = htons(pStatus->numOfCores);
+  pStatus->clusterId    = htonl(pStatus->clusterId);
   
   uint32_t version = htonl(pStatus->version);
   if (version != tsVersion) {
@@ -382,13 +384,19 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
   pDnode->diskAvailable    = pStatus->diskAvailable;
   pDnode->alternativeRole  = pStatus->alternativeRole;
   pDnode->moduleStatus     = pStatus->moduleStatus;
-  
+
   if (pStatus->dnodeId == 0) {
-    mDebug("dnode:%d %s, first access", pDnode->dnodeId, pDnode->dnodeEp);
+    mDebug("dnode:%d %s, first access, set clusterId %d", pDnode->dnodeId, pDnode->dnodeEp, mnodeGetClusterId());
   } else {
-    mTrace("dnode:%d, status received, access times %d", pDnode->dnodeId, pDnode->lastAccess);
+    if (pStatus->clusterId != mnodeGetClusterId()) {
+      mError("dnode:%d, input clusterId %d not match with exist %d", pDnode->dnodeId, pStatus->clusterId,
+             mnodeGetClusterId());
+      return TSDB_CODE_MND_INVALID_CLUSTER_ID;
+    } else {
+      mTrace("dnode:%d, status received, access times %d", pDnode->dnodeId, pDnode->lastAccess);
+    }
   }
- 
+
   int32_t openVnodes = htons(pStatus->openVnodes);
   int32_t contLen = sizeof(SDMStatusRsp) + openVnodes * sizeof(SDMVgroupAccess);
   SDMStatusRsp *pRsp = rpcMallocCont(contLen);
@@ -400,6 +408,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
   pRsp->dnodeCfg.dnodeId = htonl(pDnode->dnodeId);
   pRsp->dnodeCfg.moduleStatus = htonl((int32_t)pDnode->isMgmt);
   pRsp->dnodeCfg.numOfVnodes = htonl(openVnodes);
+  pRsp->dnodeCfg.clusterId = htonl(mnodeGetClusterId());
   SDMVgroupAccess *pAccess = (SDMVgroupAccess *)((char *)pRsp + sizeof(SDMStatusRsp));
 
   for (int32_t j = 0; j < openVnodes; ++j) {