Merge pull request #8628 from taosdata/feature/TD-6452

[TD-6452]<feature>: taoskeeper metrics collector phase 1 taosd implementation

src/common/inc/tglobal.h

@@ -148,6 +148,7 @@ extern char   tsMqttTopic[];
 
 // monitor
 extern int8_t  tsEnableMonitorModule;
+extern int8_t  tsMonitorReplica;
 extern char    tsMonitorDbName[];
 extern char    tsInternalPass[];
 extern int32_t tsMonitorInterval;

src/common/src/tglobal.c

@@ -193,6 +193,7 @@ char   tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test";  // #
 
 // monitor
 int8_t  tsEnableMonitorModule = 1;
+int8_t  tsMonitorReplica = 1;
 char    tsMonitorDbName[TSDB_DB_NAME_LEN] = "log";
 char    tsInternalPass[] = "secretkey";
 int32_t tsMonitorInterval = 30;  // seconds
@@ -669,6 +670,16 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_SECOND;
   taosInitConfigOption(cfg);
 
+  cfg.option = "monitorReplica";
+  cfg.ptr = &tsMonitorReplica;
+  cfg.valType = TAOS_CFG_VTYPE_INT8;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = 1;
+  cfg.maxValue = 3;
+  cfg.ptrLength = 1;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "offlineThreshold";
   cfg.ptr = &tsOfflineThreshold;
   cfg.valType = TAOS_CFG_VTYPE_INT32;

src/dnode/inc/dnodeInt.h

@@ -29,8 +29,8 @@ extern "C" {
 
 extern int32_t dDebugFlag;
 
-#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", 255, __VA_ARGS__); }}
-#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", 255, __VA_ARGS__); }}
+#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", 255, __VA_ARGS__); dnodeIncDnodeError(); }}
+#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", 255, __VA_ARGS__); dnodeIncDnodeError(); }}
 #define dWarn(...)  { if (dDebugFlag & DEBUG_WARN)  { taosPrintLog("DND WARN ", 255, __VA_ARGS__); }}
 #define dInfo(...)  { if (dDebugFlag & DEBUG_INFO)  { taosPrintLog("DND ", 255, __VA_ARGS__); }}
 #define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", dDebugFlag, __VA_ARGS__); }}

src/dnode/src/dnodeMain.c

@@ -54,6 +54,7 @@ void    moduleStop() {}
 
 void *tsDnodeTmr = NULL;
 static SRunStatus tsRunStatus = TSDB_RUN_STATUS_STOPPED;
+static int64_t tsDnodeErrors = 0;
 
 static int32_t dnodeInitStorage();
 static void    dnodeCleanupStorage();
@@ -225,6 +226,14 @@ static void dnodeSetRunStatus(SRunStatus status) {
   tsRunStatus = status;
 }
 
+int64_t dnodeGetDnodeError() {
+  return tsDnodeErrors;
+}
+
+void dnodeIncDnodeError() {
+  atomic_add_fetch_64(&tsDnodeErrors, 1);
+}
+
 static void dnodeCheckDataDirOpenned(char *dir) {
   char filepath[256] = {0};
   sprintf(filepath, "%s/.running", dir);

src/dnode/src/dnodeShell.c

@@ -28,8 +28,8 @@ static void  (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void    dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *);
 static int     dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 static void  * tsShellRpc = NULL;
-static int32_t tsQueryReqNum  = 0;
-static int32_t tsSubmitReqNum = 0;
+static int64_t tsQueryReqNum  = 0;
+static int64_t tsSubmitReqNum = 0;
 
 int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT]         = dnodeDispatchToVWriteQueue;
@@ -136,9 +136,9 @@ static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
   }
 
   if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) {
-    atomic_fetch_add_32(&tsQueryReqNum, 1);
+    atomic_fetch_add_64(&tsQueryReqNum, 1);
   } else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
-    atomic_fetch_add_32(&tsSubmitReqNum, 1);
+    atomic_fetch_add_64(&tsSubmitReqNum, 1);
   } else {}
 
   if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
@@ -237,15 +237,31 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) {
   }
 }
 
