monitor for vnodes

include/common/tmsg.h

@@ -671,6 +671,11 @@ typedef struct {
   int64_t totalStorage;
   int64_t compStorage;
   int64_t pointsWritten;
+  int64_t numOfSelectReqs;
+  int64_t numOfInsertReqs;
+  int64_t numOfInsertSuccessReqs;
+  int64_t numOfBatchInsertReqs;
+  int64_t numOfBatchInsertSuccessReqs;
 } SVnodeLoad;
 
 typedef struct {

source/dnode/mgmt/impl/inc/dndEnv.h

@@ -105,10 +105,23 @@ typedef struct {
 } SBnodeMgmt;
 
 typedef struct {
+  int32_t openVnodes;
+  int32_t totalVnodes;
+  int32_t masterNum;
+  int64_t numOfSelectReqs;
+  double  speedOfSelectReqs;
+  int64_t numOfInsertReqs;
+  int64_t numOfInsertSuccessReqs;
+  double  speedOfInsertReqs;
+  int64_t numOfBatchInsertReqs;
+  int64_t numOfBatchInsertSuccessReqs;
+  double  speedOfBatchInsertReqs;
+  int64_t lastTime;
+} SVnodesStat;
+
+typedef struct {
+  SVnodesStat  stat;
   SHashObj    *hash;
-  int32_t      openVnodes;
-  int32_t      totalVnodes;
-  int32_t      masterNum;
   SRWLatch     latch;
   SQWorkerPool queryPool;
   SFWorkerPool fetchPool;

source/dnode/mgmt/impl/inc/dndMnode.h

@@ -34,7 +34,6 @@ int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
 
 int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
                                SMonGrantInfo *pGrantInfo);
-int8_t  dndIsMnode(SDnode *pDnode);
 
 #ifdef __cplusplus
 }

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

@@ -491,18 +491,20 @@ static void dndGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
   pInfo->disk_total = tsDataSpace.size.total;
   taosGetBandSpeed(&pInfo->net_in, &pInfo->net_out);
   taosGetIOSpeed(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
-  pInfo->req_select = 0;
-  pInfo->req_select_rate = 0;
-  pInfo->req_insert = 0;
-  pInfo->req_insert_success = 0;
-  pInfo->req_insert_rate = 0;
-  pInfo->req_insert_batch = 0;
-  pInfo->req_insert_batch_success = 0;
-  pInfo->req_insert_batch_rate = 0;
+
+  SVnodesStat *pStat = &pDnode->vmgmt.stat;
+  pInfo->req_select = pStat->numOfSelectReqs;
+  pInfo->req_select_rate = pStat->speedOfSelectReqs;
+  pInfo->req_insert = pStat->numOfInsertReqs;
+  pInfo->req_insert_success = pStat->numOfInsertSuccessReqs;
+  pInfo->req_insert_rate = pStat->speedOfInsertReqs;
+  pInfo->req_insert_batch = pStat->numOfBatchInsertReqs;
+  pInfo->req_insert_batch_success = pStat->numOfBatchInsertSuccessReqs;
+  pInfo->req_insert_batch_rate = pStat->speedOfBatchInsertReqs;
   pInfo->errors = tsNumOfErrorLogs;
-  pInfo->vnodes_num = pDnode->vmgmt.totalVnodes;
-  pInfo->masters = pDnode->vmgmt.masterNum;
-  pInfo->has_mnode = dndIsMnode(pDnode);
+  pInfo->vnodes_num = pStat->totalVnodes;
+  pInfo->masters = pStat->masterNum;
+  pInfo->has_mnode = pDnode->mmgmt.deployed;
 }
 
 static void dndSendMonitorReport(SDnode *pDnode) {

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

@@ -640,10 +640,3 @@ int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SM
   dndReleaseMnode(pDnode, pMnode);
   return code;
 }
-
-int8_t dndIsMnode(SDnode *pDnode) {
-  SMnode *pMnode = dndAcquireMnode(pDnode);
-  if (pMnode == NULL) return 0;
-  dndReleaseMnode(pDnode, pMnode);
-  return 1;
-}
\ No newline at end of file

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

@@ -382,7 +382,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
 
     char stepDesc[TSDB_STEP_DESC_LEN] = {0};
     snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to restore, %d of %d have been opened", pCfg->vgId,
-             pMgmt->openVnodes, pMgmt->totalVnodes);
+             pMgmt->stat.openVnodes, pMgmt->stat.totalVnodes);
     dndReportStartup(pDnode, "open-vnodes", stepDesc);
 
     SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId, .dbId = pCfg->dbUid};
@@ -396,7 +396,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
       pThread->opened++;
     }
 
-    atomic_add_fetch_32(&pMgmt->openVnodes, 1);
+    atomic_add_fetch_32(&pMgmt->stat.openVnodes, 1);
   }
 
   dDebug("thread:%d, total vnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened,
@@ -422,7 +422,7 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
     return -1;
   }
 
