Merge pull request #2770 from taosdata/feature/vnode

Feature/vnode

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);
@@ -790,7 +799,8 @@ static void dnodeSaveDnodeCfg() {
   char *  content = calloc(1, maxLen + 1);
 
   len += snprintf(content + len, maxLen - len, "{\n");
-  len += snprintf(content + len, maxLen - len, "  \"dnodeId\": %d\n", tsDnodeCfg.dnodeId);
+  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/taosdef.h

@@ -377,6 +377,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_ORDER_ASC   1
 #define TSDB_ORDER_DESC  2
 
+#define TSDB_DEFAULT_CLUSTER_HASH_SIZE  1
 #define TSDB_DEFAULT_MNODES_HASH_SIZE   5
 #define TSDB_DEFAULT_DNODES_HASH_SIZE   10
 #define TSDB_DEFAULT_ACCOUNTS_HASH_SIZE 10

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

@@ -139,6 +139,7 @@ enum _mgmt_table {
   TSDB_MGMT_TABLE_GRANTS,
   TSDB_MGMT_TABLE_VNODES,
   TSDB_MGMT_TABLE_STREAMTABLES,
+  TSDB_MGMT_TABLE_CLUSTER,
   TSDB_MGMT_TABLE_MAX,
 };
 
@@ -546,6 +547,7 @@ typedef struct {
 
 typedef struct {
   int32_t  dnodeId;
+  int32_t  clusterId;
   uint32_t moduleStatus;
   uint32_t numOfVnodes;
   uint32_t reserved;
@@ -586,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/inc/mnodeCluster.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_MNODE_CLUSTER_H
+#define TDENGINE_MNODE_CLUSTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct SClusterObj;
+
+int32_t mnodeInitCluster();
+void    mnodeCleanupCluster();
+int32_t mnodeGetClusterId();
+void    mnodeUpdateClusterId();
+void *  mnodeGetCluster(int32_t clusterId);
+void *  mnodeGetNextCluster(void *pIter, struct SClusterObj **pCluster);
+void    mnodeIncClusterRef(struct SClusterObj *pCluster);
+void    mnodeDecClusterRef(struct SClusterObj *pCluster);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/mnode/inc/mnodeDef.h

@@ -36,6 +36,14 @@ struct define notes:
 3. The fields behind the updataEnd field can be changed;
 */
 
+typedef struct SClusterObj {
+  int32_t clusterId;
+  int64_t createdTime;
+  int8_t  reserved[36];
+  int8_t  updateEnd[4];
+  int32_t refCount;
+} SClusterObj;
+
 typedef struct SDnodeObj {
   int32_t    dnodeId;
   int32_t    openVnodes;

src/mnode/inc/mnodeSdb.h

@@ -23,15 +23,16 @@ extern "C" {
 struct SMnodeMsg;
 
 typedef enum {
-  SDB_TABLE_DNODE   = 0,
-  SDB_TABLE_MNODE   = 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
+  SDB_TABLE_CLUSTER = 0,
+  SDB_TABLE_DNODE   = 1,
+  SDB_TABLE_MNODE   = 2,
+  SDB_TABLE_ACCOUNT = 3,
+  SDB_TABLE_USER    = 4,
+  SDB_TABLE_DB      = 5,
+  SDB_TABLE_VGROUP  = 6,
+  SDB_TABLE_STABLE  = 7,
+  SDB_TABLE_CTABLE  = 8,
+  SDB_TABLE_MAX     = 9
 } ESdbTable;
 
 typedef enum {

src/mnode/src/mnodeAcct.c

@@ -64,7 +64,7 @@ static int32_t mnodeAcctActionUpdate(SSdbOper *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mnodeActionActionEncode(SSdbOper *pOper) {
+static int32_t mnodeAcctActionEncode(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   memcpy(pOper->rowData, pAcct, tsAcctUpdateSize);
   pOper->rowSize = tsAcctUpdateSize;
@@ -109,7 +109,7 @@ int32_t mnodeInitAccts() {
     .insertFp     = mnodeAcctActionInsert,
     .deleteFp     = mnodeAcctActionDelete,
     .updateFp     = mnodeAcctActionUpdate,
-    .encodeFp     = mnodeActionActionEncode,
+    .encodeFp     = mnodeAcctActionEncode,
     .decodeFp     = mnodeAcctActionDecode,
     .destroyFp    = mnodeAcctActionDestroy,
     .restoredFp   = mnodeAcctActionRestored

src/mnode/src/mnodeCluster.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "os.h"
+#include "taoserror.h"
+#include "ttime.h"
+#include "dnode.h"
+#include "mnodeDef.h"
+#include "mnodeInt.h"
+#include "mnodeCluster.h"
+#include "mnodeSdb.h"
+#include "mnodeShow.h"
+#include "tglobal.h"
+
+static void *  tsClusterSdb = NULL;
+static int32_t tsClusterUpdateSize;
+static int32_t tsClusterId;
+static int32_t mnodeCreateCluster();
+
+static int32_t mnodeGetClusterMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
+static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, void *pConn);
+
+static int32_t mnodeClusterActionDestroy(SSdbOper *pOper) {
+  tfree(pOper->pObj);
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionInsert(SSdbOper *pOper) {
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionDelete(SSdbOper *pOper) {
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionUpdate(SSdbOper *pOper) {
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionEncode(SSdbOper *pOper) {
+  SClusterObj *pCluster = pOper->pObj;
+  memcpy(pOper->rowData, pCluster, tsClusterUpdateSize);
+  pOper->rowSize = tsClusterUpdateSize;
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionDecode(SSdbOper *pOper) {
+  SClusterObj *pCluster = (SClusterObj *) calloc(1, sizeof(SClusterObj));
+  if (pCluster == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;
+
+  memcpy(pCluster, pOper->rowData, tsClusterUpdateSize);
+  pOper->pObj = pCluster;
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeClusterActionRestored() {
+  int32_t numOfRows = sdbGetNumOfRows(tsClusterSdb);
+  if (numOfRows <= 0 && dnodeIsFirstDeploy()) {
+    mInfo("dnode first deploy, create cluster");
+    int32_t code = mnodeCreateCluster();
+    if (code != TSDB_CODE_SUCCESS) {
+      mError("failed to create cluster, reason:%s", tstrerror(code));
+      return code;
+    }
+  }
+
+  mnodeUpdateClusterId();
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t mnodeInitCluster() {
+  SClusterObj tObj;
+  tsClusterUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;
+
+  SSdbTableDesc tableDesc = {
+    .tableId      = SDB_TABLE_CLUSTER,
+    .tableName    = "cluster",
+    .hashSessions = TSDB_DEFAULT_CLUSTER_HASH_SIZE,
+    .maxRowSize   = tsClusterUpdateSize,
+    .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
+    .keyType      = SDB_KEY_INT,
+    .insertFp     = mnodeClusterActionInsert,
+    .deleteFp     = mnodeClusterActionDelete,
+    .updateFp     = mnodeClusterActionUpdate,
+    .encodeFp     = mnodeClusterActionEncode,
+    .decodeFp     = mnodeClusterActionDecode,
+    .destroyFp    = mnodeClusterActionDestroy,
+    .restoredFp   = mnodeClusterActionRestored
+  };
+
+  tsClusterSdb = sdbOpenTable(&tableDesc);
+  if (tsClusterSdb == NULL) {
+    mError("table:%s, failed to create hash", tableDesc.tableName);
+    return -1;
+  }
+
+  mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeGetClusterMeta);
+  mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_CLUSTER, mnodeRetrieveClusters);
+
+  mDebug("table:%s, hash is created", tableDesc.tableName);
+  return TSDB_CODE_SUCCESS;
+}
+
+void mnodeCleanupCluster() {
+  sdbCloseTable(tsClusterSdb);
+  tsClusterSdb = NULL;
+}
+
+void *mnodeGetCluster(int32_t clusterId) {
+  return sdbGetRow(tsClusterSdb, &clusterId);
+}
+
+void *mnodeGetNextCluster(void *pIter, SClusterObj **pCluster) {
+  return sdbFetchRow(tsClusterSdb, pIter, (void **)pCluster); 
+}
+
+void mnodeIncClusterRef(SClusterObj *pCluster) {
+  sdbIncRef(tsClusterSdb, pCluster);
+}
+
+void mnodeDecClusterRef(SClusterObj *pCluster) {
+  sdbDecRef(tsClusterSdb, pCluster);
+}
+
+static int32_t mnodeCreateCluster() {
+  int32_t numOfClusters = sdbGetNumOfRows(tsClusterSdb);
+  if (numOfClusters != 0) return TSDB_CODE_SUCCESS;
+
+  SClusterObj *pCluster = malloc(sizeof(SClusterObj));
+  memset(pCluster, 0, sizeof(SClusterObj));
+  pCluster->createdTime = taosGetTimestampMs();
+  pCluster->clusterId = labs((pCluster->createdTime >> 32) & (pCluster->createdTime)) | (*(int32_t*)tsFirst);
+  
+  SSdbOper oper = {
+    .type = SDB_OPER_GLOBAL,
+    .table = tsClusterSdb,
+    .pObj = pCluster,
+  };
+
+  return sdbInsertRow(&oper);
+}
+
+int32_t mnodeGetClusterId() {
+  return tsClusterId;
+}
+
+void mnodeUpdateClusterId() {
+  SClusterObj *pCluster = NULL;
+  mnodeGetNextCluster(NULL, &pCluster);
+  if (pCluster != NULL) {
+    tsClusterId = pCluster->clusterId;
+    mnodeDecClusterRef(pCluster);
+    mInfo("cluster id is %d", tsClusterId);
+  } else {
+    //assert(false);
+  }
+}
+
+
+static int32_t mnodeGetClusterMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
+  int32_t cols = 0;
+  SSchema *pSchema = pMeta->schema;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_INT;
+  strcpy(pSchema[cols].name, "clusterId");
+  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++;
+
+  pMeta->numOfColumns = htons(cols);
+  strcpy(pMeta->tableId, "show cluster");
+  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 = 1;
+  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
+
+  return 0;
+}
+
+static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
+  int32_t numOfRows = 0;
+  int32_t cols = 0;
+  char *  pWrite;
+  SClusterObj *pCluster = NULL;
+
+  while (numOfRows < rows) {
+    pShow->pIter = mnodeGetNextCluster(pShow->pIter, &pCluster);
+    if (pCluster == NULL) break;
+    
+    cols = 0;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int32_t *) pWrite = pCluster->clusterId;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int32_t *) pWrite = pCluster->createdTime;
+    cols++;
+
+    mnodeDecClusterRef(pCluster);
+    numOfRows++;
+  }
+
+  pShow->numOfReads += numOfRows;
+  return numOfRows;
+}

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) {

src/mnode/src/mnodeMain.c

@@ -32,6 +32,7 @@
 #include "mnodeVgroup.h"
 #include "mnodeUser.h"
 #include "mnodeTable.h"
+#include "mnodeCluster.h"
 #include "mnodeShow.h"
 #include "mnodeProfile.h"
 
@@ -46,6 +47,7 @@ static bool tsMgmtIsRunning = false;
 
 static const SMnodeComponent tsMnodeComponents[] = {
   {"profile", mnodeInitProfile, mnodeCleanupProfile},
+  {"cluster", mnodeInitCluster, mnodeCleanupCluster},
   {"accts",   mnodeInitAccts,   mnodeCleanupAccts},
   {"users",   mnodeInitUsers,   mnodeCleanupUsers},
   {"dnodes",  mnodeInitDnodes,  mnodeCleanupDnodes},

src/mnode/src/mnodeSdb.c

@@ -29,6 +29,7 @@
 #include "mnodeInt.h"
 #include "mnodeMnode.h"
 #include "mnodeDnode.h"
+#include "mnodeCluster.h"
 #include "mnodeSdb.h"
 
 #define SDB_TABLE_LEN 12
@@ -214,6 +215,7 @@ void sdbUpdateMnodeRoles() {
     }
   }
 
+  mnodeUpdateClusterId();
   mnodeUpdateMnodeEpSet();
 }
 

src/mnode/src/mnodeShow.c

@@ -103,6 +103,8 @@ static char *mnodeGetShowType(int32_t showType) {
     case TSDB_MGMT_TABLE_SCORES:  return "show scores";
     case TSDB_MGMT_TABLE_GRANTS:  return "show grants";
     case TSDB_MGMT_TABLE_VNODES:  return "show vnodes";
+    case TSDB_MGMT_TABLE_CLUSTER: return "show clusters";
+    case TSDB_MGMT_TABLE_STREAMTABLES : return "show streamtables";
     default:                      return "undefined";
   }
 }