-SStatisInfo dnodeGetStatisInfo() {
-  SStatisInfo info = {0};
+SDnodeStatisInfo dnodeGetStatisInfo() {
+  SDnodeStatisInfo info = {0};
   if (dnodeGetRunStatus() == TSDB_RUN_STATUS_RUNING) {
 #ifdef HTTP_EMBEDDED
     info.httpReqNum   = httpGetReqCount();
 #endif
-    info.queryReqNum  = atomic_exchange_32(&tsQueryReqNum, 0);
-    info.submitReqNum = atomic_exchange_32(&tsSubmitReqNum, 0);
+    info.queryReqNum  = atomic_exchange_64(&tsQueryReqNum, 0);
+    info.submitReqNum = atomic_exchange_64(&tsSubmitReqNum, 0);
   }
 
   return info;
 }
+
+int32_t dnodeGetHttpStatusInfo(int32_t index) {
+  int32_t httpStatus = 0;
+#ifdef HTTP_EMBEDDED
+    httpStatus = httpGetStatusCodeCount(index);
+#endif
+  return httpStatus;
+}
+
+void dnodeClearHttpStatusInfo() {
+#ifdef HTTP_EMBEDDED
+  for (int i = 0; i < MAX_HTTP_STATUS_CODE_NUM; ++i) {
+    httpClearStatusCodeCount(i);
+  }
+#endif
+}

src/inc/dnode.h

@@ -23,13 +23,16 @@ extern "C" {
 #include "trpc.h"
 #include "taosmsg.h"
 
+#define MAX_HTTP_STATUS_CODE_NUM 63
 typedef struct {
-  int32_t queryReqNum;
-  int32_t submitReqNum;
-  int32_t httpReqNum;
-} SStatisInfo;
+  int64_t queryReqNum;
+  int64_t submitReqNum;
+  int64_t httpReqNum;
+} SDnodeStatisInfo;
 
-SStatisInfo dnodeGetStatisInfo();
+SDnodeStatisInfo dnodeGetStatisInfo();
+int32_t          dnodeGetHttpStatusInfo(int32_t index);
+void             dnodeClearHttpStatusInfo();
 
 bool    dnodeIsFirstDeploy();
 bool    dnodeIsMasterEp(char *ep);
@@ -37,6 +40,8 @@ void    dnodeGetEpSetForPeer(SRpcEpSet *epSet);
 void    dnodeGetEpSetForShell(SRpcEpSet *epSet);
 int32_t dnodeGetDnodeId();
 void    dnodeGetClusterId(char *clusterId);
+int64_t dnodeGetDnodeError();
+void    dnodeIncDnodeError();
 
 void    dnodeUpdateEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
 bool    dnodeCheckEpChanged(int32_t dnodeId, char *epstr);

src/inc/http.h

@@ -22,7 +22,9 @@ extern "C" {
 
 #include <stdint.h>
 
-int32_t httpGetReqCount();
+int64_t httpGetReqCount();
+int32_t httpGetStatusCodeCount(int index);
+int32_t httpClearStatusCodeCount(int index);
 int32_t httpInitSystem();
 int32_t httpStartSystem();
 void httpStopSystem();

src/inc/monitor.h

@@ -22,6 +22,17 @@ extern "C" {
 
 #include <stdint.h>
 
+#define monSaveLogs(level, ...) {      \
+  monSaveLog(level, __VA_ARGS__);      \
+  monSaveDnodeLog(level, __VA_ARGS__); \
+}
+
+typedef struct {
+  const char * name;
+  int32_t code;
+  int32_t index;
+} SMonHttpStatus;
+
 typedef struct {
   char *  acctId;
   int64_t currentPointsPerSecond;
@@ -53,9 +64,16 @@ void    monStopSystem();
 void    monCleanupSystem();
 void    monSaveAcctLog(SAcctMonitorObj *pMonObj);
 void    monSaveLog(int32_t level, const char *const format, ...);
+void    monSaveDnodeLog(int32_t level, const char *const format, ...);
 void    monExecuteSQL(char *sql);
 typedef void (*MonExecuteSQLCbFP)(void *param, TAOS_RES *, int code);
 void monExecuteSQLWithResultCallback(char *sql, MonExecuteSQLCbFP callback, void* param);
+void    monIncQueryReqCnt();
+void    monIncSubmitReqCnt();
+int32_t monFetchQueryReqCnt();
+int32_t monFetchSubmitReqCnt();
+SMonHttpStatus *monGetHttpStatusHashTableEntry(int32_t code);
+
 #ifdef __cplusplus
 }
 #endif

src/inc/vnode.h

@@ -22,6 +22,12 @@ extern "C" {
 #include "trpc.h"
 #include "twal.h"
 
+typedef struct {
+  int64_t submitReqSucNum;
+  int64_t submitRowNum;
+  int64_t submitRowSucNum;
+} SVnodeStatisInfo;
+
 typedef struct {
   int32_t len;
   void *  rsp;
@@ -80,6 +86,8 @@ int32_t vnodeWriteToWQueue(void *pVnode, void *pHead, int32_t qtype, void *pRpcM
 void    vnodeFreeFromWQueue(void *pVnode, SVWriteMsg *pWrite);
 int32_t vnodeProcessWrite(void *pVnode, void *pHead, int32_t qtype, void *pRspRet);
 
+SVnodeStatisInfo vnodeGetStatisInfo();
+
 // vnodeSync
 void    vnodeConfirmForward(void *pVnode, uint64_t version, int32_t code, bool force);
 

src/mnode/inc/mnodeInt.h

@@ -41,9 +41,9 @@ extern int32_t sdbDebugFlag;
 #define sdbDebug(...) { if (sdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
 #define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
 
-#define mLError(...) { monSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) }
-#define mLWarn(...)  { monSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)  }
-#define mLInfo(...)  { monSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
+#define mLError(...) { monSaveLogs(2, __VA_ARGS__); mError(__VA_ARGS__) }
+#define mLWarn(...)  { monSaveLogs(1, __VA_ARGS__); mWarn(__VA_ARGS__)  }
+#define mLInfo(...)  { monSaveLogs(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
 
 #ifdef __cplusplus
 }

src/mnode/src/mnodeDnode.c

@@ -1178,7 +1178,7 @@ static int32_t mnodeGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
   SUserObj *pUser = mnodeGetUserFromConn(pConn);
   if (pUser == NULL) return 0;
 
-  if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0)  {
+  if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0 ) {
     mnodeDecUserRef(pUser);
     return TSDB_CODE_MND_NO_RIGHTS;
   }
@@ -1256,10 +1256,10 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
           STR_TO_VARSTR(pWrite, syncRole[pVgid->role]);
           cols++;
           numOfRows++;
-          
         }
       }
       if (numOfRows >= rows) {
+        mnodeDecVgroupRef(pVgroup);
         break;
       }
 

src/mnode/src/mnodeProfile.c

@@ -166,7 +166,7 @@ static void mnodeCancelGetNextConn(void *pIter) {
 static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
   SUserObj *pUser = mnodeGetUserFromConn(pConn);
   if (pUser == NULL) return 0;
-  if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
+  if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
   
   int32_t cols = 0;
   SSchema *pSchema = pMeta->schema;
@@ -322,7 +322,7 @@ int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pHBMsg) {
 static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
   SUserObj *pUser = mnodeGetUserFromConn(pConn);
   if (pUser == NULL) return 0;
-  if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
+  if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
 
   int32_t cols = 0;
   SSchema *pSchema = pMeta->schema;

src/os/inc/osSysinfo.h

@@ -30,10 +30,11 @@ int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize);
 
 int32_t taosGetCpuCores();
 void taosGetSystemInfo();
-bool taosReadProcIO(int64_t* rchars, int64_t* wchars);
-bool taosGetProcIO(float *readKB, float *writeKB);
+bool taosReadProcIO(int64_t* rchars, int64_t* wchars, int64_t* rbytes, int64_t* wbytes);
+bool taosGetProcIO(float *rcharKB, float *wcharKB, float *rbyteKB, float* wbyteKB);
 bool taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes);
 bool taosGetBandSpeed(float *bandSpeedKb);
