xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat[cluster]: send monitor information in multi-process mode

This commit is contained in:
Shengliang Guan 2022-04-08 19:59:10 +08:00
parent 4105842574
commit b34ea72a04
13 changed files with 68 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
include/libs/monitor/monitor.h
View File

@ -143,7 +143,7 @@ typedef struct {
SMonVgroupInfo vgroup;
SMonGrantInfo grant;
SMonSysInfo sys;
SMonLogs logs;
SMonLogs log;
} SMonMmInfo;
int32_t tSerializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo);
@ -158,7 +158,7 @@ typedef struct {
SMonDiskInfo tfs;
SVnodesStat vstat;
SMonSysInfo sys;
SMonLogs logs;
SMonLogs log;
} SMonVmInfo;
int32_t tSerializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo);
@ -167,7 +167,7 @@ void tFreeSMonVmInfo(SMonVmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs logs;
SMonLogs log;
} SMonQmInfo;
int32_t tSerializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo);
@ -176,7 +176,7 @@ void tFreeSMonQmInfo(SMonQmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs logs;
SMonLogs log;
} SMonSmInfo;
int32_t tSerializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo);
@ -184,7 +184,7 @@ int32_t tDeserializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo);
void tFreeSMonSmInfo(SMonSmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs logs;
SMonLogs log;
} SMonBmInfo;
int32_t tSerializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo);

2
source/dnode/mgmt/bm/bmHandle.c
View File

@ -19,7 +19,7 @@
void bmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonBmInfo *bmInfo) {
if (pWrapper->procType == PROC_CHILD) {
dmGetMonitorSysInfo(&bmInfo->sys);
monGetLogs(&bmInfo->logs);
monGetLogs(&bmInfo->log);
}
}

10
source/dnode/mgmt/dm/dmWorker.c
View File

@ -119,9 +119,9 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
return -1;
}
SSingleWorkerCfg scfg = {.min = 1, .max = 1, .name = "dnode-status", .fp = (FItem)dmProcessQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->statusWorker, &scfg) != 0) {
dError("failed to start dnode status worker since %s", terrstr());
SSingleWorkerCfg scfg = {.min = 1, .max = 1, .name = "dnode-monitor", .fp = (FItem)dmProcessQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &scfg) != 0) {
dError("failed to start dnode monitor worker since %s", terrstr());
return -1;
}
@ -131,7 +131,7 @@ int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
void dmStopWorker(SDnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tSingleWorkerCleanup(&pMgmt->statusWorker);
tSingleWorkerCleanup(&pMgmt->monitorWorker);
if (pMgmt->threadId != NULL) {
taosDestoryThread(pMgmt->threadId);
@ -151,7 +151,7 @@ int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->statusWorker;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);

2
source/dnode/mgmt/inc/dmInt.h
View File

@ -32,7 +32,7 @@ typedef struct SDnodeMgmt {
TdThread *threadId;
SRWLatch latch;
SSingleWorker mgmtWorker;
SSingleWorker statusWorker;
SSingleWorker monitorWorker;
SMsgCb msgCb;
const char *path;
SDnode *pDnode;

2
source/dnode/mgmt/mm/mmHandle.c
View File

@ -21,7 +21,7 @@ void mmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonMmInfo *mmInfo) {
mndGetMonitorInfo(pMgmt->pMnode, &mmInfo->cluster, &mmInfo->vgroup, &mmInfo->grant);
if (pWrapper->procType == PROC_CHILD) {
dmGetMonitorSysInfo(&mmInfo->sys);
monGetLogs(&mmInfo->logs);
monGetLogs(&mmInfo->log);
}
}

2
source/dnode/mgmt/qm/qmHandle.c
View File

@ -19,7 +19,7 @@
void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo) {
if (pWrapper->procType == PROC_CHILD) {
dmGetMonitorSysInfo(&qmInfo->sys);
monGetLogs(&qmInfo->logs);
monGetLogs(&qmInfo->log);
}
}

2
source/dnode/mgmt/sm/smHandle.c
View File

