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

Merge pull request #10507 from taosdata/feature/config 

config
This commit is contained in:
Shengliang Guan 2022-03-03 10:21:41 +08:00 committed by GitHub
parent 2a18512e92 5dc2cfffde
commit 11064f1837
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
49 changed files with 598 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
contrib/test/craft/CMakeLists.txt
View File

@ -1,2 +1,2 @@
add_executable(simulate_vnode "simulate_vnode.c")
target_link_libraries(simulate_vnode PUBLIC craft lz4 uv_a)
target_link_libraries(simulate_vnode craft lz4 uv_a)

2
contrib/test/traft/single_node/CMakeLists.txt
View File

@ -3,4 +3,4 @@ target_sources(singleNode
PRIVATE
"singleNode.c"
)
target_link_libraries(singleNode PUBLIC traft lz4 uv_a)
target_link_libraries(singleNode traft lz4 uv_a)

6
example/CMakeLists.txt
View File

@ -2,11 +2,7 @@ aux_source_directory(src TMQ_DEMO_SRC)
add_executable(tmq ${TMQ_DEMO_SRC})
target_link_libraries(
tmq
PUBLIC taos
#PUBLIC util
#PUBLIC common
#PUBLIC os
tmq taos
)
target_include_directories(
tmq

7
include/common/tglobal.h
View File

@ -52,6 +52,13 @@ extern bool tsEnableSlaveQuery;
extern bool tsPrintAuth;
extern int64_t tsTickPerDay[3];
// monitor
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
extern int64_t tsQueryBufferSizeBytes; // maximum allowed usage buffer size in byte for each data node

2
include/common/ttime.h
View File

@ -52,6 +52,8 @@ void deltaToUtcInitOnce();
int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision);
void taosFormatUtcTime(char *buf, int32_t bufLen, int64_t time, int32_t precision);
#ifdef __cplusplus
}
#endif

25
include/dnode/mnode/mnode.h
View File

@ -16,6 +16,8 @@
#ifndef _TD_MND_H_
#define _TD_MND_H_
#include "monitor.h"
#ifdef __cplusplus
extern "C" {
#endif
@ -30,20 +32,6 @@ typedef int32_t (*PutReqToMWriteQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef int32_t (*PutReqToMReadQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef struct SMnodeLoad {
int64_t numOfDnode;
int64_t numOfMnode;
int64_t numOfVgroup;
int64_t numOfDatabase;
int64_t numOfSuperTable;
int64_t numOfChildTable;
int64_t numOfNormalTable;
int64_t numOfColumn;
int64_t totalPoints;
int64_t totalStorage;
int64_t compStorage;
} SMnodeLoad;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
@ -92,13 +80,16 @@ int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption);
void mndDestroy(const char *path);
/**
* @brief Get mnode statistics info.
* @brief Get mnode monitor info.
*
* @param pMnode The mnode object.
* @param pLoad Statistics of the mnode.
* @param pClusterInfo
* @param pVgroupInfo
* @param pGrantInfo
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
/**
* @brief Get user authentication info.

155
include/libs/monitor/monitor.h Normal file
View File

@ -0,0 +1,155 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* 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/>.
*/
#ifndef _TD_MONITOR_H_
#define _TD_MONITOR_H_
#include "tarray.h"
#include "tdef.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
} SMonBasicInfo;
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
char status[8];
} SMonDnodeDesc;
typedef struct {
int32_t mnode_id;
char mnode_ep[TSDB_EP_LEN];
char role[8];
} SMonMnodeDesc;
typedef struct {
char first_ep[TSDB_EP_LEN];
int32_t first_ep_dnode_id;
char version[12];
float master_uptime; // day
int32_t monitor_interval; // sec
int32_t vgroups_total;
int32_t vgroups_alive;
int32_t vnodes_total;
int32_t vnodes_alive;
int32_t connections_total;
SArray *dnodes; // array of SMonDnodeDesc
SArray *mnodes; // array of SMonMnodeDesc
} SMonClusterInfo;
typedef struct {
int32_t dnode_id;
int8_t vnode_online;
char vnode_role[8];
} SMonVnodeDesc;
typedef struct {
int32_t vgroup_id;
SMonVnodeDesc vnodes[TSDB_MAX_REPLICA];
} SMonVgroupDesc;
typedef struct {
char database_name[TSDB_DB_NAME_LEN];
int32_t tables_num;
int8_t status;
SArray *vgroups; // array of SMonVgroupDesc
} SMonVgroupInfo;
typedef struct {
int32_t expire_time;
int32_t timeseries_used;
int32_t timeseries_total;
} SMonGrantInfo;
typedef struct {
float uptime; // day
float cpu_engine;
float cpu_system;
float cpu_cores;
float mem_engine; // MB
float mem_system; // MB
float mem_total; // MB
float disk_engine; // GB
float disk_used; // GB
float disk_total; // GB
float net_in; // Kb/s
float net_out; // Kb/s
float io_read; // Mb/s
float io_write; // Mb/s
float io_read_disk; // Mb/s
float io_write_disk; // Mb/s
int32_t req_select;
float req_select_rate;
int32_t req_insert;
int32_t req_insert_success;
float req_insert_rate;
int32_t req_insert_batch;
int32_t req_insert_batch_success;
float req_insert_batch_rate;
int32_t errors;
int32_t vnodes_num;
int32_t masters;
int32_t has_mnode;
} SMonDnodeInfo;
typedef struct {
char name[TSDB_FILENAME_LEN];
int32_t level;
SDiskSize size;
} SMonDiskDesc;
typedef struct {
SArray *disks; // array of SMonDiskDesc
} SMonDiskInfo;
typedef struct {
int64_t ts;
int8_t level;
char content[1024];
} SMonLogItem;
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 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);
#ifdef __cplusplus
}
#endif
#endif /*_TD_MONITOR_H_*/