+bool taosGetNetworkIO(float *netInKb, float *netOutKb);
 void taosGetDisk();
 bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage) ;
 bool taosGetProcMemory(float *memoryUsedMB) ;

src/os/src/darwin/dwSysInfo.c

@@ -191,15 +191,19 @@ void taosGetSystemInfo() {
   taosGetSystemLocale();
 }
 
-bool taosReadProcIO(int64_t *rchars, int64_t *wchars) {
+bool taosReadProcIO(int64_t *rchars, int64_t *wchars, int64_t *rbytes, int64_t *wbytes) {
   if (rchars) *rchars = 0;
   if (wchars) *wchars = 0;
+  if (rbytes) *rbytes = 0;
+  if (wbytes) *wbytes = 0;
   return true;
 }
 
-bool taosGetProcIO(float *readKB, float *writeKB) {
-  *readKB = 0;
-  *writeKB = 0;
+bool taosGetProcIO(float *rcharKB, float *wcharKB, float *rbyteKB, float *wbyteKB) {
+  *rcharKB = 0;
+  *wcharKB = 0;
+  *rbyteKB = 0;
+  *wbyteKB = 0;
   return true;
 }
 
@@ -215,6 +219,12 @@ bool taosGetBandSpeed(float *bandSpeedKb) {
   return true;
 }
 
+bool taosGetNetworkIO(float *netInKb, float *netOutKb) {
+  *netInKb = 0;
+  *netOutKb = 0;
+  return true;
+}
+
 bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage) {
   *sysCpuUsage = 0;
   *procCpuUsage = 0;

src/os/src/detail/osSysinfo.c

@@ -335,7 +335,9 @@ int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize) {
 }
 
 bool taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes) {
-  *bytes = 0;
+  if (bytes) *bytes = 0;
+  if (rbytes) *rbytes = 0;
+  if (tbytes) *tbytes = 0;
   FILE *fp = fopen(tsSysNetFile, "r");
   if (fp == NULL) {
     uError("open file:%s failed", tsSysNetFile);
@@ -350,7 +352,7 @@ bool taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes) {
     memset(line, 0, len);
 
     int64_t o_rbytes = 0;
-    int64_t rpackts = 0;
+    int64_t rpackets = 0;
     int64_t o_tbytes = 0;
     int64_t tpackets = 0;
     int64_t nouse1 = 0;
@@ -376,10 +378,10 @@ bool taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes) {
     sscanf(line,
            "%s %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64
            " %" PRId64,
-           nouse0, &o_rbytes, &rpackts, &nouse1, &nouse2, &nouse3, &nouse4, &nouse5, &nouse6, &o_tbytes, &tpackets);
+           nouse0, &o_rbytes, &rpackets, &nouse1, &nouse2, &nouse3, &nouse4, &nouse5, &nouse6, &o_tbytes, &tpackets);
     if (rbytes) *rbytes = o_rbytes;
     if (tbytes) *tbytes = o_tbytes;
-    *bytes += (o_rbytes + o_tbytes);
+    if (bytes)  *bytes += (o_rbytes + o_tbytes);
   }
 
   tfree(line);
