monitor

include/common/tglobal.h

@@ -57,6 +57,7 @@ extern bool     tsEnableMonitor;
 extern int32_t  tsMonitorInterval;
 extern char     tsMonitorFqdn[];
 extern uint16_t tsMonitorPort;
+extern int32_t  tsMonitorMaxLogs;
 
 // query buffer management
 extern int32_t tsQueryBufferSize;  // maximum allowed usage buffer size in MB for each data node during query processing

include/libs/monitor/monitor.h

@@ -23,8 +23,6 @@
 extern "C" {
 #endif
 
-typedef struct SMonitor SMonitor;
-
 typedef struct {
   int32_t dnode_id;
   char    dnode_ep[TSDB_EP_LEN];
@@ -122,16 +120,34 @@ typedef struct {
   char    content[1024];
 } SMonLogItem;
 
-typedef struct {
-  SArray *logs;  // array of SMonLogItem
-} SMonLogInfo;
-
 typedef struct {
   int32_t expire_time;
   int32_t timeseries_used;
   int32_t timeseries_total;
 } SMonGrantInfo;
 
+typedef struct SMonInfo SMonInfo;
+
+typedef struct {
+  const char *server;
+  uint16_t    port;
+  int32_t     maxLogs;
+} SMonCfg;
+
+int32_t monInit(const SMonCfg *pCfg);
+void    monCleanup();
+void    monAddLogItem(SMonLogItem *pItem);
+
+SMonInfo *monCreateMonitorInfo();
+void      monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo);
+void      monSetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo);
+void      monSetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo);
+void      monSetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo);
+void      monSetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);
+void      monSetGrantInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);
+void      monSendReport(SMonInfo *pMonitor);
+void      monCleanupMonitorInfo(SMonInfo *pMonitor);
+
 #ifdef __cplusplus
 }
 #endif

source/common/src/tglobal.c

@@ -51,6 +51,7 @@ bool     tsEnableMonitor = 1;
 int32_t  tsMonitorInterval = 5;
 char     tsMonitorFqdn[TSDB_FQDN_LEN] = {0};
 uint16_t tsMonitorPort = 6043;