@ -19,7 +19,7 @@
void smGetMonitorInfo(SMgmtWrapper *pWrapper, SMonSmInfo *smInfo) {
if (pWrapper->procType == PROC_CHILD) {
dmGetMonitorSysInfo(&smInfo->sys);
monGetLogs(&smInfo->logs);
monGetLogs(&smInfo->log);
}
}

2
source/dnode/mgmt/vm/vmHandle.c
View File

@ -34,7 +34,7 @@ void vmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonVmInfo *vmInfo) {
if (pWrapper->procType == PROC_CHILD) {
dmGetMonitorSysInfo(&vmInfo->sys);
monGetLogs(&vmInfo->logs);
monGetLogs(&vmInfo->log);
}
}

2
source/libs/monitor/inc/monInt.h
View File

@ -24,7 +24,7 @@ typedef struct {
int64_t curTime;
int64_t lastTime;
SJson *pJson;
SMonLogs logs;
SMonLogs log;
SMonDmInfo dmInfo;
SMonMmInfo mmInfo;
SMonVmInfo vmInfo;

32
source/libs/monitor/src/monMain.c
View File

@ -124,7 +124,7 @@ void monCleanup() {
static void monCleanupMonitorInfo(SMonInfo *pMonitor) {
tsMonitor.lastTime = pMonitor->curTime;
taosArrayDestroy(pMonitor->logs.logs);
taosArrayDestroy(pMonitor->log.logs);
tFreeSMonMmInfo(&pMonitor->mmInfo);
tFreeSMonVmInfo(&pMonitor->vmInfo);
tFreeSMonSmInfo(&pMonitor->smInfo);
@ -141,7 +141,7 @@ static SMonInfo *monCreateMonitorInfo() {
return NULL;
}
monGetLogs(&pMonitor->logs);
monGetLogs(&pMonitor->log);
taosThreadMutexLock(&tsMonitor.lock);
memcpy(&pMonitor->dmInfo, &tsMonitor.dmInfo, sizeof(SMonDmInfo));
@ -159,7 +159,7 @@ static SMonInfo *monCreateMonitorInfo() {
taosThreadMutexUnlock(&tsMonitor.lock);
pMonitor->pJson = tjsonCreateObject();
if (pMonitor->pJson == NULL || pMonitor->logs.logs == NULL) {
if (pMonitor->pJson == NULL || pMonitor->log.logs == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
monCleanupMonitorInfo(pMonitor);
return NULL;
@ -447,12 +447,12 @@ static void monGenLogJson(SMonInfo *pMonitor) {
if (pLogsJson == NULL) return;
SMonLogs *logs[6];
logs[0] = &pMonitor->logs;
logs[1] = &pMonitor->mmInfo.logs;
logs[2] = &pMonitor->vmInfo.logs;
logs[3] = &pMonitor->smInfo.logs;
logs[4] = &pMonitor->qmInfo.logs;
logs[5] = &pMonitor->bmInfo.logs;
logs[0] = &pMonitor->log;
logs[1] = &pMonitor->mmInfo.log;
logs[2] = &pMonitor->vmInfo.log;
logs[3] = &pMonitor->smInfo.log;
logs[4] = &pMonitor->qmInfo.log;
logs[5] = &pMonitor->bmInfo.log;
int32_t numOfErrorLogs = 0;
int32_t numOfInfoLogs = 0;
@ -460,17 +460,17 @@ static void monGenLogJson(SMonInfo *pMonitor) {
int32_t numOfTraceLogs = 0;
for (int32_t j = 0; j < 6; j++) {
SMonLogs *pLogs = logs[j];
numOfErrorLogs += pLogs->numOfErrorLogs;
numOfInfoLogs += pLogs->numOfInfoLogs;
numOfDebugLogs += pLogs->numOfDebugLogs;
numOfTraceLogs += pLogs->numOfTraceLogs;
SMonLogs *pLog = logs[j];
numOfErrorLogs += pLog->numOfErrorLogs;
numOfInfoLogs += pLog->numOfInfoLogs;
numOfDebugLogs += pLog->numOfDebugLogs;
numOfTraceLogs += pLog->numOfTraceLogs;
for (int32_t i = 0; i < taosArrayGetSize(pLogs->logs); ++i) {
for (int32_t i = 0; i < taosArrayGetSize(pLog->logs); ++i) {
SJson *pLogJson = tjsonCreateObject();
if (pLogJson == NULL) continue;
SMonLogItem *pLogItem = taosArrayGet(pLogs->logs, i);
SMonLogItem *pLogItem = taosArrayGet(pLog->logs, i);
char buf[40] = {0};
taosFormatUtcTime(buf, sizeof(buf), pLogItem->ts, TSDB_TIME_PRECISION_MILLI);

48
source/libs/monitor/src/monMsg.c
View File

@ -226,7 +226,7 @@ int32_t tSerializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo) {
if (tEncodeSMonVgroupInfo(&encoder, &pInfo->vgroup) < 0) return -1;
if (tEncodeSMonGrantInfo(&encoder, &pInfo->grant) < 0) return -1;
if (tEncodeSMonSysInfo(&encoder, &pInfo->sys) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->logs) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->log) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -243,7 +243,7 @@ int32_t tDeserializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo) {
if (tDecodeSMonVgroupInfo(&decoder, &pInfo->vgroup) < 0) return -1;
if (tDecodeSMonGrantInfo(&decoder, &pInfo->grant) < 0) return -1;
if (tDecodeSMonSysInfo(&decoder, &pInfo->sys) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->logs) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->log) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -251,8 +251,14 @@ int32_t tDeserializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo) {
}
void tFreeSMonMmInfo(SMonMmInfo *pInfo) {
taosArrayDestroy(pInfo->logs.logs);
pInfo->logs.logs = NULL;
taosArrayDestroy(pInfo->log.logs);
taosArrayDestroy(pInfo->cluster.mnodes);
taosArrayDestroy(pInfo->cluster.dnodes);
taosArrayDestroy(pInfo->vgroup.vgroups);
pInfo->cluster.mnodes = NULL;
pInfo->cluster.dnodes = NULL;
pInfo->vgroup.vgroups = NULL;
pInfo->log.logs = NULL;
}
int32_t tEncodeSMonDiskDesc(SCoder *encoder, const SMonDiskDesc *pDesc) {
@ -331,7 +337,7 @@ int32_t tSerializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo) {
if (tEncodeSMonDiskInfo(&encoder, &pInfo->tfs) < 0) return -1;
if (tEncodeSVnodesStat(&encoder, &pInfo->vstat) < 0) return -1;
if (tEncodeSMonSysInfo(&encoder, &pInfo->sys) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->logs) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->log) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -347,7 +353,7 @@ int32_t tDeserializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo) {
if (tDecodeSMonDiskInfo(&decoder, &pInfo->tfs) < 0) return -1;
if (tDecodeSVnodesStat(&decoder, &pInfo->vstat) < 0) return -1;
if (tDecodeSMonSysInfo(&decoder, &pInfo->sys) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->logs) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->log) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -355,8 +361,10 @@ int32_t tDeserializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo) {
}
void tFreeSMonVmInfo(SMonVmInfo *pInfo) {
taosArrayDestroy(pInfo->logs.logs);
pInfo->logs.logs = NULL;
taosArrayDestroy(pInfo->log.logs);
taosArrayDestroy(pInfo->tfs.datadirs);
pInfo->log.logs = NULL;
pInfo->tfs.datadirs = NULL;
}
int32_t tSerializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo) {
@ -365,7 +373,7 @@ int32_t tSerializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo) {
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeSMonSysInfo(&encoder, &pInfo->sys) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->logs) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->log) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -379,7 +387,7 @@ int32_t tDeserializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo) {
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeSMonSysInfo(&decoder, &pInfo->sys) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->logs) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->log) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -387,8 +395,8 @@ int32_t tDeserializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo) {
}
void tFreeSMonQmInfo(SMonQmInfo *pInfo) {
taosArrayDestroy(pInfo->logs.logs);
pInfo->logs.logs = NULL;
taosArrayDestroy(pInfo->log.logs);
pInfo->log.logs = NULL;
}
int32_t tSerializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo) {
@ -397,7 +405,7 @@ int32_t tSerializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo) {
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeSMonSysInfo(&encoder, &pInfo->sys) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->logs) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->log) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -411,7 +419,7 @@ int32_t tDeserializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo) {
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeSMonSysInfo(&decoder, &pInfo->sys) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->logs) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->log) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -419,8 +427,8 @@ int32_t tDeserializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo) {
}
void tFreeSMonSmInfo(SMonSmInfo *pInfo) {
taosArrayDestroy(pInfo->logs.logs);
pInfo->logs.logs = NULL;
taosArrayDestroy(pInfo->log.logs);
pInfo->log.logs = NULL;
}
int32_t tSerializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo) {
@ -429,7 +437,7 @@ int32_t tSerializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo) {
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeSMonSysInfo(&encoder, &pInfo->sys) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->logs) < 0) return -1;
if (tEncodeSMonLogs(&encoder, &pInfo->log) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -443,7 +451,7 @@ int32_t tDeserializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo) {
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeSMonSysInfo(&decoder, &pInfo->sys) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->logs) < 0) return -1;
if (tDecodeSMonLogs(&decoder, &pInfo->log) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
@ -451,8 +459,8 @@ int32_t tDeserializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo) {
}
void tFreeSMonBmInfo(SMonBmInfo *pInfo) {
taosArrayDestroy(pInfo->logs.logs);
pInfo->logs.logs = NULL;
taosArrayDestroy(pInfo->log.logs);
pInfo->log.logs = NULL;
}
int32_t tSerializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo) {

10
source/libs/monitor/test/monTest.cpp
View File

@ -248,25 +248,25 @@ TEST_F(MonitorTest, 01_Full) {
GetVgroupInfo(&mmInfo.vgroup);
GetGrantInfo(&mmInfo.grant);
GetSysInfo(&mmInfo.sys);
GetLogInfo(&mmInfo.logs);
GetLogInfo(&mmInfo.log);
SMonVmInfo vmInfo = {0};
GetDiskInfo(&vmInfo.tfs);
GetVnodeStat(&vmInfo.vstat);
GetSysInfo(&vmInfo.sys);
GetLogInfo(&vmInfo.logs);
GetLogInfo(&vmInfo.log);
SMonQmInfo qmInfo = {0};
GetSysInfo(&qmInfo.sys);
GetLogInfo(&qmInfo.logs);
GetLogInfo(&qmInfo.log);
SMonSmInfo smInfo = {0};
GetSysInfo(&smInfo.sys);
GetLogInfo(&smInfo.logs);
GetLogInfo(&smInfo.log);
SMonBmInfo bmInfo = {0};
GetSysInfo(&bmInfo.sys);
GetLogInfo(&bmInfo.logs);
GetLogInfo(&bmInfo.log);
monSetDmInfo(&dmInfo);
monSetMmInfo(&mmInfo);

4
source/os/src/osSysinfo.c
View File

@ -623,7 +623,7 @@ void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, i
static int64_t cur_wchars = 0;
static int64_t cur_read_bytes = 0;
static int64_t cur_write_bytes = 0;
if (taosGetProcIO(&cur_rchars, &cur_wchars, &cur_read_bytes, &cur_write_bytes) != 0) {
if (taosGetProcIO(&cur_rchars, &cur_wchars, &cur_read_bytes, &cur_write_bytes) == 0) {
*rchars = cur_rchars - last_rchars;
*wchars = cur_wchars - last_wchars;
*read_bytes = cur_read_bytes - last_read_bytes;
@ -696,7 +696,7 @@ void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
static int64_t cur_receive_bytes = 0;
static int64_t cur_transmit_bytes = 0;
if (taosGetCardInfo(&cur_receive_bytes, &cur_transmit_bytes) != 0) {
if (taosGetCardInfo(&cur_receive_bytes, &cur_transmit_bytes) == 0) {
*receive_bytes = cur_receive_bytes - last_receive_bytes;
*transmit_bytes = cur_transmit_bytes - last_transmit_bytes;
last_receive_bytes = cur_receive_bytes;