@@ -424,7 +426,46 @@ bool taosGetBandSpeed(float *bandSpeedKb) {
   return true;
 }
 
-bool taosReadProcIO(int64_t *rchars, int64_t *wchars) {
+bool taosGetNetworkIO(float *netInKb, float *netOutKb) {
+  static int64_t lastBytesIn = 0, lastBytesOut = 0;
+  static time_t  lastTimeIO = 0;
+  int64_t        curBytesIn = 0, curBytesOut = 0;
+  time_t         curTime = time(NULL);
+
+  if (!taosGetCardInfo(NULL, &curBytesIn, &curBytesOut)) {
+    return false;
+  }
+
+  if (lastTimeIO == 0 || lastBytesIn == 0 || lastBytesOut == 0) {
+    lastTimeIO = curTime;
+    lastBytesIn = curBytesIn; lastBytesOut = curBytesOut;
+    *netInKb = 0;
+    *netOutKb = 0;
+    return true;
+  }
+
+  if (lastTimeIO >= curTime || lastBytesIn > curBytesIn || lastBytesOut > curBytesOut) {
+    lastTimeIO = curTime;
+    lastBytesIn = curBytesIn; lastBytesOut = curBytesOut;
+    *netInKb = 0;
+    *netOutKb = 0;
+    return true;
+  }
+
+  double totalBytesIn = (double)(curBytesIn - lastBytesIn) / 1024 * 8;  // Kb
+  *netInKb = (float)(totalBytesIn / (double)(curTime - lastTimeIO));
+
+  double totalBytesOut = (double)(curBytesOut - lastBytesOut) / 1024 * 8;  // Kb
+  *netOutKb = (float)(totalBytesOut / (double)(curTime - lastTimeIO));
+
+  lastTimeIO = curTime;
+  lastBytesIn = curBytesIn;
+  lastBytesOut = curBytesOut;
+
+  return true;
+}
+
+bool taosReadProcIO(int64_t *rchars, int64_t *wchars, int64_t *rbytes, int64_t *wbytes) {
   FILE *fp = fopen(tsProcIOFile, "r");
   if (fp == NULL) {
     uError("open file:%s failed", tsProcIOFile);
@@ -434,7 +475,7 @@ bool taosReadProcIO(int64_t *rchars, int64_t *wchars) {
   ssize_t _bytes = 0;
   size_t len;
   char * line = NULL;
-  char   tmp[10];
+  char   tmp[15];
   int    readIndex = 0;
 
   while (!feof(fp)) {
@@ -450,16 +491,21 @@ bool taosReadProcIO(int64_t *rchars, int64_t *wchars) {
     } else if (strstr(line, "wchar:") != NULL) {
       sscanf(line, "%s %" PRId64, tmp, wchars);
       readIndex++;
-    } else {
+    } else if (strstr(line, "read_bytes:") != NULL){
+      sscanf(line, "%s %" PRId64, tmp, rbytes);
+      readIndex++;
+    } else if (strstr(line, "write_bytes:") != NULL){
+      sscanf(line, "%s %" PRId64, tmp, wbytes);
+      readIndex++;
     }
 
-    if (readIndex >= 2) break;
+    if (readIndex >= 4) break;
   }
 
   tfree(line);
   fclose(fp);
 
-  if (readIndex < 2) {
+  if (readIndex < 4) {
     uError("read file:%s failed", tsProcIOFile);
     return false;
   }
@@ -467,30 +513,43 @@ bool taosReadProcIO(int64_t *rchars, int64_t *wchars) {
   return true;
 }
 
-bool taosGetProcIO(float *readKB, float *writeKB) {
-  static int64_t lastReadbyte = -1;
-  static int64_t lastWritebyte = -1;
+bool taosGetProcIO(float *rcharKB, float *wcharKB, float *rbyteKB, float *wbyteKB) {
+  static int64_t lastRchar = -1, lastRbyte = -1;
+  static int64_t lastWchar = -1, lastWbyte = -1;
+  static time_t  lastTime = 0;
+  time_t         curTime = time(NULL);
 
-  int64_t curReadbyte = 0;
-  int64_t curWritebyte = 0;
+  int64_t curRchar = 0, curRbyte = 0;
+  int64_t curWchar = 0, curWbyte = 0;
 
-  if (!taosReadProcIO(&curReadbyte, &curWritebyte)) {
+  if (!taosReadProcIO(&curRchar, &curWchar, &curRbyte, &curWbyte)) {
     return false;
   }
 
-  if (lastReadbyte == -1 || lastWritebyte == -1) {
-    lastReadbyte = curReadbyte;
-    lastWritebyte = curWritebyte;
+  if (lastTime == 0 || lastRchar == -1 || lastWchar == -1 || lastRbyte == -1 || lastWbyte == -1) {
+    lastTime  = curTime;
+    lastRchar = curRchar;
+    lastWchar = curWchar;
+    lastRbyte = curRbyte;
+    lastWbyte = curWbyte;
     return false;
   }
 
-  *readKB = (float)((double)(curReadbyte - lastReadbyte) / 1024);
-  *writeKB = (float)((double)(curWritebyte - lastWritebyte) / 1024);
-  if (*readKB < 0) *readKB = 0;
-  if (*writeKB < 0) *writeKB = 0;
+  *rcharKB = (float)((double)(curRchar - lastRchar) / 1024 / (double)(curTime - lastTime));
+  *wcharKB = (float)((double)(curWchar - lastWchar) / 1024 / (double)(curTime - lastTime));
+  if (*rcharKB < 0) *rcharKB = 0;
+  if (*wcharKB < 0) *wcharKB = 0;
+
+  *rbyteKB = (float)((double)(curRbyte - lastRbyte) / 1024 / (double)(curTime - lastTime));
+  *wbyteKB = (float)((double)(curWbyte - lastWbyte) / 1024 / (double)(curTime - lastTime));
+  if (*rbyteKB < 0) *rbyteKB = 0;
+  if (*wbyteKB < 0) *wbyteKB = 0;
 
-  lastReadbyte = curReadbyte;
-  lastWritebyte = curWritebyte;
+  lastRchar = curRchar;
+  lastWchar = curWchar;
+  lastRbyte = curRbyte;
+  lastWbyte = curWbyte;
+  lastTime  = curTime;
 
   return true;
 }
@@ -501,13 +560,13 @@ void taosGetSystemInfo() {
   tsNumOfCores = taosGetCpuCores();
   tsTotalMemoryMB = taosGetTotalMemory();
 
-  float tmp1, tmp2;
+  float tmp1, tmp2, tmp3, tmp4;
   taosGetSysMemory(&tmp1);
   taosGetProcMemory(&tmp2);
   // taosGetDisk();
   taosGetBandSpeed(&tmp1);
   taosGetCpuUsage(&tmp1, &tmp2);
-  taosGetProcIO(&tmp1, &tmp2);
+  taosGetProcIO(&tmp1, &tmp2, &tmp3, &tmp4);
 
   taosGetSystemTimezone();
   taosGetSystemLocale();

src/os/src/windows/wSysinfo.c

@@ -169,40 +169,59 @@ bool taosGetBandSpeed(float *bandSpeedKb) {
   return true;
 }
 
-bool taosReadProcIO(int64_t *readbyte, int64_t *writebyte) {
+bool taosGetNetworkIO(float *netInKb, float *netOutKb) {
+  *netInKb = 0;
+  *netOutKb = 0;
+  return true;
+}
+
+bool taosReadProcIO(int64_t *rchars, int64_t *wchars, int64_t *rbytes, int64_t *wbytes) {
   IO_COUNTERS io_counter;
   if (GetProcessIoCounters(GetCurrentProcess(), &io_counter)) {
-    if (readbyte) *readbyte = io_counter.ReadTransferCount;
-    if (writebyte) *writebyte = io_counter.WriteTransferCount;
+    if (rchars) *rchars = io_counter.ReadTransferCount;
+    if (wchars) *wchars = io_counter.WriteTransferCount;
     return true;
   }
   return false;
 }
 
-bool taosGetProcIO(float *readKB, float *writeKB) {
-  static int64_t lastReadbyte = -1;
-  static int64_t lastWritebyte = -1;
+bool taosGetProcIO(float *rcharKB, float *wcharKB, float *rbyteKB, float *wbyteKB) {
+  static int64_t lastRchar = -1, lastRbyte = -1;
+  static int64_t lastWchar = -1, lastWbyte = -1;
+  static time_t  lastTime = 0;
+  time_t         curTime = time(NULL);
 
-  int64_t curReadbyte = 0;
-  int64_t curWritebyte = 0;
+  int64_t curRchar = 0, curRbyte = 0;
+  int64_t curWchar = 0, curWbyte = 0;
 
-  if (!taosReadProcIO(&curReadbyte, &curWritebyte)) {
+  if (!taosReadProcIO(&curRchar, &curWchar, &curRbyte, &curWbyte)) {
     return false;
   }
 
-  if (lastReadbyte == -1 || lastWritebyte == -1) {
-    lastReadbyte = curReadbyte;
-    lastWritebyte = curWritebyte;
+  if (lastTime == 0 || lastRchar == -1 || lastWchar == -1 || lastRbyte == -1 || lastWbyte == -1) {
+    lastTime  = curTime;
+    lastRchar = curRchar;
+    lastWchar = curWchar;
+    lastRbyte = curRbyte;
+    lastWbyte = curWbyte;
     return false;
   }
 
-  *readKB = (float)((double)(curReadbyte - lastReadbyte) / 1024);
-  *writeKB = (float)((double)(curWritebyte - lastWritebyte) / 1024);
-  if (*readKB < 0) *readKB = 0;
-  if (*writeKB < 0) *writeKB = 0;
+  *rcharKB = (float)((double)(curRchar - lastRchar) / 1024 / (double)(curTime - lastTime));
+  *wcharKB = (float)((double)(curWchar - lastWchar) / 1024 / (double)(curTime - lastTime));
+  if (*rcharKB < 0) *rcharKB = 0;
+  if (*wcharKB < 0) *wcharKB = 0;
+
+  *rbyteKB = (float)((double)(curRbyte - lastRbyte) / 1024 / (double)(curTime - lastTime));
+  *wbyteKB = (float)((double)(curWbyte - lastWbyte) / 1024 / (double)(curTime - lastTime));
+  if (*rbyteKB < 0) *rbyteKB = 0;
+  if (*wbyteKB < 0) *wbyteKB = 0;
 
-  lastReadbyte = curReadbyte;
-  lastWritebyte = curWritebyte;
+  lastRchar = curRchar;
+  lastWchar = curWchar;
+  lastRbyte = curRbyte;
+  lastWbyte = curWbyte;
+  lastTime  = curTime;
 
   return true;
 }
@@ -211,11 +230,11 @@ void taosGetSystemInfo() {
   tsNumOfCores = taosGetCpuCores();
   tsTotalMemoryMB = taosGetTotalMemory();
 
-  float tmp1, tmp2;
+  float tmp1, tmp2, tmp3, tmp4;
   // taosGetDisk();
   taosGetBandSpeed(&tmp1);
   taosGetCpuUsage(&tmp1, &tmp2);
-  taosGetProcIO(&tmp1, &tmp2);
+  taosGetProcIO(&tmp1, &tmp2, &tmp3, &tmp4);
 
   taosGetSystemTimezone();
   taosGetSystemLocale();

src/plugins/http/inc/httpInt.h

@@ -42,6 +42,7 @@
 #define HTTP_WRITE_WAIT_TIME_MS     5
 #define HTTP_PASSWORD_LEN           TSDB_UNI_LEN
 #define HTTP_SESSION_ID_LEN         (TSDB_USER_LEN + HTTP_PASSWORD_LEN)
+#define HTTP_STATUS_CODE_NUM        63
 
 typedef enum HttpReqType {
   HTTP_REQTYPE_OTHERS = 0,
@@ -187,8 +188,9 @@ typedef struct HttpServer {
   SOCKET            fd;
   int32_t           numOfThreads;
   int32_t           methodScannerLen;
-  int32_t           requestNum;
+  int64_t           requestNum;
   int32_t           status;
+  int32_t           statusCodeErrs[HTTP_STATUS_CODE_NUM];
   pthread_t         thread;
   HttpThread *      pThreads;
   void *            contextCache;

src/plugins/http/src/httpMetricsHandle.c

@@ -123,9 +123,9 @@ bool metricsProcessRequest(HttpContext* pContext) {
   }
 
   {
-    int64_t rchars = 0;
-    int64_t wchars = 0;
-    bool succeeded = taosReadProcIO(&rchars, &wchars);
+    int64_t rchars = 0, rbytes = 0;
+    int64_t wchars = 0, wbytes = 0;
+    bool succeeded = taosReadProcIO(&rchars, &wchars, &rbytes, &wbytes);
     if (!succeeded) {
       httpError("failed to get io info");
     } else {
@@ -164,7 +164,7 @@ bool metricsProcessRequest(HttpContext* pContext) {
   }
 
   {
-    SStatisInfo info = dnodeGetStatisInfo();
+    SDnodeStatisInfo info = dnodeGetStatisInfo();
     {
       char* keyReqHttp = "req_http";
       char* keyReqSelect = "req_select";

src/plugins/http/src/httpResp.c

@@ -21,6 +21,7 @@
 #include "httpResp.h"
 #include "httpJson.h"
 #include "httpContext.h"
+#include "monitor.h"
 
 const char *httpKeepAliveStr[] = {"", "Connection: Keep-Alive\r\n", "Connection: Close\r\n"};
 
@@ -153,6 +154,10 @@ void httpSendErrorResp(HttpContext *pContext, int32_t errNo) {
     httpCode = pContext->parser->httpCode;
   }
 
+  HttpServer *pServer = &tsHttpServer;
+  SMonHttpStatus *httpStatus = monGetHttpStatusHashTableEntry(httpCode);
+  pServer->statusCodeErrs[httpStatus->index] += 1;
+
   pContext->error = true;
 
   char *httpCodeStr = httpGetStatusDesc(httpCode);

src/plugins/http/src/httpServer.c

@@ -190,7 +190,7 @@ static void httpProcessHttpData(void *param) {
       } else {
         if (httpReadData(pContext)) {
           (*(pThread->processData))(pContext);
-          atomic_fetch_add_32(&pServer->requestNum, 1);
+          atomic_fetch_add_64(&pServer->requestNum, 1);
         }
       }
     }

src/plugins/http/src/httpSystem.c

@@ -120,4 +120,10 @@ void httpCleanUpSystem() {
   tsHttpServer.status = HTTP_SERVER_CLOSED;
 }
 
-int32_t httpGetReqCount() { return atomic_exchange_32(&tsHttpServer.requestNum, 0); }
+int64_t httpGetReqCount() { return atomic_exchange_64(&tsHttpServer.requestNum, 0); }
+int32_t httpGetStatusCodeCount(int index) {
+  return atomic_load_32(&tsHttpServer.statusCodeErrs[index]);
+}
+int32_t httpClearStatusCodeCount(int index) {
+  return atomic_exchange_32(&tsHttpServer.statusCodeErrs[index], 0);
+}

src/tsdb/src/tsdbMemTable.c

@@ -57,7 +57,7 @@ int32_t tsdbInsertData(STsdbRepo *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pR
   STsdbRepo *    pRepo = repo;
   SSubmitMsgIter msgIter = {0};
   SSubmitBlk *   pBlock = NULL;
-  int32_t        affectedrows = 0;
+  int32_t        affectedrows = 0, numOfRows = 0;
 
   if (tsdbScanAndConvertSubmitMsg(pRepo, pMsg) < 0) {
     if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
@@ -73,9 +73,13 @@ int32_t tsdbInsertData(STsdbRepo *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pR
     if (tsdbInsertDataToTable(pRepo, pBlock, &affectedrows) < 0) {
       return -1;
     }
+    numOfRows += pBlock->numOfRows;
   }
 
-  if (pRsp != NULL) pRsp->affectedRows = htonl(affectedrows);
+  if (pRsp != NULL) {
+    pRsp->affectedRows = htonl(affectedrows);
+    pRsp->numOfRows = htonl(numOfRows);
+  }
 
   if (tsdbCheckCommit(pRepo) < 0) return -1;
   return 0;

src/vnode/src/vnodeWrite.c

@@ -27,6 +27,10 @@
 #define MAX_QUEUED_MSG_NUM 100000
 #define MAX_QUEUED_MSG_SIZE 1024*1024*1024  //1GB
 
+static int64_t tsSubmitReqSucNum = 0;
+static int64_t tsSubmitRowNum = 0;
+static int64_t tsSubmitRowSucNum = 0;
+
 extern void *  tsDnodeTmr;
 static int32_t (*vnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *, void *pCont, SRspRet *);
 static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pCont, SRspRet *);
@@ -163,7 +167,16 @@ static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pR
     pRsp = pRet->rsp;
   }
 
-  if (tsdbInsertData(pVnode->tsdb, pCont, pRsp) < 0) code = terrno;
+  if (tsdbInsertData(pVnode->tsdb, pCont, pRsp) < 0) {
+    code = terrno;
+  } else {
+    if (pRsp != NULL) atomic_fetch_add_64(&tsSubmitReqSucNum, 1);
+  }
+
+  if (pRsp) {
+    atomic_fetch_add_64(&tsSubmitRowNum, ntohl(pRsp->numOfRows));
+    atomic_fetch_add_64(&tsSubmitRowSucNum, ntohl(pRsp->affectedRows));
+  }
 
   return code;
 }
@@ -425,3 +438,12 @@ void vnodeWaitWriteCompleted(SVnodeObj *pVnode) {
   if (extraSleep)
     taosMsleep(900);
 }
+
+SVnodeStatisInfo vnodeGetStatisInfo() {
+  SVnodeStatisInfo info = {0};
+  info.submitReqSucNum = atomic_exchange_64(&tsSubmitReqSucNum, 0);
+  info.submitRowNum = atomic_exchange_64(&tsSubmitRowNum, 0);
+  info.submitRowSucNum = atomic_exchange_64(&tsSubmitRowSucNum, 0);
+
+  return info;
+}

tests/script/unique/cluster/cache.sim

@@ -41,7 +41,7 @@ sql create dnode $hostname2
 sleep 10000
 
 sql show log.tables;
-if $rows > 6 then
+if $rows > 20 then
   return -1
 endi
 
@@ -50,7 +50,7 @@ print ===>rows $rows
 print $data00 $data01 $data02
 print $data10 $data11 $data12
 print $data20 $data21 $data22
-if $rows < 10 then
+if $rows < 9 then
   return -1
 endi
 

tests/script/unique/dnode/monitor.sim

@@ -42,7 +42,7 @@ print dnode2 openVnodes $data2_2
 if $data2_1 != 0 then
 	return -1
 endi
-if $data2_2 != 1 then
+if $data2_2 != 2 then
 	return -1
 endi
 
@@ -56,7 +56,25 @@ print $data30
 print $data40
 print $data50
 
-if $rows > 6 then
+print *num of tables $rows
+
+if $rows > 17 then
+	return -1
+endi
+
+sql show log.stables
+
+print $data00
+print $data10
+print $data20
+print $data30
+print $data40
+print $data50
+print $data60
+
+print *num of stables $rows
+
+if $rows > 7 then
 	return -1
 endi
 

tests/script/unique/dnode/monitor_bug.sim

@@ -19,7 +19,7 @@ sleep 3000
 
 sql show dnodes
 print dnode1 openVnodes $data2_1
-if $data2_1 > 2 then
+if $data2_1 > 4 then
 	return -1
 endi
 
@@ -41,7 +41,7 @@ print dnode2 openVnodes $data2_2
 if $data2_1 != 0 then
 	goto show2
 endi
-if $data2_2 > 2 then
+if $data2_2 > 4 then
 	goto show2
 endi
 
@@ -55,7 +55,7 @@ print $data30
 print $data40
 print $data50
 
-if $rows > 5 then
+if $rows > 14 then
 	return -1
 endi