+int32_t  tsMonitorMaxLogs = 100;
 
 /*
  * denote if the server needs to compress response message at the application layer to client, including query rsp,
@@ -325,6 +326,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 360000, 0) != 0) return -1;
   if (cfgAddString(pCfg, "monitorFqdn", tsMonitorFqdn, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "monitorPort", tsMonitorPort, 1, 65056, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "monitorMaxLogs", tsMonitorMaxLogs, 1, 1000000, 0) != 0) return -1;
 
   return 0;
 }
@@ -441,6 +443,7 @@ static void taosSetServerCfg(SConfig *pCfg) {
   tsMonitorInterval = cfgGetItem(pCfg, "monitorInterval")->i32;
   tstrncpy(tsMonitorFqdn, cfgGetItem(pCfg, "monitorFqdn")->str, TSDB_FQDN_LEN);
   tsMonitorPort = (uint16_t)cfgGetItem(pCfg, "monitorPort")->i32;
+  tsMonitorMaxLogs = cfgGetItem(pCfg, "monitorMaxLogs")->i32;
 
   if (tsQueryBufferSize >= 0) {
     tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;

source/dnode/mgmt/impl/CMakeLists.txt

@@ -12,6 +12,7 @@ target_link_libraries(
     PUBLIC sync
     PUBLIC taos
     PUBLIC tfs
+    PUBLIC monitor
 )
 target_include_directories(
     dnode

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

@@ -21,6 +21,7 @@
 #include "dndSnode.h"
 #include "dndTransport.h"
 #include "dndVnodes.h"
+#include "monitor.h"
 #include "sync.h"
 #include "tfs.h"
 #include "wal.h"
@@ -297,6 +298,13 @@ int32_t dndInit() {
     return -1;
   }
 
+  SMonCfg monCfg = {.maxLogs = tsMonitorMaxLogs, .port = tsMonitorPort, .server = tsMonitorFqdn};
+  if (monInit(&monCfg) != 0) {
+    dError("failed to init monitor since %s", terrstr());
+    dndCleanup();
+    return -1;
+  }
+
   dInfo("dnode env is initialized");
   return 0;
 }
@@ -310,6 +318,7 @@ void dndCleanup() {
   walCleanUp();
   vnodeCleanup();
   rpcCleanup();
+  monCleanup();
 
   taosStopCacheRefreshWorker();
   dInfo("dnode env is cleaned up");

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

@@ -22,6 +22,7 @@
 #include "dndTransport.h"
 #include "dndVnodes.h"
 #include "dndWorker.h"
+#include "monitor.h"
 
 static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg);
 
@@ -475,8 +476,17 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
 
 static void dndSendMonitorReport(SDnode *pDnode) {
   if (!tsEnableMonitor || tsMonitorFqdn[0] == 0) return;
+  SMonInfo *pMonitor = monCreateMonitorInfo();
+  if (pMonitor == NULL) return;
 
   dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort);
+
+  SMonBasicInfo basicInfo = {.dnode_id = dndGetDnodeId(pDnode)};
+  tstrncpy(basicInfo.dnode_ep, tsLocalEp, TSDB_EP_LEN);
+  monSetBasicInfo(pMonitor, &basicInfo);
+
+  monSendReport(pMonitor);
+  monCleanupMonitorInfo(pMonitor);
 }
 
 static void *dnodeThreadRoutine(void *param) {

source/libs/monitor/inc/monInt.h

@@ -18,6 +18,23 @@
 
 #include "monitor.h"
 
+#include "tarray.h"
+#include "tlockfree.h"
+#include "tjson.h"
+
+typedef struct {
+  SRWLatch    lock;
+  SArray     *logs;  // array of SMonLogItem
+  int32_t     maxLogs;
+  const char *server;
+  uint16_t    port;
+} SMonitor;
+
+typedef struct SMonInfo {
+  SArray *logs;  // array of SMonLogItem
+  SJson  *pJson;
+} SMonInfo;
+
 #ifdef __cplusplus
 }
 #endif

source/libs/monitor/src/monitor.c

@@ -15,4 +15,83 @@
 
 #define _DEFAULT_SOURCE
 #include "monInt.h"
+#include "taoserror.h"
+#include "thttp.h"
+#include "tlog.h"
 
+static SMonitor tsMonitor = {0};
+
+int32_t monInit(const SMonCfg *pCfg) {
+  tsMonitor.logs = taosArrayInit(16, sizeof(SMonInfo));
+  if (tsMonitor.logs == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  tsMonitor.maxLogs = pCfg->maxLogs;
+  tsMonitor.server = pCfg->server;
+  tsMonitor.port = pCfg->port;
+  taosInitRWLatch(&tsMonitor.lock);
+  return 0;
+}
+
+void monCleanup() {
+  taosArrayDestroy(tsMonitor.logs);
+  tsMonitor.logs = NULL;
+}
+
+void monAddLogItem(SMonLogItem *pItem) {
+  taosWLockLatch(&tsMonitor.lock);
+  int32_t size = taosArrayGetSize(tsMonitor.logs);
+  if (size > tsMonitor.maxLogs) {
+    uInfo("too many logs for monitor");
+  } else {
+    taosArrayPush(tsMonitor.logs, pItem);
+  }
+  taosWUnLockLatch(&tsMonitor.lock);
+}
+
+SMonInfo *monCreateMonitorInfo() {
+  SMonInfo *pMonitor = calloc(1, sizeof(SMonInfo));
+  if (pMonitor == NULL) return NULL;
+
+  taosWLockLatch(&tsMonitor.lock);
+  pMonitor->logs = taosArrayDup(pMonitor->logs);
+  taosArrayClear(tsMonitor.logs);
+  taosWUnLockLatch(&tsMonitor.lock);
+
+  pMonitor->pJson = tjsonCreateObject();
+  if (pMonitor->pJson == NULL || pMonitor->logs == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    monCleanupMonitorInfo(pMonitor);
+    return NULL;
+  }
+
+  return pMonitor;
+}
+
+void monCleanupMonitorInfo(SMonInfo *pMonitor) {
+  taosArrayDestroy(pMonitor->logs);
+  tjsonDelete(pMonitor->pJson);
+  free(pMonitor);
+}
+
+void monSendReport(SMonInfo *pMonitor) {
+  char *pCont = tjsonToString(pMonitor->pJson);
+  if (pCont != NULL) {
+    taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont));
+    free(pCont);
+  }
+}
+
+void monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo) {
+  SJson *pJson = pMonitor->pJson;
+  tjsonAddDoubleToObject(pJson, "dnode_id", pInfo->dnode_id);
+  tjsonAddStringToObject(pJson, "dnode_ep", pInfo->dnode_ep);
+}
+
+void monSetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo);
+void monSetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo);
+void monSetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo);
+void monSetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);
+void monSetGrantInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);

source/util/src/tjson.c

@@ -26,7 +26,11 @@ SJson* tjsonCreateObject() {
   return pJson;
 }
 
-void tjsonDelete(SJson* pJson) { cJSON_Delete((cJSON*)pJson); }
+void tjsonDelete(SJson* pJson) {
+  if (pJson != NULL) {
+    cJSON_Delete((cJSON*)pJson);
+  }
+}
 
 int32_t tjsonAddIntegerToObject(SJson* pJson, const char* pName, const uint64_t number) {
   char tmp[40] = {0};