1
include/os/osSysinfo.h
View File

@ -42,7 +42,6 @@ bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage);
bool taosGetTotalSysMemoryKB(uint64_t *kb);
bool taosGetProcMemory(float *memoryUsedMB); //
bool taosGetSysMemory(float *memoryUsedMB); //
void taosGetDisk();
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize);
bool taosReadProcIO(int64_t *rchars, int64_t *wchars);
bool taosGetProcIO(float *readKB, float *writeKB);

2
source/common/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src COMMON_SRC)
add_library(common ${COMMON_SRC})
add_library(common STATIC ${COMMON_SRC})
target_include_directories(
common
PUBLIC "${CMAKE_SOURCE_DIR}/include/common"

22
source/common/src/tglobal.c
View File

@ -46,6 +46,13 @@ int32_t tsMaxBinaryDisplayWidth = 30;
bool tsEnableSlaveQuery = 1;
bool tsPrintAuth = 0;
// monitor
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,
* metricmeta rsp, and multi-meter query rsp message body. The client compress the submit message to server.
@ -314,6 +321,13 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "printAuth", tsPrintAuth, 0) != 0) return -1;
if (cfgAddBool(pCfg, "slaveQuery", tsEnableSlaveQuery, 0) != 0) return -1;
if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, 0) != 0) return -1;
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;
}
@ -345,7 +359,7 @@ static void taosSetServerLogCfg(SConfig *pCfg) {
}
static void taosSetClientCfg(SConfig *pCfg) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_EP_LEN);
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
@ -425,6 +439,12 @@ static void taosSetServerCfg(SConfig *pCfg) {
tsEnableSlaveQuery = cfgGetItem(pCfg, "slaveQuery")->bval;
tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->bval;
tsEnableMonitor = cfgGetItem(pCfg, "monitor")->bval;
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;
}

3
source/common/src/tname.c
View File

