未验证 提交 284f84ac 编写于 作者: S Shengliang Guan 提交者: GitHub

Merge pull request #11334 from taosdata/feature/shm

feat[cluster]: send monitor information in multi-process mode
......@@ -717,6 +717,18 @@ typedef struct {
char charset[TD_LOCALE_LEN]; // tsCharset
} SClusterCfg;
typedef struct {
int32_t openVnodes;
int32_t totalVnodes;
int32_t masterNum;
int64_t numOfSelectReqs;
int64_t numOfInsertReqs;
int64_t numOfInsertSuccessReqs;
int64_t numOfBatchInsertReqs;
int64_t numOfBatchInsertSuccessReqs;
int64_t errors;
} SVnodesStat;
typedef struct {
int32_t vgId;
int8_t role;
......
......@@ -218,6 +218,15 @@ enum {
TD_NEW_MSG_SEG(TDMT_SCH_MSG)
TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "scheduler-link-broken", NULL, NULL)
// Monitor info exchange between processes
TD_NEW_MSG_SEG(TDMT_MON_MSG)
TD_DEF_MSG_TYPE(TDMT_MON_MM_INFO, "monitor-minfo", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MON_VM_INFO, "monitor-vinfo", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MON_QM_INFO, "monitor-qinfo", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MON_SM_INFO, "monitor-sinfo", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MON_BM_INFO, "monitor-binfo", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MON_VM_LOAD, "monitor-vload", NULL, NULL)
#if defined(TD_MSG_NUMBER_)
TDMT_MAX
#endif
......
......@@ -19,6 +19,7 @@
#include "tarray.h"
#include "tdef.h"
#include "tlog.h"
#include "tmsg.h"
#ifdef __cplusplus
extern "C" {
......@@ -29,6 +30,44 @@ extern "C" {
#define MON_VER_LEN 12
#define MON_LOG_LEN 1024
typedef struct {
int64_t ts;
ELogLevel level;
char content[MON_LOG_LEN];
} SMonLogItem;
typedef struct {
SArray *logs; // array of SMonLogItem
int32_t numOfErrorLogs;
int32_t numOfInfoLogs;
int32_t numOfDebugLogs;
int32_t numOfTraceLogs;
} SMonLogs;
typedef struct {
char name[TSDB_FILENAME_LEN];
int8_t level;
SDiskSize size;
} SMonDiskDesc;
typedef struct {
double cpu_engine;
double cpu_system;
float cpu_cores;
int64_t mem_engine; // KB
int64_t mem_system; // KB
int64_t mem_total; // KB
int64_t disk_engine; // Byte
int64_t disk_used; // Byte
int64_t disk_total; // Byte
int64_t net_in; // bytes
int64_t net_out; // bytes
int64_t io_read; // bytes
int64_t io_write; // bytes
int64_t io_read_disk; // bytes
int64_t io_write_disk; // bytes
} SMonSysInfo;
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
......@@ -36,6 +75,19 @@ typedef struct {
int32_t protocol;
} SMonBasicInfo;
typedef struct {
float uptime; // day
int8_t has_mnode;
SMonDiskDesc logdir;
SMonDiskDesc tempdir;
} SMonDnodeInfo;
typedef struct {
SMonBasicInfo basic;
SMonDnodeInfo dnode;
SMonSysInfo sys;
} SMonDmInfo;
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
......@@ -87,46 +139,65 @@ typedef struct {
} SMonGrantInfo;
typedef struct {
float uptime; // day
double cpu_engine;
double cpu_system;
float cpu_cores;
int64_t mem_engine; // KB
int64_t mem_system; // KB
int64_t mem_total; // KB
int64_t disk_engine; // Byte
int64_t disk_used; // Byte
int64_t disk_total; // Byte
int64_t net_in; // bytes
int64_t net_out; // bytes
int64_t io_read; // bytes
int64_t io_write; // bytes
int64_t io_read_disk; // bytes
int64_t io_write_disk; // bytes
int64_t req_select;
int64_t req_insert;
int64_t req_insert_success;
int64_t req_insert_batch;
int64_t req_insert_batch_success;
int32_t errors;
int32_t vnodes_num;
int32_t masters;
int8_t has_mnode;
} SMonDnodeInfo;
SMonClusterInfo cluster;
SMonVgroupInfo vgroup;
SMonGrantInfo grant;
SMonSysInfo sys;
SMonLogs log;
} SMonMmInfo;
typedef struct {
char name[TSDB_FILENAME_LEN];
int8_t level;
SDiskSize size;
} SMonDiskDesc;
int32_t tSerializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo);
int32_t tDeserializeSMonMmInfo(void *buf, int32_t bufLen, SMonMmInfo *pInfo);
void tFreeSMonMmInfo(SMonMmInfo *pInfo);
typedef struct {
SArray *datadirs; // array of SMonDiskDesc
SMonDiskDesc logdir;
SMonDiskDesc tempdir;
SArray *datadirs; // array of SMonDiskDesc
} SMonDiskInfo;
typedef struct SMonInfo SMonInfo;
typedef struct {
SMonDiskInfo tfs;
SVnodesStat vstat;
SMonSysInfo sys;
SMonLogs log;
} SMonVmInfo;
int32_t tSerializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo);
int32_t tDeserializeSMonVmInfo(void *buf, int32_t bufLen, SMonVmInfo *pInfo);
void tFreeSMonVmInfo(SMonVmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs log;
} SMonQmInfo;
int32_t tSerializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo);
int32_t tDeserializeSMonQmInfo(void *buf, int32_t bufLen, SMonQmInfo *pInfo);
void tFreeSMonQmInfo(SMonQmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs log;
} SMonSmInfo;
int32_t tSerializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo);
int32_t tDeserializeSMonSmInfo(void *buf, int32_t bufLen, SMonSmInfo *pInfo);
void tFreeSMonSmInfo(SMonSmInfo *pInfo);
typedef struct {
SMonSysInfo sys;
SMonLogs log;
} SMonBmInfo;
int32_t tSerializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo);
int32_t tDeserializeSMonBmInfo(void *buf, int32_t bufLen, SMonBmInfo *pInfo);
void tFreeSMonBmInfo(SMonBmInfo *pInfo);
typedef struct {
SArray *pVloads; // SVnodeLoad
} SMonVloadInfo;
int32_t tSerializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo);
int32_t tDeserializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo);
void tFreeSMonVloadInfo(SMonVloadInfo *pInfo);
typedef struct {
const char *server;
......@@ -138,16 +209,14 @@ typedef struct {
int32_t monInit(const SMonCfg *pCfg);
void monCleanup();
void monRecordLog(int64_t ts, ELogLevel level, const char *content);
SMonInfo *monCreateMonitorInfo();
void monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo);
void monSetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo);
void monSetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);
void monSetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo);
void monSetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo);
void monSetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo);
void monSendReport(SMonInfo *pMonitor);
void monCleanupMonitorInfo(SMonInfo *pMonitor);
int32_t monGetLogs(SMonLogs *logs);
void monSetDmInfo(SMonDmInfo *pInfo);
void monSetMmInfo(SMonMmInfo *pInfo);
void monSetVmInfo(SMonVmInfo *pInfo);
void monSetQmInfo(SMonQmInfo *pInfo);
void monSetSmInfo(SMonSmInfo *pInfo);
void monSetBmInfo(SMonBmInfo *pInfo);
void monSendReport();
#ifdef __cplusplus
}
......
......@@ -44,8 +44,8 @@ int32_t taosGetTotalMemory(int64_t *totalKB);
int32_t taosGetProcMemory(int64_t *usedKB);
int32_t taosGetSysMemory(int64_t *usedKB);
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize);
int32_t taosGetProcIO(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes);
int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes);
void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes);
void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes);
void taosKillSystem();
int32_t taosGetSystemUUID(char *uid, int32_t uidlen);
......
......@@ -16,6 +16,34 @@
#define _DEFAULT_SOURCE
#include "bmInt.h"
void bmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonBmInfo *bmInfo) {
}
int32_t bmProcessGetMonBmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonBmInfo bmInfo = {0};
bmGetMonitorInfo(pWrapper, &bmInfo);
dmGetMonitorSysInfo(&bmInfo.sys);
monGetLogs(&bmInfo.log);
int32_t rspLen = tSerializeSMonBmInfo(NULL, 0, &bmInfo);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonBmInfo(pRsp, rspLen, &bmInfo);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonBmInfo(&bmInfo);
return 0;
}
int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnode *pDnode = pWrapper->pDnode;
SRpcMsg *pReq = &pMsg->rpcMsg;
......@@ -54,4 +82,6 @@ int32_t bmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
}
void bmInitMsgHandle(SMgmtWrapper *pWrapper) {}
void bmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_MON_BM_INFO, bmProcessMonitorMsg, DEFAULT_HANDLE);
}
......@@ -33,7 +33,37 @@ static void bmSendErrorRsps(SMgmtWrapper *pWrapper, STaosQall *qall, int32_t num
}
}
static void bmProcessQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
static inline void bmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle,
.code = code,
.pCont = pMsg->pRsp,
.contLen = pMsg->rspLen};
tmsgSendRsp(&rsp);
}
static void bmProcessMonQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SBnodeMgmt *pMgmt = pInfo->ahandle;
dTrace("msg:%p, get from bnode monitor queue", pMsg);
SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1;
if (pMsg->rpcMsg.msgType == TDMT_MON_BM_INFO) {
code = bmProcessGetMonBmInfoReq(pMgmt->pWrapper, pMsg);
}
if (pRpc->msgType & 1U) {
if (code != 0 && terrno != 0) code = terrno;
bmSendRsp(pMgmt->pWrapper, pMsg, code);
}
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
rpcFreeCont(pRpc->pCont);
taosFreeQitem(pMsg);
}
static void bmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SBnodeMgmt *pMgmt = pInfo->ahandle;
SMgmtWrapper *pWrapper = pMgmt->pWrapper;
......@@ -72,18 +102,37 @@ int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SBnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
int32_t bmStartWorker(SBnodeMgmt *pMgmt) {
SMultiWorkerCfg cfg = {.max = 1, .name = "bnode-write", .fp = (FItems)bmProcessQueue, .param = pMgmt};
SMultiWorkerCfg cfg = {.max = 1, .name = "bnode-write", .fp = (FItems)bmProcessWriteQueue, .param = pMgmt};
if (tMultiWorkerInit(&pMgmt->writeWorker, &cfg) != 0) {
dError("failed to start bnode write worker since %s", terrstr());
dError("failed to start bnode-write worker since %s", terrstr());
return -1;
}
if (tsMultiProcess) {
SSingleWorkerCfg mCfg = {
.min = 1, .max = 1, .name = "bnode-monitor", .fp = (FItem)bmProcessMonQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
dError("failed to start bnode-monitor worker since %s", terrstr());
return -1;
}
}
dDebug("bnode workers are initialized");
return 0;
}
void bmStopWorker(SBnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->monitorWorker);
tMultiWorkerCleanup(&pMgmt->writeWorker);
dDebug("bnode workers are closed");
}
......@@ -42,8 +42,9 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
if (pWrapper != NULL) {
req.pVloads = taosArrayInit(TSDB_MAX_VNODES, sizeof(SVnodeLoad));
vmMonitorVnodeLoads(pWrapper, req.pVloads);
SMonVloadInfo info = {0};
dmGetVnodeLoads(pWrapper, &info);
req.pVloads = info.pVloads;
dndReleaseWrapper(pWrapper);
}
......@@ -117,7 +118,6 @@ int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
static int32_t dmProcessCreateNodeMsg(SDnode *pDnode, EDndType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
......@@ -209,7 +209,7 @@ void dmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_DND_NETWORK_TEST, dmProcessMgmtMsg, DEFAULT_HANDLE);
// Requests handled by MNODE
dndSetMsgHandle(pWrapper, TDMT_MND_STATUS_RSP, dmProcessStatusMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_STATUS_RSP, dmProcessMonitorMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
}
......@@ -14,21 +14,7 @@
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
static int32_t dmGetMonitorDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo) {
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size;
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
if (pWrapper != NULL) {
vmMonitorTfsInfo(pWrapper, pInfo);
dndReleaseWrapper(pWrapper);
}
return 0;
}
#include "dmInt.h"
static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->protocol = 1;
......@@ -39,69 +25,171 @@ static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
static void dmGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pDnode->rebootTime) / (86400000.0f);
taosGetCpuUsage(&pInfo->cpu_engine, &pInfo->cpu_system);
taosGetCpuCores(&pInfo->cpu_cores);
taosGetProcMemory(&pInfo->mem_engine);
taosGetSysMemory(&pInfo->mem_system);
pInfo->mem_total = tsTotalMemoryKB;
pInfo->disk_engine = 0;
pInfo->disk_used = tsDataSpace.size.used;
pInfo->disk_total = tsDataSpace.size.total;
taosGetCardInfo(&pInfo->net_in, &pInfo->net_out);
taosGetProcIO(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, VNODES);
if (pWrapper != NULL) {
vmMonitorVnodeReqs(pWrapper, pInfo);
dndReleaseWrapper(pWrapper);
}
pWrapper = dndAcquireWrapper(pDnode, MNODE);
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, MNODE);
if (pWrapper != NULL) {
pInfo->has_mnode = pWrapper->required;
dndReleaseWrapper(pWrapper);
}
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size;
}
static void dmGetMonitorInfo(SDnode *pDnode, SMonDmInfo *pInfo) {
dmGetMonitorBasicInfo(pDnode, &pInfo->basic);
dmGetMonitorSysInfo(&pInfo->sys);
dmGetMonitorDnodeInfo(pDnode, &pInfo->dnode);
}
void dmSendMonitorReport(SDnode *pDnode) {
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
dTrace("send monitor report to %s:%u", tsMonitorFqdn, tsMonitorPort);
SMonInfo *pMonitor = monCreateMonitorInfo();
if (pMonitor == NULL) return;
SMonBasicInfo basicInfo = {0};
dmGetMonitorBasicInfo(pDnode, &basicInfo);
monSetBasicInfo(pMonitor, &basicInfo);
SMonDmInfo dmInfo = {0};
SMonMmInfo mmInfo = {0};
SMonVmInfo vmInfo = {0};
SMonQmInfo qmInfo = {0};
SMonSmInfo smInfo = {0};
SMonBmInfo bmInfo = {0};
SRpcMsg req = {0};
SRpcMsg rsp;
SEpSet epset = {.inUse = 0, .numOfEps = 1};
tstrncpy(epset.eps[0].fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
epset.eps[0].port = tsServerPort;
SMgmtWrapper *pWrapper = NULL;
dmGetMonitorInfo(pDnode, &dmInfo);
bool getFromAPI = !tsMultiProcess;
pWrapper = &pDnode->wrappers[MNODE];
if (getFromAPI) {
if (dndMarkWrapper(pWrapper) != 0) {
mmGetMonitorInfo(pWrapper, &mmInfo);
dndReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_MM_INFO;
dndSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonMmInfo(rsp.pCont, rsp.contLen, &mmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
SMonClusterInfo clusterInfo = {0};
SMonVgroupInfo vgroupInfo = {0};
SMonGrantInfo grantInfo = {0};
pWrapper = &pDnode->wrappers[VNODES];
if (getFromAPI) {
if (dndMarkWrapper(pWrapper) != 0) {
vmGetMonitorInfo(pWrapper, &vmInfo);
dndReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_VM_INFO;
dndSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonVmInfo(rsp.pCont, rsp.contLen, &vmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, MNODE);
if (pWrapper != NULL) {
if (mmMonitorMnodeInfo(pWrapper, &clusterInfo, &vgroupInfo, &grantInfo) == 0) {
monSetClusterInfo(pMonitor, &clusterInfo);
monSetVgroupInfo(pMonitor, &vgroupInfo);
monSetGrantInfo(pMonitor, &grantInfo);
pWrapper = &pDnode->wrappers[QNODE];
if (getFromAPI) {
if (dndMarkWrapper(pWrapper) != 0) {
qmGetMonitorInfo(pWrapper, &qmInfo);
dndReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_QM_INFO;
dndSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonQmInfo(rsp.pCont, rsp.contLen, &qmInfo);
}
rpcFreeCont(rsp.pCont);
}
dndReleaseWrapper(pWrapper);
}
SMonDnodeInfo dnodeInfo = {0};
dmGetMonitorDnodeInfo(pDnode, &dnodeInfo);
monSetDnodeInfo(pMonitor, &dnodeInfo);
pWrapper = &pDnode->wrappers[SNODE];
if (getFromAPI) {
if (dndMarkWrapper(pWrapper) != 0) {
smGetMonitorInfo(pWrapper, &smInfo);
dndReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_SM_INFO;
dndSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonSmInfo(rsp.pCont, rsp.contLen, &smInfo);
}
rpcFreeCont(rsp.pCont);
}
}
SMonDiskInfo diskInfo = {0};
if (dmGetMonitorDiskInfo(pDnode, &diskInfo) == 0) {
monSetDiskInfo(pMonitor, &diskInfo);
pWrapper = &pDnode->wrappers[BNODE];
if (getFromAPI) {
if (dndMarkWrapper(pWrapper) != 0) {
bmGetMonitorInfo(pWrapper, &bmInfo);
dndReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_BM_INFO;
dndSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonBmInfo(rsp.pCont, rsp.contLen, &bmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
taosArrayDestroy(clusterInfo.dnodes);
taosArrayDestroy(clusterInfo.mnodes);
taosArrayDestroy(vgroupInfo.vgroups);
taosArrayDestroy(diskInfo.datadirs);
monSetDmInfo(&dmInfo);
monSetMmInfo(&mmInfo);
monSetVmInfo(&vmInfo);
monSetQmInfo(&qmInfo);
monSetSmInfo(&smInfo);
monSetBmInfo(&bmInfo);
tFreeSMonMmInfo(&mmInfo);
tFreeSMonVmInfo(&vmInfo);
tFreeSMonQmInfo(&qmInfo);
tFreeSMonSmInfo(&smInfo);
tFreeSMonBmInfo(&bmInfo);
monSendReport();
}
void dmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo) {
bool getFromAPI = !tsMultiProcess;
if (getFromAPI) {
vmGetVnodeLoads(pWrapper, pInfo);
} else {
SRpcMsg req = {.msgType = TDMT_MON_VM_LOAD};
SRpcMsg rsp = {0};
SEpSet epset = {.inUse = 0, .numOfEps = 1};
tstrncpy(epset.eps[0].fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
epset.eps[0].port = tsServerPort;
dndSendRecv(pWrapper->pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonVloadInfo(rsp.pCont, rsp.contLen, pInfo);
}
rpcFreeCont(rsp.pCont);
}
}
monSendReport(pMonitor);
monCleanupMonitorInfo(pMonitor);
}
\ No newline at end of file
void dmGetMonitorSysInfo(SMonSysInfo *pInfo) {
taosGetCpuUsage(&pInfo->cpu_engine, &pInfo->cpu_system);
taosGetCpuCores(&pInfo->cpu_cores);
taosGetProcMemory(&pInfo->mem_engine);
taosGetSysMemory(&pInfo->mem_system);
pInfo->mem_total = tsTotalMemoryKB;
pInfo->disk_engine = 0;
pInfo->disk_used = tsDataSpace.size.used;
pInfo->disk_total = tsDataSpace.size.total;
taosGetCardInfoDelta(&pInfo->net_in, &pInfo->net_out);
taosGetProcIODelta(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
}
......@@ -101,9 +101,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;
}
......@@ -113,7 +113,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);
......@@ -131,9 +131,9 @@ int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t dmProcessStatusMsg(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);
......
......@@ -17,6 +17,7 @@
#define _TD_DND_BNODE_INT_H_
#include "dndInt.h"
#include "bnode.h"
#ifdef __cplusplus
......@@ -29,6 +30,7 @@ typedef struct SBnodeMgmt {
SMgmtWrapper *pWrapper;
const char *path;
SMultiWorker writeWorker;
SSingleWorker monitorWorker;
} SBnodeMgmt;
// bmInt.c
......@@ -39,11 +41,13 @@ int32_t bmDrop(SMgmtWrapper *pWrapper);
void bmInitMsgHandle(SMgmtWrapper *pWrapper);
int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t bmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t bmProcessGetMonBmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
// bmWorker.c
int32_t bmStartWorker(SBnodeMgmt *pMgmt);
void bmStopWorker(SBnodeMgmt *pMgmt);
int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus
}
......
......@@ -32,7 +32,7 @@ typedef struct SDnodeMgmt {
TdThread *threadId;
SRWLatch latch;
SSingleWorker mgmtWorker;
SSingleWorker statusWorker;
SSingleWorker monitorWorker;
SMsgCb msgCb;
const char *path;
SDnode *pDnode;
......@@ -54,6 +54,7 @@ int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessCDnodeReq(SDnode *pDnode, SNodeMsg *pMsg);
// dmMonitor.c
void dmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo);
void dmSendMonitorReport(SDnode *pDnode);
// dmWorker.c
......@@ -61,7 +62,7 @@ int32_t dmStartThread(SDnodeMgmt *pMgmt);
int32_t dmStartWorker(SDnodeMgmt *pMgmt);
void dmStopWorker(SDnodeMgmt *pMgmt);
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t dmProcessStatusMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t dmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus
}
......
......@@ -169,6 +169,7 @@ void dndCleanupTrans(SDnode *pDnode);
SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper);
SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper);
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRecv(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
// mgmt
void dmSetMgmtFp(SMgmtWrapper *pWrapper);
......@@ -182,22 +183,13 @@ void dmGetMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmUpdateMnodeEpSet(SDnodeMgmt *pMgmt, SEpSet *pEpSet);
void dmSendRedirectRsp(SDnodeMgmt *pMgmt, const SRpcMsg *pMsg);
typedef struct {
int32_t openVnodes;
int32_t totalVnodes;
int32_t masterNum;
int64_t numOfSelectReqs;
int64_t numOfInsertReqs;
int64_t numOfInsertSuccessReqs;
int64_t numOfBatchInsertReqs;
int64_t numOfBatchInsertSuccessReqs;
} SVnodesStat;
void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads);
int32_t vmMonitorTfsInfo(SMgmtWrapper *pWrapper, SMonDiskInfo *pInfo);
void vmMonitorVnodeReqs(SMgmtWrapper *pWrapper, SMonDnodeInfo *pInfo);
int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
void dmGetMonitorSysInfo(SMonSysInfo *pInfo);
void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo);
void mmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonMmInfo *mmInfo);
void vmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonVmInfo *vmInfo);
void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo);
void smGetMonitorInfo(SMgmtWrapper *pWrapper, SMonSmInfo *smInfo);
void bmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonBmInfo *bmInfo);
#ifdef __cplusplus
}
......
......@@ -32,6 +32,7 @@ typedef struct SMnodeMgmt {
SSingleWorker readWorker;
SSingleWorker writeWorker;
SSingleWorker syncWorker;
SSingleWorker monitorWorker;
SReplica replicas[TSDB_MAX_REPLICA];
int8_t replica;
int8_t selfIndex;
......@@ -51,6 +52,7 @@ void mmInitMsgHandle(SMgmtWrapper *pWrapper);
int32_t mmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t mmProcessGetMonMmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
// mmWorker.c
int32_t mmStartWorker(SMnodeMgmt *pMgmt);
......@@ -59,6 +61,7 @@ int32_t mmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessSyncMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessReadMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t mmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc);
int32_t mmPutMsgToReadQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc);
int32_t mmPutMsgToWriteQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc);
......
......@@ -30,6 +30,7 @@ typedef struct SQnodeMgmt {
const char *path;
SSingleWorker queryWorker;
SSingleWorker fetchWorker;
SSingleWorker monitorWorker;
} SQnodeMgmt;
// qmInt.c
......@@ -40,6 +41,7 @@ int32_t qmDrop(SMgmtWrapper *pWrapper);
void qmInitMsgHandle(SMgmtWrapper *pWrapper);
int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t qmProcessGetMonQmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
// qmWorker.c
int32_t qmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
......@@ -50,6 +52,7 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt);
void qmStopWorker(SQnodeMgmt *pMgmt);
int32_t qmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t qmProcessFetchMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t qmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus
}
......
......@@ -32,6 +32,7 @@ typedef struct SSnodeMgmt {
int8_t uniqueWorkerInUse;
SArray *uniqueWorkers; // SArray<SMultiWorker*>
SSingleWorker sharedWorker;
SSingleWorker monitorWorker;
} SSnodeMgmt;
// smInt.c
......@@ -42,6 +43,7 @@ int32_t smDrop(SMgmtWrapper *pWrapper);
void smInitMsgHandle(SMgmtWrapper *pWrapper);
int32_t smProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessGetMonSmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
// smWorker.c
int32_t smStartWorker(SSnodeMgmt *pMgmt);
......@@ -50,6 +52,7 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessSharedMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessExecMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus
}
......
......@@ -28,6 +28,7 @@ typedef struct SVnodesMgmt {
SHashObj *hash;
SRWLatch latch;
SVnodesStat state;
SVnodesStat lastState;
STfs *pTfs;
SQWorkerPool queryPool;
SQWorkerPool fetchPool;
......@@ -38,6 +39,7 @@ typedef struct SVnodesMgmt {
SDnode *pDnode;
SMgmtWrapper *pWrapper;
SSingleWorker mgmtWorker;
SSingleWorker monitorWorker;
} SVnodesMgmt;
typedef struct {
......@@ -91,6 +93,8 @@ int32_t vmProcessAlterVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pReq);
int32_t vmProcessDropVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pReq);
int32_t vmProcessSyncVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pReq);
int32_t vmProcessCompactVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pReq);
int32_t vmProcessGetMonVmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
int32_t vmProcessGetVnodeLoadsReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq);
// vmFile.c
int32_t vmGetVnodesFromFile(SVnodesMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes);
......@@ -114,6 +118,7 @@ int32_t vmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t vmProcessFetchMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrappert, SNodeMsg *pMsg);
int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg);
#ifdef __cplusplus
}
......
......@@ -12,7 +12,7 @@
* 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 "dndInt.h"
......@@ -482,4 +482,8 @@ SProcCfg dndGenProcCfg(SMgmtWrapper *pWrapper) {
.parent = pWrapper,
.name = pWrapper->name};
return cfg;
}
void dndSendRecv(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp) {
rpcSendRecv(pDnode->trans.clientRpc, pEpSet, pReq, pRsp);
}
\ No newline at end of file
......@@ -16,6 +16,36 @@
#define _DEFAULT_SOURCE
#include "mmInt.h"
void mmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonMmInfo *mmInfo) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt;
mndGetMonitorInfo(pMgmt->pMnode, &mmInfo->cluster, &mmInfo->vgroup, &mmInfo->grant);
}
int32_t mmProcessGetMonMmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonMmInfo mmInfo = {0};
mmGetMonitorInfo(pWrapper, &mmInfo);
dmGetMonitorSysInfo(&mmInfo.sys);
monGetLogs(&mmInfo.log);
int32_t rspLen = tSerializeSMonMmInfo(NULL, 0, &mmInfo);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonMmInfo(pRsp, rspLen, &mmInfo);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonMmInfo(&mmInfo);
return 0;
}
int32_t mmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnode *pDnode = pWrapper->pDnode;
SRpcMsg *pReq = &pMsg->rpcMsg;
......@@ -74,6 +104,8 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
}
void mmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_MON_MM_INFO, mmProcessMonitorMsg, DEFAULT_HANDLE);
// Requests handled by DNODE
dndSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_DND_ALTER_MNODE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
......@@ -163,5 +195,4 @@ void mmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_VND_FETCH, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_DROP_TASK, mmProcessQueryMsg, MNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_HEARTBEAT, mmProcessQueryMsg, MNODE_HANDLE);
}
......@@ -241,8 +241,3 @@ void mmSetMgmtFp(SMgmtWrapper *pWrapper) {
pWrapper->fp = mgmtFp;
}
int32_t mmMonitorMnodeInfo(SMgmtWrapper *pWrapper, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt;
return mndGetMonitorInfo(pMgmt->pMnode, pClusterInfo, pVgroupInfo, pGrantInfo);
}
......@@ -23,19 +23,18 @@ static void mmProcessQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1;
if (pMsg->rpcMsg.msgType != TDMT_DND_ALTER_MNODE) {
if (pMsg->rpcMsg.msgType == TDMT_DND_ALTER_MNODE) {
code = mmProcessAlterReq(pMgmt, pMsg);
} else if (pMsg->rpcMsg.msgType == TDMT_MON_MM_INFO) {
code = mmProcessGetMonMmInfoReq(pMgmt->pWrapper, pMsg);
} else {
pMsg->pNode = pMgmt->pMnode;
code = mndProcessMsg(pMsg);
} else {
code = mmProcessAlterReq(pMgmt, pMsg);
}
if (pRpc->msgType & 1U) {
if (pRpc->handle != NULL && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
if (code != 0) {
code = terrno;
dError("msg:%p, failed to process since %s", pMsg, terrstr());
}
if (code != 0 && terrno != 0) code = terrno;
SRpcMsg rsp = {.handle = pRpc->handle, .code = code, .contLen = pMsg->rspLen, .pCont = pMsg->pRsp};
tmsgSendRsp(&rsp);
}
......@@ -98,6 +97,15 @@ int32_t mmProcessQueryMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t mmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SMnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
static int32_t mmPutRpcMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pRpc) {
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg == NULL) return -1;
......@@ -157,15 +165,24 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
SSingleWorkerCfg sCfg = {.min = 1, .max = 1, .name = "mnode-sync", .fp = (FItem)mmProcessQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->syncWorker, &sCfg) != 0) {
dError("failed to start mnode sync-worker since %s", terrstr());
dError("failed to start mnode mnode-sync worker since %s", terrstr());
return -1;
}
if (tsMultiProcess) {
SSingleWorkerCfg mCfg = {.min = 1, .max = 1, .name = "mnode-monitor", .fp = (FItem)mmProcessQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
dError("failed to start mnode mnode-monitor worker since %s", terrstr());
return -1;
}
}
dDebug("mnode workers are initialized");
return 0;
}
void mmStopWorker(SMnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->monitorWorker);
tSingleWorkerCleanup(&pMgmt->queryWorker);
tSingleWorkerCleanup(&pMgmt->readWorker);
tSingleWorkerCleanup(&pMgmt->writeWorker);
......
......@@ -16,6 +16,34 @@
#define _DEFAULT_SOURCE
#include "qmInt.h"
void qmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonQmInfo *qmInfo) {
}
int32_t qmProcessGetMonQmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonQmInfo qmInfo = {0};
qmGetMonitorInfo(pWrapper, &qmInfo);
dmGetMonitorSysInfo(&qmInfo.sys);
monGetLogs(&qmInfo.log);
int32_t rspLen = tSerializeSMonQmInfo(NULL, 0, &qmInfo);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonQmInfo(pRsp, rspLen, &qmInfo);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonQmInfo(&qmInfo);
return 0;
}
int32_t qmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnode *pDnode = pWrapper->pDnode;
SRpcMsg *pReq = &pMsg->rpcMsg;
......@@ -55,6 +83,8 @@ int32_t qmProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
void qmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_MON_QM_INFO, qmProcessMonitorMsg, DEFAULT_HANDLE);
// Requests handled by VNODE
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, qmProcessQueryMsg, QNODE_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, qmProcessQueryMsg, QNODE_HANDLE);
......
......@@ -16,11 +16,36 @@
#define _DEFAULT_SOURCE
#include "qmInt.h"
static void qmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
static inline void qmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle,
.code = code,
.pCont = pMsg->pRsp,
.contLen = pMsg->rspLen};
tmsgSendRsp(&rsp);
}
static void qmProcessMonQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SQnodeMgmt *pMgmt = pInfo->ahandle;
dTrace("msg:%p, get from qnode monitor queue", pMsg);
SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1;
if (pMsg->rpcMsg.msgType == TDMT_MON_SM_INFO) {
code = qmProcessGetMonQmInfoReq(pMgmt->pWrapper, pMsg);
}
if (pRpc->msgType & 1U) {
if (code != 0 && terrno != 0) code = terrno;
qmSendRsp(pMgmt->pWrapper, pMsg, code);
}
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
rpcFreeCont(pRpc->pCont);
taosFreeQitem(pMsg);
}
static void qmProcessQueryQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SQnodeMgmt *pMgmt = pInfo->ahandle;
......@@ -66,6 +91,15 @@ int32_t qmProcessFetchMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t qmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SQnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) {
SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
if (pMsg == NULL) {
......@@ -128,11 +162,21 @@ int32_t qmStartWorker(SQnodeMgmt *pMgmt) {
return -1;
}
if (tsMultiProcess) {
SSingleWorkerCfg mCfg = {
.min = 1, .max = 1, .name = "qnode-monitor", .fp = (FItem)qmProcessMonQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
dError("failed to start qnode-monitor worker since %s", terrstr());
return -1;
}
}
dDebug("qnode workers are initialized");
return 0;
}
void qmStopWorker(SQnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->monitorWorker);
tSingleWorkerCleanup(&pMgmt->queryWorker);
tSingleWorkerCleanup(&pMgmt->fetchWorker);
dDebug("qnode workers are closed");
......
......@@ -16,6 +16,33 @@
#define _DEFAULT_SOURCE
#include "smInt.h"
void smGetMonitorInfo(SMgmtWrapper *pWrapper, SMonSmInfo *smInfo) {}
int32_t smProcessGetMonSmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonSmInfo smInfo = {0};
smGetMonitorInfo(pWrapper, &smInfo);
dmGetMonitorSysInfo(&smInfo.sys);
monGetLogs(&smInfo.log);
int32_t rspLen = tSerializeSMonSmInfo(NULL, 0, &smInfo);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonSmInfo(pRsp, rspLen, &smInfo);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonSmInfo(&smInfo);
return 0;
}
int32_t smProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SDnode *pDnode = pWrapper->pDnode;
SRpcMsg *pReq = &pMsg->rpcMsg;
......@@ -55,6 +82,8 @@ int32_t smProcessDropReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
}
void smInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_MON_SM_INFO, smProcessMonitorMsg, DEFAULT_HANDLE);
// Requests handled by SNODE
dndSetMsgHandle(pWrapper, TDMT_SND_TASK_DEPLOY, smProcessMgmtMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_SND_TASK_EXEC, smProcessExecMsg, DEFAULT_HANDLE);
......
......@@ -16,6 +16,36 @@
#define _DEFAULT_SOURCE
#include "smInt.h"
static inline void smSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle,
.code = code,
.pCont = pMsg->pRsp,
.contLen = pMsg->rspLen};
tmsgSendRsp(&rsp);
}
static void smProcessMonitorQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pInfo->ahandle;
dTrace("msg:%p, get from snode monitor queue", pMsg);
SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1;
if (pMsg->rpcMsg.msgType == TDMT_MON_SM_INFO) {
code = smProcessGetMonSmInfoReq(pMgmt->pWrapper, pMsg);
}
if (pRpc->msgType & 1U) {
if (code != 0 && terrno != 0) code = terrno;
smSendRsp(pMgmt->pWrapper, pMsg, code);
}
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
rpcFreeCont(pRpc->pCont);
taosFreeQitem(pMsg);
}
static void smProcessUniqueQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SSnodeMgmt *pMgmt = pInfo->ahandle;
......@@ -80,11 +110,21 @@ int32_t smStartWorker(SSnodeMgmt *pMgmt) {
return -1;
}
if (tsMultiProcess) {
SSingleWorkerCfg mCfg = {
.min = 1, .max = 1, .name = "snode-monitor", .fp = (FItem)smProcessMonitorQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
dError("failed to start snode-monitor worker since %s", terrstr());
return -1;
}
}
dDebug("snode workers are initialized");
return 0;
}
void smStopWorker(SSnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->monitorWorker);
for (int32_t i = 0; i < taosArrayGetSize(pMgmt->uniqueWorkers); i++) {
SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, i);
tMultiWorkerCleanup(pWorker);
......@@ -120,6 +160,15 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt;
int32_t index = smGetSWIdFromMsg(&pMsg->rpcMsg);
......
......@@ -16,6 +16,71 @@
#define _DEFAULT_SOURCE
#include "vmInt.h"
void vmGetMonitorInfo(SMgmtWrapper *pWrapper, SMonVmInfo *vmInfo) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
tfsGetMonitorInfo(pMgmt->pTfs, &vmInfo->tfs);
taosWLockLatch(&pMgmt->latch);
vmInfo->vstat.totalVnodes = pMgmt->state.totalVnodes;
vmInfo->vstat.masterNum = pMgmt->state.masterNum;
vmInfo->vstat.numOfSelectReqs = pMgmt->state.numOfSelectReqs - pMgmt->lastState.numOfSelectReqs;
vmInfo->vstat.numOfInsertReqs = pMgmt->state.numOfInsertReqs - pMgmt->lastState.numOfInsertReqs;
vmInfo->vstat.numOfInsertSuccessReqs = pMgmt->state.numOfInsertSuccessReqs - pMgmt->lastState.numOfInsertSuccessReqs;
vmInfo->vstat.numOfBatchInsertReqs = pMgmt->state.numOfBatchInsertReqs - pMgmt->lastState.numOfBatchInsertReqs;
vmInfo->vstat.numOfBatchInsertSuccessReqs =
pMgmt->state.numOfBatchInsertSuccessReqs - pMgmt->lastState.numOfBatchInsertSuccessReqs;
pMgmt->lastState = pMgmt->state;
taosWUnLockLatch(&pMgmt->latch);
}
int32_t vmProcessGetMonVmInfoReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonVmInfo vmInfo = {0};
vmGetMonitorInfo(pWrapper, &vmInfo);
dmGetMonitorSysInfo(&vmInfo.sys);
monGetLogs(&vmInfo.log);
int32_t rspLen = tSerializeSMonVmInfo(NULL, 0, &vmInfo);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonVmInfo(pRsp, rspLen, &vmInfo);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonVmInfo(&vmInfo);
return 0;
}
int32_t vmProcessGetVnodeLoadsReq(SMgmtWrapper *pWrapper, SNodeMsg *pReq) {
SMonVloadInfo vloads = {0};
vmGetVnodeLoads(pWrapper, &vloads);
int32_t rspLen = tSerializeSMonVloadInfo(NULL, 0, &vloads);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSMonVloadInfo(pRsp, rspLen, &vloads);
pReq->pRsp = pRsp;
pReq->rspLen = rspLen;
tFreeSMonVloadInfo(&vloads);
return 0;
}
static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->vgId = pCreate->vgId;
pCfg->wsize = pCreate->cacheBlockSize;
......@@ -239,6 +304,9 @@ int32_t vmProcessCompactVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
}
void vmInitMsgHandle(SMgmtWrapper *pWrapper) {
dndSetMsgHandle(pWrapper, TDMT_MON_VM_INFO, vmProcessMonitorMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_MON_VM_LOAD, vmProcessMonitorMsg, DEFAULT_HANDLE);
// Requests handled by VNODE
dndSetMsgHandle(pWrapper, TDMT_VND_SUBMIT, (NodeMsgFp)vmProcessWriteMsg, DEFAULT_HANDLE);
dndSetMsgHandle(pWrapper, TDMT_VND_QUERY, (NodeMsgFp)vmProcessQueryMsg, DEFAULT_HANDLE);
......
......@@ -344,38 +344,21 @@ void vmSetMgmtFp(SMgmtWrapper *pWrapper) {
pWrapper->fp = mgmtFp;
}
int32_t vmMonitorTfsInfo(SMgmtWrapper *pWrapper, SMonDiskInfo *pInfo) {
void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
if (pMgmt == NULL) return -1;
return tfsGetMonitorInfo(pMgmt->pTfs, pInfo);
}
void vmMonitorVnodeReqs(SMgmtWrapper *pWrapper, SMonDnodeInfo *pInfo) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
if (pMgmt == NULL) return;
SVnodesStat *pStat = &pMgmt->state;
pInfo->req_select = pStat->numOfSelectReqs;
pInfo->req_insert = pStat->numOfInsertReqs;
pInfo->req_insert_success = pStat->numOfInsertSuccessReqs;
pInfo->req_insert_batch = pStat->numOfBatchInsertReqs;
pInfo->req_insert_batch_success = pStat->numOfBatchInsertSuccessReqs;
pInfo->errors = tsNumOfErrorLogs;
pInfo->vnodes_num = pStat->totalVnodes;
pInfo->masters = pStat->masterNum;
}
SArray *pLoads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SVnodesStat *pStat = &pMgmt->state;
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;
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;
pInfo->pVloads = pLoads;
if (pLoads == NULL) return;
taosRLockLatch(&pMgmt->latch);
......@@ -402,6 +385,7 @@ void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads) {
taosRUnLockLatch(&pMgmt->latch);
taosWLockLatch(&pMgmt->latch);
pStat->totalVnodes = totalVnodes;
pStat->masterNum = masterNum;
pStat->numOfSelectReqs = numOfSelectReqs;
......@@ -409,4 +393,5 @@ void vmMonitorVnodeLoads(SMgmtWrapper *pWrapper, SArray *pLoads) {
pStat->numOfInsertSuccessReqs = numOfInsertSuccessReqs;
pStat->numOfBatchInsertReqs = numOfBatchInsertReqs;
pStat->numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;
taosWUnLockLatch(&pMgmt->latch);
}
\ No newline at end of file
......@@ -16,8 +16,12 @@
#define _DEFAULT_SOURCE
#include "vmInt.h"
static void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
static inline void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle,
.code = code,
.pCont = pMsg->pRsp,
.contLen = pMsg->rspLen};
tmsgSendRsp(&rsp);
}
......@@ -26,9 +30,15 @@ static void vmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
int32_t code = -1;
tmsg_t msgType = pMsg->rpcMsg.msgType;
dTrace("msg:%p, will be processed in vnode-mgmt queue", pMsg);
dTrace("msg:%p, will be processed in vnode-m queue", pMsg);
switch (msgType) {
case TDMT_MON_VM_INFO:
code = vmProcessGetMonVmInfoReq(pMgmt->pWrapper, pMsg);
break;
case TDMT_MON_VM_LOAD:
code = vmProcessGetVnodeLoadsReq(pMgmt->pWrapper, pMsg);
break;
case TDMT_DND_CREATE_VNODE:
code = vmProcessCreateVnodeReq(pMgmt, pMsg);
break;
......@@ -255,6 +265,15 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
return 0;
}
int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;
}
static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SMsgHead *pHead = pRpc->pCont;
......@@ -412,11 +431,21 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
return -1;
}
if (tsMultiProcess) {
SSingleWorkerCfg mCfg = {
.min = 1, .max = 1, .name = "vnode-monitor", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
dError("failed to start mnode vnode-monitor worker since %s", terrstr());
return -1;
}
}
dDebug("vnode workers are initialized");
return 0;
}
void vmStopWorker(SVnodesMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->monitorWorker);
tSingleWorkerCleanup(&pMgmt->mgmtWorker);
tQWorkerCleanup(&pMgmt->fetchPool);
tQWorkerCleanup(&pMgmt->queryPool);
......
......@@ -237,7 +237,7 @@ int32_t mndGetDnodeSize(SMnode *pMnode) {
bool mndIsDnodeOnline(SMnode *pMnode, SDnodeObj *pDnode, int64_t curMs) {
int64_t interval = TABS(pDnode->lastAccessTime - curMs);
if (interval > 3500 * tsStatusInterval) {
if (interval > 30000 * tsStatusInterval) {
if (pDnode->rebootTime > 0) {
pDnode->offlineReason = DND_REASON_STATUS_MSG_TIMEOUT;
}
......
......@@ -502,7 +502,11 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
SMonVgroupDesc desc = {0};
desc.vgroup_id = pVgroup->vgId;
strncpy(desc.database_name, pVgroup->dbName, sizeof(desc.database_name));
SName name = {0};
tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
tNameGetDbName(&name, desc.database_name);
desc.tables_num = pVgroup->numOfTables;
pGrantInfo->timeseries_used += pVgroup->numOfTimeSeries;
tstrncpy(desc.status, "unsynced", sizeof(desc.status));
......
......@@ -18,43 +18,32 @@
#include "monitor.h"
#include "tarray.h"
#include "tjson.h"
typedef struct {
int64_t ts;
ELogLevel level;
char content[MON_LOG_LEN];
} SMonLogItem;
typedef struct {
int64_t time;
int64_t req_select;
int64_t req_insert;
int64_t req_insert_batch;
int64_t net_in;
int64_t net_out;
int64_t io_read;
int64_t io_write;
int64_t io_read_disk;
int64_t io_write_disk;
} SMonState;
typedef struct SMonInfo {
int64_t curTime;
SMonState lastState;
SArray *logs; // array of SMonLogItem
SJson *pJson;
int64_t curTime;
int64_t lastTime;
SJson *pJson;
SMonLogs log;
SMonDmInfo dmInfo;
SMonMmInfo mmInfo;
SMonVmInfo vmInfo;
SMonSmInfo smInfo;
SMonQmInfo qmInfo;
SMonBmInfo bmInfo;
} SMonInfo;
typedef struct {
TdThreadMutex lock;
SArray *logs; // array of SMonLogItem
int32_t maxLogs;
const char *server;
uint16_t port;
bool comp;
SMonState state;
SArray *logs; // array of SMonLogItem
SMonCfg cfg;
int64_t lastTime;
SMonDmInfo dmInfo;
SMonMmInfo mmInfo;
SMonVmInfo vmInfo;
SMonSmInfo smInfo;
SMonQmInfo qmInfo;
SMonBmInfo bmInfo;
} SMonitor;
#ifdef __cplusplus
......
此差异已折叠。
enable_testing()
aux_source_directory(. MONITOR_TEST_SRC)
add_executable(monitor_test ${MONITOR_TEST_SRC})
add_executable(monitorTest ${MONITOR_TEST_SRC})
target_link_libraries(
monitor_test
monitorTest
PUBLIC monitor
PUBLIC gtest_main
)
add_test(
NAME monitor_test
COMMAND monitor_test
NAME monitorTest
COMMAND monitorTest
)
......@@ -22,7 +22,7 @@ class MonitorTest : public ::testing::Test {
cfg.maxLogs = 2;
cfg.port = 80;
cfg.server = "localhost";
cfg.comp = 0;
cfg.comp = 1;
monInit(&cfg);
}
......@@ -32,24 +32,64 @@ class MonitorTest : public ::testing::Test {
void SetUp() override {}
void TearDown() override {}
void GetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo);
void GetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo);
void GetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo);
void GetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo);
void GetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo);
void GetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo);
void GetBasicInfo(SMonBasicInfo *pInfo);
void GetDnodeInfo(SMonDnodeInfo *pInfo);
void GetSysInfo(SMonSysInfo *pInfo);
void GetClusterInfo(SMonClusterInfo *pInfo);
void GetVgroupInfo(SMonVgroupInfo *pInfo);
void GetGrantInfo(SMonGrantInfo *pInfo);
void GetVnodeStat(SVnodesStat *pStat);
void GetDiskInfo(SMonDiskInfo *pInfo);
void GetLogInfo(SMonLogs *logs);
void AddLogInfo1();
void AddLogInfo2();
};
void MonitorTest::GetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo) {
void MonitorTest::GetBasicInfo(SMonBasicInfo *pInfo) {
pInfo->dnode_id = 1;
strcpy(pInfo->dnode_ep, "localhost");
pInfo->cluster_id = 6980428120398645172;
pInfo->protocol = 1;
}
void MonitorTest::GetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo) {
void MonitorTest::GetDnodeInfo(SMonDnodeInfo *pInfo) {
pInfo->uptime = 1.2;
pInfo->has_mnode = 1;
strcpy(pInfo->logdir.name, "/log/dir/d");
pInfo->logdir.size.avail = 41;
pInfo->logdir.size.total = 42;
pInfo->logdir.size.used = 43;
strcpy(pInfo->tempdir.name, "/data/dir/d");
pInfo->tempdir.size.avail = 51;
pInfo->tempdir.size.total = 52;
pInfo->tempdir.size.used = 53;
}
void MonitorTest::GetSysInfo(SMonSysInfo *pInfo) {
pInfo->cpu_engine = 2.1;
pInfo->cpu_system = 2.1;
pInfo->cpu_cores = 2;
pInfo->mem_engine = 3.1;
pInfo->mem_system = 3.2;
pInfo->mem_total = 3.3;
pInfo->disk_engine = 4.1;
pInfo->disk_used = 4.2;
pInfo->disk_total = 4.3;
pInfo->net_in = 5.1;
pInfo->net_out = 5.2;
pInfo->io_read = 6.1;
pInfo->io_write = 6.2;
pInfo->io_read_disk = 7.1;
pInfo->io_write_disk = 7.2;
}
void MonitorTest::GetClusterInfo(SMonClusterInfo *pInfo) {
strcpy(pInfo->first_ep, "localhost:6030");
pInfo->first_ep_dnode_id = 1;
strcpy(pInfo->version, "3.0.0.0");
......@@ -86,7 +126,7 @@ void MonitorTest::GetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo) {
taosArrayPush(pInfo->mnodes, &m2);
}
void MonitorTest::GetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo) {
void MonitorTest::GetVgroupInfo(SMonVgroupInfo *pInfo) {
pInfo->vgroups = taosArrayInit(4, sizeof(SMonVgroupDesc));
SMonVgroupDesc vg1 = {0};
......@@ -121,41 +161,24 @@ void MonitorTest::GetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo) {
taosArrayPush(pInfo->vgroups, &vg3);
}
void MonitorTest::GetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo) {
void MonitorTest::GetGrantInfo(SMonGrantInfo *pInfo) {
pInfo->expire_time = 1234567;
pInfo->timeseries_total = 234567;
pInfo->timeseries_used = 34567;
}
void MonitorTest::GetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo) {
pInfo->uptime = 1.2;
pInfo->cpu_engine = 2.1;
pInfo->cpu_system = 2.1;
pInfo->cpu_cores = 2;
pInfo->mem_engine = 3.1;
pInfo->mem_system = 3.2;
pInfo->mem_total = 3.3;
pInfo->disk_engine = 4.1;
pInfo->disk_used = 4.2;
pInfo->disk_total = 4.3;
pInfo->net_in = 5.1;
pInfo->net_out = 5.2;
pInfo->io_read = 6.1;
pInfo->io_write = 6.2;
pInfo->io_read_disk = 7.1;
pInfo->io_write_disk = 7.2;
pInfo->req_select = 8;
pInfo->req_insert = 9;
pInfo->req_insert_success = 10;
pInfo->req_insert_batch = 11;
pInfo->req_insert_batch_success = 12;
void MonitorTest::GetVnodeStat(SVnodesStat *pInfo) {
pInfo->numOfSelectReqs = 8;
pInfo->numOfInsertReqs = 9;
pInfo->numOfInsertSuccessReqs = 10;
pInfo->numOfBatchInsertReqs = 11;
pInfo->numOfBatchInsertSuccessReqs = 12;
pInfo->errors = 4;
pInfo->vnodes_num = 5;
pInfo->masters = 6;
pInfo->has_mnode = 1;
pInfo->totalVnodes = 5;
pInfo->masterNum = 6;
}
void MonitorTest::GetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo) {
void MonitorTest::GetDiskInfo(SMonDiskInfo *pInfo) {
pInfo->datadirs = taosArrayInit(2, sizeof(SMonDiskDesc));
SMonDiskDesc d1 = {0};
strcpy(d1.name, "/t1/d1/d");
......@@ -180,16 +203,25 @@ void MonitorTest::GetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo) {
d3.size.total = 32;
d3.size.used = 33;
taosArrayPush(pInfo->datadirs, &d3);
}
strcpy(pInfo->logdir.name, "/log/dir/d");
pInfo->logdir.size.avail = 41;
pInfo->logdir.size.total = 42;
pInfo->logdir.size.used = 43;
void MonitorTest::GetLogInfo(SMonLogs *logs) {
logs->logs = taosArrayInit(4, sizeof(SMonLogItem));
strcpy(pInfo->tempdir.name, "/data/dir/d");
pInfo->tempdir.size.avail = 51;
pInfo->tempdir.size.total = 52;
pInfo->tempdir.size.used = 53;
SMonLogItem item1 = {.level = DEBUG_INFO};
item1.ts = taosGetTimestampMs();
strcpy(item1.content, "log test1");
taosArrayPush(logs->logs, &item1);
SMonLogItem item2 = {.level = DEBUG_ERROR};
item2.ts = taosGetTimestampMs();
strcpy(item2.content, "log test2");
taosArrayPush(logs->logs, &item2);
logs->numOfErrorLogs = 1;
logs->numOfInfoLogs = 2;
logs->numOfDebugLogs = 3;
logs->numOfTraceLogs = 4;
}
void MonitorTest::AddLogInfo1() {
......@@ -206,46 +238,52 @@ void MonitorTest::AddLogInfo2() {
TEST_F(MonitorTest, 01_Full) {
AddLogInfo1();
SMonInfo *pMonitor = monCreateMonitorInfo();
if (pMonitor == NULL) return;
SMonBasicInfo basicInfo = {0};
GetBasicInfo(pMonitor, &basicInfo);
monSetBasicInfo(pMonitor, &basicInfo);
SMonClusterInfo clusterInfo = {0};
SMonVgroupInfo vgroupInfo = {0};
SMonGrantInfo grantInfo = {0};
GetClusterInfo(pMonitor, &clusterInfo);
GetVgroupInfo(pMonitor, &vgroupInfo);
GetGrantInfo(pMonitor, &grantInfo);
monSetClusterInfo(pMonitor, &clusterInfo);
monSetVgroupInfo(pMonitor, &vgroupInfo);
monSetGrantInfo(pMonitor, &grantInfo);
SMonDnodeInfo dnodeInfo = {0};
GetDnodeInfo(pMonitor, &dnodeInfo);
monSetDnodeInfo(pMonitor, &dnodeInfo);
SMonDiskInfo diskInfo = {0};
GetDiskInfo(pMonitor, &diskInfo);
monSetDiskInfo(pMonitor, &diskInfo);
monSendReport(pMonitor);
monCleanupMonitorInfo(pMonitor);
taosArrayDestroy(clusterInfo.dnodes);
taosArrayDestroy(clusterInfo.mnodes);
taosArrayDestroy(vgroupInfo.vgroups);
taosArrayDestroy(diskInfo.datadirs);
SMonDmInfo dmInfo = {0};
GetBasicInfo(&dmInfo.basic);
GetDnodeInfo(&dmInfo.dnode);
GetSysInfo(&dmInfo.sys);
SMonMmInfo mmInfo = {0};
GetClusterInfo(&mmInfo.cluster);
GetVgroupInfo(&mmInfo.vgroup);
GetGrantInfo(&mmInfo.grant);
GetSysInfo(&mmInfo.sys);
GetLogInfo(&mmInfo.log);
SMonVmInfo vmInfo = {0};
GetDiskInfo(&vmInfo.tfs);
GetVnodeStat(&vmInfo.vstat);
GetSysInfo(&vmInfo.sys);
GetLogInfo(&vmInfo.log);
SMonQmInfo qmInfo = {0};
GetSysInfo(&qmInfo.sys);
GetLogInfo(&qmInfo.log);
SMonSmInfo smInfo = {0};
GetSysInfo(&smInfo.sys);
GetLogInfo(&smInfo.log);
SMonBmInfo bmInfo = {0};
GetSysInfo(&bmInfo.sys);
GetLogInfo(&bmInfo.log);
monSetDmInfo(&dmInfo);
monSetMmInfo(&mmInfo);
monSetVmInfo(&vmInfo);
monSetQmInfo(&qmInfo);
monSetSmInfo(&smInfo);
monSetBmInfo(&bmInfo);
tFreeSMonMmInfo(&mmInfo);
tFreeSMonVmInfo(&vmInfo);
tFreeSMonSmInfo(&smInfo);
tFreeSMonQmInfo(&qmInfo);
tFreeSMonBmInfo(&bmInfo);
monSendReport();
}
TEST_F(MonitorTest, 02_Log) {
AddLogInfo2();
SMonInfo *pMonitor = monCreateMonitorInfo();
if (pMonitor == NULL) return;
monSendReport(pMonitor);
monCleanupMonitorInfo(pMonitor);
monSendReport();
}
......@@ -117,7 +117,7 @@ _OVER:
static void clientConnCb(uv_connect_t* req, int32_t status) {
if (status < 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("Connection error %s\n", uv_strerror(status));
uError("connection error %s", uv_strerror(status));
uv_close((uv_handle_t*)req->handle, NULL);
return;
}
......
......@@ -129,14 +129,12 @@ static void taosGetProcIOnfos() {
static int32_t taosGetSysCpuInfo(SysCpuInfo *cpuInfo) {
TdFilePtr pFile = taosOpenFile(tsSysCpuFile, TD_FILE_READ | TD_FILE_STREAM);
if (pFile == NULL) {
// printf("open file:%s failed", tsSysCpuFile);
return -1;
}
char *line = NULL;
ssize_t _bytes = taosGetLineFile(pFile, &line);
if ((_bytes < 0) || (line == NULL)) {
// printf("read file:%s failed", tsSysCpuFile);
taosCloseFile(&pFile);
return -1;
}
......@@ -153,14 +151,12 @@ static int32_t taosGetSysCpuInfo(SysCpuInfo *cpuInfo) {
static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
TdFilePtr pFile = taosOpenFile(tsProcCpuFile, TD_FILE_READ | TD_FILE_STREAM);
if (pFile == NULL) {
// printf("open file:%s failed", tsProcCpuFile);
return -1;
}
char *line = NULL;
ssize_t _bytes = taosGetLineFile(pFile, &line);
if ((_bytes < 0) || (line == NULL)) {
// printf("read file:%s failed", tsProcCpuFile);
taosCloseFile(&pFile);
return -1;
}
......@@ -182,12 +178,12 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
#endif
bool taosCheckSystemIsSmallEnd() {
union check{
int16_t i;
char ch[2];
}c;
c.i=1;
return c.ch[0]==1;
union check {
int16_t i;
char ch[2];
} c;
c.i = 1;
return c.ch[0] == 1;
}
void taosGetSystemInfo() {
......@@ -617,6 +613,28 @@ int32_t taosGetProcIO(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int
#endif
}
void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes) {
static int64_t last_rchars = 0;
static int64_t last_wchars = 0;
static int64_t last_read_bytes = 0;
static int64_t last_write_bytes = 0;
static int64_t cur_rchars = 0;
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) {
*rchars = cur_rchars - last_rchars;
*wchars = cur_wchars - last_wchars;
*read_bytes = cur_read_bytes - last_read_bytes;
*write_bytes = cur_write_bytes - last_write_bytes;
last_rchars = cur_rchars;
last_wchars = cur_wchars;
last_read_bytes = cur_read_bytes;
last_write_bytes = cur_write_bytes;
}
}
int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes) {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
*receive_bytes = 0;
......@@ -672,6 +690,20 @@ int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes) {
#endif
}
void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
static int64_t last_receive_bytes = 0;
static int64_t last_transmit_bytes = 0;
static int64_t cur_receive_bytes = 0;
static int64_t 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;
last_transmit_bytes = cur_transmit_bytes;
}
}
void taosKillSystem() {
#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
printf("function taosKillSystem, exit!");
......
......@@ -348,7 +348,7 @@ SConfigItem *cfgGetItem(SConfig *pCfg, const char *name) {
}
}
uError("name:%s, cfg not found", name);
// uError("name:%s, cfg not found", name);
terrno = TSDB_CODE_CFG_NOT_FOUND;
return NULL;
}
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册