-  pMgmt->totalVnodes = numOfVnodes;
+  pMgmt->stat.totalVnodes = numOfVnodes;
 
   int32_t threadNum = tsNumOfCores;
 #if 1
@@ -470,11 +470,11 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
   free(threads);
   free(pCfgs);
 
-  if (pMgmt->openVnodes != pMgmt->totalVnodes) {
-    dError("there are total vnodes:%d, opened:%d", pMgmt->totalVnodes, pMgmt->openVnodes);
+  if (pMgmt->stat.openVnodes != pMgmt->stat.totalVnodes) {
+    dError("there are total vnodes:%d, opened:%d", pMgmt->stat.totalVnodes, pMgmt->stat.openVnodes);
     return -1;
   } else {
-    dInfo("total vnodes:%d open successfully", pMgmt->totalVnodes);
+    dInfo("total vnodes:%d open successfully", pMgmt->stat.totalVnodes);
     return 0;
   }
 }
@@ -982,10 +982,14 @@ void dndGetVnodeLoads(SDnode *pDnode, SArray *pLoads) {
   SVnodesMgmt *pMgmt = &pDnode->vmgmt;
   int32_t      totalVnodes = 0;
   int32_t      masterNum = 0;
+  int64_t      numOfSelectReqs = 0;
+  int64_t      numOfInsertReqs = 0;
+  int64_t      numOfInsertSuccessReqs = 0;
+  int64_t      numOfBatchInsertReqs = 0;
+  int64_t      numOfBatchInsertSuccessReqs = 0;
 
   taosRLockLatch(&pMgmt->latch);
 
-  int32_t v = 0;
   void   *pIter = taosHashIterate(pMgmt->hash, NULL);
   while (pIter) {
     SVnodeObj **ppVnode = pIter;
@@ -996,12 +1000,41 @@ void dndGetVnodeLoads(SDnode *pDnode, SArray *pLoads) {
     vnodeGetLoad(pVnode->pImpl, &vload);
     taosArrayPush(pLoads, &vload);
 
+    numOfSelectReqs += vload.numOfSelectReqs;
+    numOfInsertReqs += vload.numOfInsertReqs;
+    numOfInsertSuccessReqs += vload.numOfInsertSuccessReqs;
+    numOfBatchInsertReqs += vload.numOfBatchInsertReqs;
+    numOfBatchInsertSuccessReqs += vload.numOfBatchInsertSuccessReqs;
     totalVnodes++;
     if (vload.role == TAOS_SYNC_STATE_LEADER) masterNum++;
+
     pIter = taosHashIterate(pMgmt->hash, pIter);
   }
 
   taosRUnLockLatch(&pMgmt->latch);
-  pMgmt->totalVnodes = totalVnodes;
-  pMgmt->masterNum = masterNum;
+
+  SVnodesStat *pStat = &pMgmt->stat;
+  pStat->totalVnodes = totalVnodes;
+  pStat->masterNum = masterNum;
+
+  int64_t curTime = taosGetTimestampMs();
+  if (pStat->lastTime == 0 || pStat->lastTime >= curTime) {
+    pStat->lastTime = curTime;
+    pStat->numOfSelectReqs = numOfSelectReqs;
+    pStat->numOfInsertReqs = numOfInsertReqs;
+    pStat->numOfInsertSuccessReqs = numOfInsertSuccessReqs;
+    pStat->numOfBatchInsertReqs = numOfBatchInsertReqs;
+    pStat->numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;
+    return;
+  }
+
+  double interval = (curTime - pStat->lastTime) * 1000.0;
+  pStat->speedOfSelectReqs = (numOfSelectReqs - pStat->numOfSelectReqs) / interval;
+  pStat->speedOfInsertReqs = (numOfInsertReqs - pStat->numOfInsertReqs) / interval;
+  pStat->speedOfBatchInsertReqs = (numOfBatchInsertReqs - pStat->numOfBatchInsertReqs) / interval;
+  pStat->numOfSelectReqs = numOfSelectReqs;
+  pStat->numOfInsertReqs = numOfInsertReqs;
+  pStat->numOfInsertSuccessReqs = numOfInsertSuccessReqs;
+  pStat->numOfBatchInsertReqs = numOfBatchInsertReqs;
+  pStat->numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;
 }

source/dnode/vnode/src/vnd/vnodeInt.c

@@ -32,6 +32,11 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
   pLoad->totalStorage = 300;
   pLoad->compStorage = 200;
   pLoad->pointsWritten = 100;
+  pLoad->numOfSelectReqs = 1;
+  pLoad->numOfInsertReqs = 3;
+  pLoad->numOfInsertSuccessReqs = 2;
+  pLoad->numOfBatchInsertReqs = 5;
+  pLoad->numOfBatchInsertSuccessReqs = 4;
   return 0;
 }