@ -22,6 +22,7 @@ bool tscValidateTableNameLength(size_t len) {
return len < TSDB_TABLE_NAME_LEN;
}
#if 0
// TODO refactor
SColumnFilterInfo* tFilterInfoDup(const SColumnFilterInfo* src, int32_t numOfFilters) {
if (numOfFilters == 0 || src == NULL) {
@ -46,7 +47,7 @@ SColumnFilterInfo* tFilterInfoDup(const SColumnFilterInfo* src, int32_t numOfFil
return pFilter;
}
#endif
#if 0
int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, int64_t intervalTime, char timeUnit, int16_t precision) {
if (slidingTime == 0) {

47
source/common/src/ttime.c
View File

@ -627,3 +627,50 @@ const char* fmtts(int64_t ts) {
return buf;
}
void taosFormatUtcTime(char* buf, int32_t bufLen, int64_t t, int32_t precision) {
char ts[40] = {0};
struct tm* ptm;
int32_t fractionLen;
char* format = NULL;
time_t quot = 0;
long mod = 0;
switch (precision) {
case TSDB_TIME_PRECISION_MILLI: {
quot = t / 1000;
fractionLen = 5;
format = ".%03" PRId64;
mod = t % 1000;
break;
}
case TSDB_TIME_PRECISION_MICRO: {
quot = t / 1000000;
fractionLen = 8;
format = ".%06" PRId64;
mod = t % 1000000;
break;
}
case TSDB_TIME_PRECISION_NANO: {
quot = t / 1000000000;
fractionLen = 11;
format = ".%09" PRId64;
mod = t % 1000000000;
break;
}
default:
fractionLen = 0;
assert(false);
}
ptm = localtime(&quot);
int32_t length = (int32_t)strftime(ts, 40, "%Y-%m-%dT%H:%M:%S", ptm);
length += snprintf(ts + length, fractionLen, format, mod);
length += (int32_t)strftime(ts + length, 40 - length, "%z", ptm);
tstrncpy(buf, ts, bufLen);
}

2
source/dnode/bnode/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src BNODE_SRC)
add_library(bnode ${BNODE_SRC})
add_library(bnode STATIC ${BNODE_SRC})
target_include_directories(
bnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/bnode"

2
source/dnode/mgmt/daemon/CMakeLists.txt
View File

@ -6,4 +6,4 @@ target_include_directories(
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(taosd dnode util os)
target_link_libraries(taosd dnode)

12
source/dnode/mgmt/impl/CMakeLists.txt
View File

@ -1,17 +1,7 @@
aux_source_directory(src DNODE_SRC)
add_library(dnode STATIC ${DNODE_SRC})
target_link_libraries(
dnode
PUBLIC cjson
PUBLIC mnode
PUBLIC vnode
PUBLIC qnode
PUBLIC snode
PUBLIC bnode
PUBLIC wal
PUBLIC sync
PUBLIC taos
PUBLIC tfs
dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor
)
target_include_directories(
dnode

1
source/dnode/mgmt/impl/inc/dndInt.h
View File

@ -36,6 +36,7 @@ extern "C" {
#include "ttime.h"
#include "tworker.h"
#include "tglobal.h"
#include "monitor.h"
#include "dnode.h"

3
source/dnode/mgmt/impl/inc/dndMnode.h
View File

@ -32,6 +32,9 @@ int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
#ifdef __cplusplus
}
#endif

31
source/dnode/mgmt/impl/src/dndEnv.c
View File

@ -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"
@ -140,7 +141,7 @@ static int32_t dndInitDir(SDnode *pDnode, SDnodeObjCfg *pCfg) {
return 0;
}
static void dndCloseImp(SDnode *pDnode) {
static void dndCloseDir(SDnode *pDnode) {
tfree(pDnode->dir.mnode);
tfree(pDnode->dir.vnodes);
tfree(pDnode->dir.dnode);
@ -260,7 +261,7 @@ void dndClose(SDnode *pDnode) {
dndCleanupMgmt(pDnode);
tfsClose(pDnode->pTfs);
dndCloseImp(pDnode);
dndCloseDir(pDnode);
free(pDnode);
dInfo("dnode object is closed, data:%p", pDnode);
}
@ -289,11 +290,7 @@ int32_t dndInit() {
}
SVnodeOpt vnodeOpt = {
.sver = tsVersion,
.nthreads = tsNumOfCommitThreads,
.putReqToVQueryQFp = dndPutReqToVQueryQ,
.sendReqToDnodeFp = dndSendReqToDnode
};
.nthreads = tsNumOfCommitThreads, .putReqToVQueryQFp = dndPutReqToVQueryQ, .sendReqToDnodeFp = dndSendReqToDnode};
if (vnodeInit(&vnodeOpt) != 0) {
dError("failed to init vnode since %s", terrstr());
@ -301,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;
}
@ -314,19 +318,8 @@ void dndCleanup() {
walCleanUp();
vnodeCleanup();
rpcCleanup();
monCleanup();
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}
// OTHER FUNCTIONS ===================================
void taosGetDisk() {
#if 0
const double unit = 1024 * 1024 * 1024;
SDiskSize diskSize = tfsGetSize(pTfs);
tfsUpdateSize(&fsMeta);
#endif
}

50
source/dnode/mgmt/impl/src/dndMgmt.c
View File

@ -22,6 +22,7 @@
#include "dndTransport.h"
#include "dndVnodes.h"
#include "dndWorker.h"
#include "monitor.h"
static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg);
@ -473,19 +474,62 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
rpcSendResponse(&rpcRsp);
}
void dndGetBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->dnode_id = dndGetDnodeId(pDnode);
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
}
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 = {0};
dndGetBasicInfo(pDnode, &basicInfo);
monSetBasicInfo(pMonitor, &basicInfo);
SMonClusterInfo clusterInfo = {0};
SMonVgroupInfo vgroupInfo = {0};
SMonGrantInfo grantInfo = {0};
if (dndGetMnodeMonitorInfo(pDnode, &clusterInfo, &vgroupInfo, &grantInfo) == 0) {
monSetClusterInfo(pMonitor, &clusterInfo);
monSetVgroupInfo(pMonitor, &vgroupInfo);
monSetGrantInfo(pMonitor, &grantInfo);
}
monSendReport(pMonitor);
monCleanupMonitorInfo(pMonitor);
}
static void *dnodeThreadRoutine(void *param) {
SDnode *pDnode = param;
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
int32_t ms = tsStatusInterval * 1000;
int64_t lastStatusTime = taosGetTimestampMs();
int64_t lastMonitorTime = lastStatusTime;
setThreadName("dnode-hb");
while (true) {
pthread_testcancel();
taosMsleep(ms);
taosMsleep(200);
if (dndGetStat(pDnode) != DND_STAT_RUNNING || pMgmt->dropped) {
continue;
}
if (dndGetStat(pDnode) == DND_STAT_RUNNING && !pMgmt->statusSent && !pMgmt->dropped) {
int64_t curTime = taosGetTimestampMs();
float statusInterval = (curTime - lastStatusTime) / 1000.0f;
if (statusInterval >= tsStatusInterval && !pMgmt->statusSent) {
dndSendStatusReq(pDnode);
lastStatusTime = curTime;
}
float monitorInterval = (curTime - lastMonitorTime) / 1000.0f;
if (monitorInterval >= tsMonitorInterval) {
dndSendMonitorReport(pDnode);
lastMonitorTime = curTime;
}
}
}

10
source/dnode/mgmt/impl/src/dndMnode.c
View File

@ -630,3 +630,13 @@ int32_t dndGetUserAuthFromMnode(SDnode *pDnode, char *user, char *spi, char *enc
dTrace("user:%s, retrieve auth spi:%d encrypt:%d", user, *spi, *encrypt);
return code;
}
int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) {
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) return -1;
int32_t code = mndGetMonitorInfo(pMnode, pClusterInfo, pVgroupInfo, pGrantInfo);
dndReleaseMnode(pDnode, pMnode);
return code;
}

2
source/dnode/mgmt/impl/test/sut/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src SUT_SRC)
add_library(sut STATIC ${SUT_SRC})
add_library(sut STATIC STATIC ${SUT_SRC})
target_link_libraries(
sut
PUBLIC dnode

10
source/dnode/mnode/impl/CMakeLists.txt
View File

@ -1,18 +1,12 @@
aux_source_directory(src MNODE_SRC)
add_library(mnode ${MNODE_SRC})
add_library(mnode STATIC ${MNODE_SRC})
target_include_directories(
mnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mnode"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
mnode
PRIVATE scheduler
PRIVATE sdb
PRIVATE wal
PRIVATE transport
PRIVATE cjson
PRIVATE sync
mnode scheduler sdb wal transport cjson sync monitor
)
if(${BUILD_TEST})

22
source/dnode/mnode/impl/inc/mndInt.h
View File

@ -21,11 +21,11 @@
#include "sdb.h"
#include "tcache.h"
#include "tep.h"
#include "tglobal.h"
#include "tqueue.h"
#include "ttime.h"
#include "wal.h"
#include "version.h"
#include "tglobal.h"
#include "wal.h"
#ifdef __cplusplus
extern "C" {
@ -38,6 +38,20 @@ typedef int32_t (*ShowMetaFp)(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *p
typedef int32_t (*ShowRetrieveFp)(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter);
typedef struct SMnodeLoad {
int64_t numOfDnode;
int64_t numOfMnode;
int64_t numOfVgroup;
int64_t numOfDatabase;
int64_t numOfSuperTable;
int64_t numOfChildTable;
int64_t numOfNormalTable;
int64_t numOfColumn;
int64_t totalPoints;
int64_t totalStorage;
int64_t compStorage;
} SMnodeLoad;
typedef struct {
const char *name;
MndInitFp initFp;
@ -104,7 +118,9 @@ int32_t mndSendReqToMnode(SMnode *pMnode, SRpcMsg *pMsg);
void mndSendRedirectRsp(SMnode *pMnode, SRpcMsg *pMsg);
void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp);
uint64_t mndGenerateUid(char *name, int32_t len) ;
uint64_t mndGenerateUid(char *name, int32_t len);
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
#ifdef __cplusplus
}

10
source/dnode/mnode/impl/src/mnode.c
View File

@ -22,6 +22,7 @@
#include "mndDb.h"
#include "mndDnode.h"
#include "mndFunc.h"
#include "mndInfoSchema.h"
#include "mndMnode.h"
#include "mndOffset.h"
#include "mndProfile.h"
@ -36,7 +37,6 @@
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
#include "mndInfoSchema.h"
#define MQ_TIMER_MS 3000
#define TRNAS_TIMER_MS 6000
@ -400,6 +400,11 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
return 0;
}
int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) {
return 0;
}
SMnodeMsg *mndInitMsg(SMnode *pMnode, SRpcMsg *pRpcMsg) {
SMnodeMsg *pMsg = taosAllocateQitem(sizeof(SMnodeMsg));
if (pMsg == NULL) {
@ -505,10 +510,9 @@ void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
}
}
// Note: uid 0 is reserved
uint64_t mndGenerateUid(char *name, int32_t len) {
int32_t hashval = MurmurHash3_32(name, len);
int32_t hashval = MurmurHash3_32(name, len);
do {
int64_t us = taosGetTimestampUs();

7
source/dnode/mnode/sdb/CMakeLists.txt
View File

@ -1,13 +1,10 @@
aux_source_directory(src MNODE_SRC)
add_library(sdb ${MNODE_SRC})
add_library(sdb STATIC ${MNODE_SRC})
target_include_directories(
sdb
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mnode/sdb"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
sdb
PRIVATE os
PRIVATE common
PRIVATE util
sdb os common util
)

2
source/dnode/qnode/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src QNODE_SRC)
add_library(qnode ${QNODE_SRC})
add_library(qnode STATIC ${QNODE_SRC})
target_include_directories(
qnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/qnode"

2
source/dnode/snode/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src SNODE_SRC)
add_library(snode ${SNODE_SRC})
add_library(snode STATIC ${SNODE_SRC})
target_include_directories(
snode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/snode"

4
source/dnode/vnode/inc/vnode.h
View File

@ -60,10 +60,6 @@ typedef struct {
} SVnodeCfg;
typedef struct {
int32_t sver;
const char *timezone;
const char *locale;
const char *charset;
uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
PutReqToVQueryQFp putReqToVQueryQFp;
SendReqToDnodeFp sendReqToDnodeFp;

1
source/libs/CMakeLists.txt
View File

@ -14,5 +14,6 @@ add_subdirectory(function)
add_subdirectory(qcom)
add_subdirectory(qworker)
add_subdirectory(tfs)
add_subdirectory(monitor)
add_subdirectory(nodes)
add_subdirectory(scalar)

2
source/libs/cache/CMakeLists.txt vendored
View File

@ -1,5 +1,5 @@
aux_source_directory(src CACHE_SRC)
add_library(cache ${CACHE_SRC})
add_library(cache STATIC ${CACHE_SRC})
target_include_directories(
cache
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/cache"

2
source/libs/catalog/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src CATALOG_SRC)
add_library(catalog ${CATALOG_SRC})
add_library(catalog STATIC ${CATALOG_SRC})
target_include_directories(
catalog
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/catalog"

2
source/libs/function/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src FUNCTION_SRC)
add_library(function ${FUNCTION_SRC})
add_library(function STATIC ${FUNCTION_SRC})
target_include_directories(
function
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/function"

2
source/libs/index/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src INDEX_SRC)
add_library(index ${INDEX_SRC})
add_library(index STATIC ${INDEX_SRC})
target_include_directories(
index
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/index"

13
source/libs/monitor/CMakeLists.txt Normal file
View File

@ -0,0 +1,13 @@
aux_source_directory(src MONITOR_SRC)
add_library(monitor STATIC ${MONITOR_SRC})
target_include_directories(
monitor
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/monitor"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(monitor os util common)
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

42
source/libs/monitor/inc/monInt.h Normal file
View File

@ -0,0 +1,42 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* 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/>.
*/
#ifndef _TD_MONITOR_INT_H_
#define _TD_MONITOR_INT_H_
#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
#endif /*_TD_MONITOR_INT_H_*/

117
source/libs/monitor/src/monitor.c Normal file
View File

@ -0,0 +1,117 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* 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 "monInt.h"
#include "taoserror.h"
#include "thttp.h"
#include "tlog.h"
#include "ttime.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(tsMonitor.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);
int64_t ms = taosGetTimestampMs();
char buf[40] = {0};
taosFormatUtcTime(buf, sizeof(buf), ms, TSDB_TIME_PRECISION_MILLI);
tjsonAddStringToObject(pJson, "ts", buf);
}
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) {
}

14
source/libs/monitor/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,14 @@
enable_testing()
aux_source_directory(. MONITOR_TEST_SRC)
add_executable(monitor_test ${MONITOR_TEST_SRC})
target_link_libraries(
monitor_test
PUBLIC monitor
PUBLIC gtest_main
)
add_test(
NAME monitor_test
COMMAND monitor_test
)

33
source/libs/monitor/test/monTest.cpp Normal file
View File

@ -0,0 +1,33 @@
/**
* @file monTest.cpp
* @author slguan (slguan@taosdata.com)
* @brief monitor module tests
* @version 1.0
* @date 2022-03-05
*
* @copyright Copyright (c) 2022
*
*/
#include <gtest/gtest.h>
#include "os.h"
#include "monitor.h"
class MonitorTest : public ::testing::Test {
protected:
static void SetUpTestSuite() { root = "/tmp/monTest"; }
static void TearDownTestSuite() {}
public:
void SetUp() override {}
void TearDown() override {}
static const char *root;
};
const char *MonitorTest::root;
TEST_F(MonitorTest, 01_Open_Close) {
}

2
source/libs/parser/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src PARSER_SRC)
add_library(parser ${PARSER_SRC})
add_library(parser STATIC ${PARSER_SRC})
target_include_directories(
parser
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/parser"

2
source/libs/planner/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src PLANNER_SRC)
add_library(planner ${PLANNER_SRC})
add_library(planner STATIC ${PLANNER_SRC})
target_include_directories(
planner
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/planner"

2
source/libs/qcom/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src QUERY_SRC)
add_library(qcom ${QUERY_SRC})
add_library(qcom STATIC ${QUERY_SRC})
target_include_directories(
qcom
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/qcom"

2
source/libs/scheduler/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src SCHEDULER_SRC)
add_library(scheduler ${SCHEDULER_SRC})
add_library(scheduler STATIC ${SCHEDULER_SRC})
target_include_directories(
scheduler

2
source/libs/sync/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src SYNC_SRC)
add_library(sync ${SYNC_SRC})
add_library(sync STATIC ${SYNC_SRC})
target_link_libraries(
sync

2
source/libs/transport/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src TRANSPORT_SRC)
add_library(transport ${TRANSPORT_SRC})
add_library(transport STATIC ${TRANSPORT_SRC})
target_include_directories(
transport
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/transport"

2
source/libs/wal/CMakeLists.txt
View File

@ -1,5 +1,5 @@
aux_source_directory(src WAL_SRC)
add_library(wal ${WAL_SRC})
add_library(wal STATIC ${WAL_SRC})
target_include_directories(
wal
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/wal"

8
source/os/CMakeLists.txt
View File

@ -1,14 +1,10 @@
aux_source_directory(src OS_SRC)
add_library(os ${OS_SRC})
add_library(os STATIC ${OS_SRC})
target_include_directories(
os
PUBLIC "${CMAKE_SOURCE_DIR}/include/os"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/include"
)
target_link_libraries(
os
PUBLIC pthread
PUBLIC dl
PUBLIC rt
PUBLIC m
os pthread dl rt m
)

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

@ -167,11 +167,9 @@ void taosGetSystemInfo() {
tsTotalMemoryMB = taosGetTotalMemory();
float tmp1, tmp2;
// taosGetDisk();
taosGetBandSpeed(&tmp1);
taosGetCpuUsage(&tmp1, &tmp2);
taosGetProcIO(&tmp1, &tmp2);
}
void taosKillSystem() {
@ -712,7 +710,6 @@ void taosGetSystemInfo() {
float tmp1, tmp2;
taosGetSysMemory(&tmp1);
taosGetProcMemory(&tmp2);
// taosGetDisk();
taosGetBandSpeed(&tmp1);
taosGetCpuUsage(&tmp1, &tmp2);
taosGetProcIO(&tmp1, &tmp2);

10
source/util/src/thttp.c
View File

@ -32,7 +32,7 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont,
fd = taosOpenTcpClientSocket(ip, port, 0);
if (fd < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to create http socket since %s", terrstr());
uError("failed to create http socket to %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
@ -46,24 +46,24 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont,
if (taosWriteSocket(fd, (void*)header, headLen) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to send http header since %s", terrstr());
uError("failed to send http header to %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
if (taosWriteSocket(fd, (void*)pCont, contLen) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to send http content since %s", terrstr());
uError("failed to send http content to %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
// read something to avoid nginx error 499
if (taosReadSocket(fd, header, 10) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to receive response since %s", terrstr());
uError("failed to receive response from %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
uInfo("send http to %s:%d, len:%d content: %s", server, port, contLen, pCont);
uInfo("send http to %s:%u, len:%d content: %s", server, port, contLen, pCont);
code = 0;
SEND_OVER:

6
source/util/src/tjson.c
View File

@ -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};