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

Merge pull request #1699 from taosdata/feature/alter 

Feature/alter
This commit is contained in:
slguan 2020-04-23 22:06:21 +08:00 committed by GitHub
parent 87158aade4 1c3021cd43
commit 0c23737dc5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 534 additions and 602 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/dnode/inc/dnodeLog.h
View File

@ -39,10 +39,6 @@ extern int32_t ddebugFlag;
#define dPrint(...) \
{ taosPrintLog("DND ", 255, __VA_ARGS__); }
#define dLError(...) taosLogError(__VA_ARGS__) dError(__VA_ARGS__)
#define dLWarn(...) taosLogWarn(__VA_ARGS__) dWarn(__VA_ARGS__)
#define dLPrint(...) taosLogPrint(__VA_ARGS__) dPrint(__VA_ARGS__)
#ifdef __cplusplus
}
#endif

4
src/dnode/src/dnodeModule.c
View File

@ -90,7 +90,7 @@ void dnodeCleanUpModules() {
int32_t dnodeInitModules() {
dnodeAllocModules();
for (int32_t module = 0; module < TSDB_MOD_MAX; ++module) {
for (EModuleType module = 0; module < TSDB_MOD_MAX; ++module) {
if (tsModule[module].initFp) {
if ((*tsModule[module].initFp)() != 0) {
dError("failed to init module:%s", tsModule[module].name);
@ -103,7 +103,7 @@ int32_t dnodeInitModules() {
}
void dnodeStartModules() {
for (int32_t module = 1; module < TSDB_MOD_MAX; ++module) {
for (EModuleType module = 1; module < TSDB_MOD_MAX; ++module) {
if (tsModule[module].enable && tsModule[module].startFp) {
if ((*tsModule[module].startFp)() != 0) {
dError("failed to start module:%s", tsModule[module].name);

11
src/dnode/src/dnodeShell.c
View File

@ -20,16 +20,17 @@
#include "taosmsg.h"
#include "trpc.h"
#include "tglobal.h"
#include "http.h"
#include "dnode.h"
#include "dnodeLog.h"
#include "dnodeRead.h"
#include "dnodeWrite.h"
#include "dnodeShell.h"
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void *tsDnodeShellRpc = NULL;
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void * tsDnodeShellRpc = NULL;
static int32_t tsDnodeQueryReqNum = 0;
static int32_t tsDnodeSubmitReqNum = 0;
@ -110,7 +111,7 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
SDnodeStatisInfo dnodeGetStatisInfo() {
SDnodeStatisInfo info = {0};
if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_RUNING) {
//info.httpReqNum = httpGetReqCount();
info.httpReqNum = httpGetReqCount();
info.queryReqNum = atomic_exchange_32(&tsDnodeQueryReqNum, 0);
info.submitReqNum = atomic_exchange_32(&tsDnodeSubmitReqNum, 0);
}

32
src/inc/monitor.h
View File

@ -22,10 +22,38 @@ extern "C" {
#include <stdint.h>
typedef struct {
char * acctId;
int64_t currentPointsPerSecond;
int64_t maxPointsPerSecond;
int64_t totalTimeSeries;
int64_t maxTimeSeries;
int64_t totalStorage;
int64_t maxStorage;
int64_t totalQueryTime;
int64_t maxQueryTime;
int64_t totalInbound;
int64_t maxInbound;
int64_t totalOutbound;
int64_t maxOutbound;
int64_t totalDbs;
int64_t maxDbs;
int64_t totalUsers;
int64_t maxUsers;
int64_t totalStreams;
int64_t maxStreams;
int64_t totalConns;
int64_t maxConns;
int8_t accessState;
} SAcctMonitorObj;
int32_t monitorInitSystem();
int32_t monitorStartSystem();
void monitorStopSystem();
void monitorCleanUpSystem();
void monitorStopSystem();
void monitorCleanUpSystem();
void monitorSaveAcctLog(SAcctMonitorObj *pMonObj);
void monitorSaveLog(int32_t level, const char *const format, ...);
void monitorExecuteSQL(char *sql);
#ifdef __cplusplus
}

1
src/inc/taosmsg.h
View File

@ -509,7 +509,6 @@ typedef struct {
int64_t pointsWritten;
uint8_t status;
uint8_t role;
uint8_t accessState;
uint8_t replica;
uint8_t reserved[5];
} SVnodeLoad;

13
src/mnode/inc/mgmtLog.h
View File

@ -21,6 +21,7 @@ extern "C" {
#endif
#include "tlog.h"
#include "monitor.h"
extern int32_t mdebugFlag;
extern int32_t sdbDebugFlag;
@ -41,9 +42,9 @@ extern int32_t sdbDebugFlag;
#define mPrint(...) \
{ taosPrintLog("MND ", 255, __VA_ARGS__); }
#define mLError(...) mError(__VA_ARGS__)
#define mLWarn(...) mWarn(__VA_ARGS__)
#define mLPrint(...) mPrint(__VA_ARGS__)
#define mLError(...) monitorSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__)
#define mLWarn(...) monitorSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)
#define mLPrint(...) monitorSaveLog(0, __VA_ARGS__); mPrint(__VA_ARGS__)
#define sdbError(...) \
if (sdbDebugFlag & DEBUG_ERROR) { \
@ -60,9 +61,9 @@ extern int32_t sdbDebugFlag;
#define sdbPrint(...) \
{ taosPrintLog("MND-SDB ", 255, __VA_ARGS__); }
#define sdbLError(...) sdbError(__VA_ARGS__)
#define sdbLWarn(...) sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) sdbPrint(__VA_ARGS__)
#define sdbLError(...) monitorSaveLog(2, __VA_ARGS__); sdbError(__VA_ARGS__)
#define sdbLWarn(...) monitorSaveLog(1, __VA_ARGS__); sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) monitorSaveLog(0, __VA_ARGS__); sdbPrint(__VA_ARGS__)
#ifdef __cplusplus
}

4
src/mnode/src/mgmtDb.c
View File

@ -884,6 +884,8 @@ void mgmtDropAllDbs(SAcctObj *pAcct) {
SDbObj *pDb = NULL;
void * pNode = NULL;
mPrint("acct:%s, all dbs will be dropped from sdb", pAcct->user);
while (1) {
pNode = sdbFetchRow(tsDbSdb, pNode, (void **)&pDb);
if (pDb == NULL) break;
@ -902,5 +904,5 @@ void mgmtDropAllDbs(SAcctObj *pAcct) {
mgmtDecDbRef(pDb);
}
mTrace("acct:%s, all dbs is is dropped from sdb", pAcct->user, numOfDbs);
mPrint("acct:%s, all dbs:%d is dropped from sdb", pAcct->user, numOfDbs);
}

6
src/mnode/src/mgmtDnode.c
View File

@ -260,7 +260,7 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
}
uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip);
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(dnodeIp);
@ -469,7 +469,7 @@ static void mgmtProcessCreateDnodeMsg(SQueuedMsg *pMsg) {
SCMCreateDnodeMsg *pCreate = pMsg->pCont;
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
uint32_t ip = inet_addr(pCreate->ip);
@ -489,7 +489,7 @@ static void mgmtProcessDropDnodeMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SCMDropDnodeMsg *pDrop = pMsg->pCont;
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
uint32_t ip = inet_addr(pDrop->ip);

2
src/mnode/src/mgmtSdb.c
View File

@ -400,6 +400,8 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
if (pTable->keyType == SDB_KEY_AUTO) {
pTable->autoIndex = MAX(pTable->autoIndex, *((uint32_t *)pOper->pObj));
} else {
pTable->autoIndex++;
}
pthread_mutex_unlock(&pTable->mutex);

36
src/mnode/src/mgmtTable.c
View File

@ -1112,12 +1112,11 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
int32_t dbNameLen = strlen(pDropDb->name);
SSuperTableObj *pTable = NULL;
mPrint("db:%s, all super tables will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsSuperTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
SSdbOper oper = {
@ -1128,10 +1127,12 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
static int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable) {
@ -1681,12 +1682,11 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
int32_t dbNameLen = strlen(pDropDb->name);
SChildTableObj *pTable = NULL;
mPrint("db:%s, all child tables will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
SSdbOper oper = {
@ -1697,11 +1697,11 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
@ -1710,12 +1710,11 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
int32_t numOfTables = 0;
SChildTableObj *pTable = NULL;
mPrint("stable:%s, all child tables will dropped from sdb", pStable->info.tableId, numOfTables);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (pTable->superTable == pStable) {
SSdbOper oper = {
@ -1726,11 +1725,12 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
mPrint("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
}
static SChildTableObj* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_t sid) {

4
src/mnode/src/mgmtUser.c
View File

@ -353,6 +353,7 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg) {
mLPrint("user:%s, is created by %s", pCreate->user, pOperUser->user);
}
} else {
mError("user:%s, no rights to create user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
@ -398,6 +399,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
code = mgmtUpdateUser(pUser);
mLPrint("user:%s, password is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
@ -440,11 +442,13 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
code = mgmtUpdateUser(pUser);
mLPrint("user:%s, privilege is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
mgmtSendSimpleResp(pMsg->thandle, code);
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
}

13
src/mnode/src/mgmtVgroup.c
View File

@ -744,12 +744,13 @@ static void mgmtProcessVnodeCfgMsg(SRpcMsg *rpcMsg) {
void mgmtDropAllVgroups(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
int32_t numOfVgroups = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SVgObj *pVgroup = NULL;
mPrint("db:%s, all vgroups will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecVgroupRef(pVgroup);
pNode = sdbFetchRow(tsVgroupSdb, pNode, (void **)&pVgroup);
if (pVgroup == NULL) break;
@ -761,12 +762,14 @@ void mgmtDropAllVgroups(SDbObj *pDropDb) {
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
numOfVgroups++;
}
mgmtSendDropVgroupMsg(pVgroup, NULL);
mgmtDecVgroupRef(pVgroup);
}
mTrace("db:%s, all vgroups is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all vgroups:%d is dropped from sdb", pDropDb->name, numOfVgroups);
}
void mgmtAlterVgroup(SVgObj *pVgroup, void *ahandle) {

5
src/plugins/http/inc/httpLog.h
View File

@ -39,9 +39,4 @@ extern int32_t httpDebugFlag;
#define httpPrint(...) \
{ taosPrintLog("HTP ", 255, __VA_ARGS__); }
#define httpLError(...) taosLogError(__VA_ARGS__) httpError(__VA_ARGS__)
#define httpLWarn(...) taosLogWarn(__VA_ARGS__) httpWarn(__VA_ARGS__)
#define httpLPrint(...) taosLogPrint(__VA_ARGS__) httpPrint(__VA_ARGS__)
#endif

42
src/plugins/monitor/inc/monitorSystem.h
View File

@ -1,42 +0,0 @@
/*
* 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 TDENGINE_MONITOR_SYSTEM_H
#define TDENGINE_MONITOR_SYSTEM_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdbool.h>
#include <stdint.h>
int32_t monitorInitSystem();
int32_t monitorStartSystem();
void monitorStopSystem();
void monitorCleanUpSystem();
void monitorSaveAcctLog(char *acctId, int64_t currentPointsPerSecond, int64_t maxPointsPerSecond,
int64_t totalTimeSeries, int64_t maxTimeSeries, int64_t totalStorage, int64_t maxStorage,
int64_t totalQueryTime, int64_t maxQueryTime, int64_t totalInbound, int64_t maxInbound,
int64_t totalOutbound, int64_t maxOutbound, int64_t totalDbs, int64_t maxDbs,
int64_t totalUsers, int64_t maxUsers, int64_t totalStreams, int64_t maxStreams,
int64_t totalConns, int64_t maxConns, int8_t accessState);
void monitorSaveLog(int level, const char *const format, ...);
#ifdef __cplusplus
}
#endif
#endif

431
src/plugins/monitor/src/monitorMain.c Normal file
View File

@ -0,0 +1,431 @@
/*
* 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 "os.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tlog.h"
#include "ttime.h"
#include "ttimer.h"
#include "tutil.h"
#include "tsystem.h"
#include "tscUtil.h"
#include "tsclient.h"
#include "dnode.h"
#include "monitor.h"
#define monitorError(...) \
if (monitorDebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR MON ", 255, __VA_ARGS__); \
}
#define monitorWarn(...) \
if (monitorDebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN MON ", monitorDebugFlag, __VA_ARGS__); \
}
#define monitorTrace(...) \
if (monitorDebugFlag & DEBUG_TRACE) { \
taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); \
}
#define monitorPrint(...) \
{ taosPrintLog("MON ", 255, __VA_ARGS__); }
#define SQL_LENGTH 1024
#define LOG_LEN_STR 80
#define IP_LEN_STR 15
#define CHECK_INTERVAL 1000
typedef enum {
MONITOR_CMD_CREATE_DB,
MONITOR_CMD_CREATE_TB_LOG,
MONITOR_CMD_CREATE_MT_DN,
MONITOR_CMD_CREATE_MT_ACCT,
MONITOR_CMD_CREATE_TB_DN,
MONITOR_CMD_CREATE_TB_ACCT_ROOT,
MONITOR_CMD_CREATE_TB_SLOWQUERY,
MONITOR_CMD_MAX
} EMonitorCommand;
typedef enum {
MONITOR_STATE_UN_INIT,
MONITOR_STATE_INITIALIZING,
MONITOR_STATE_INITIALIZED,
MONITOR_STATE_STOPPED
} EMonitorState;
typedef struct {
void * conn;
void * timer;
char privateIpStr[TSDB_IPv4ADDR_LEN];
int8_t cmdIndex;
int8_t state;
char sql[SQL_LENGTH];
void * initTimer;
void * diskTimer;
} SMonitorConn;
static SMonitorConn tsMonitorConn;
static void monitorInitConn(void *para, void *unused);
static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code);
static void monitorInitDatabase();
static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code);
static void monitorStartTimer();
static void monitorSaveSystemInfo();
static void monitorCheckDiskUsage(void *para, void *unused) {
taosGetDisk();
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
}
int32_t monitorInitSystem() {
taos_init();
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
return 0;
}
int32_t monitorStartSystem() {
monitorPrint("start monitor module");
monitorInitSystem();
taosTmrReset(monitorInitConn, 10, NULL, tscTmr, &tsMonitorConn.initTimer);
return 0;
}
static void monitorStartSystemRetry() {
if (tsMonitorConn.initTimer != NULL) {
taosTmrReset(monitorInitConn, 3000, NULL, tscTmr, &tsMonitorConn.initTimer);
}
}
static void monitorInitConn(void *para, void *unused) {
monitorPrint("starting to initialize monitor service ..");
tsMonitorConn.state = MONITOR_STATE_INITIALIZING;
if (tsMonitorConn.privateIpStr[0] == 0) {
strcpy(tsMonitorConn.privateIpStr, tsPrivateIp);
for (int32_t i = 0; i < TSDB_IPv4ADDR_LEN; ++i) {
if (tsMonitorConn.privateIpStr[i] == '.') {
tsMonitorConn.privateIpStr[i] = '_';
}
}
}
if (tsMonitorConn.conn == NULL) {
taos_connect_a(NULL, "monitor", tsInternalPass, "", 0, monitorInitConnCb, &tsMonitorConn, &(tsMonitorConn.conn));
} else {
monitorInitDatabase();
}
}
static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code) {
if (code < 0) {
monitorError("monitor:%p, connect to database failed, reason:%s", tsMonitorConn.conn, tstrerror(code));
taos_close(tsMonitorConn.conn);
tsMonitorConn.conn = NULL;
tsMonitorConn.state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
return;
}
monitorTrace("monitor:%p, connect to database success, reason:%s", tsMonitorConn.conn, tstrerror(code));
monitorInitDatabase();
}
static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
memset(sql, 0, SQL_LENGTH);
if (cmd == MONITOR_CMD_CREATE_DB) {
snprintf(sql, SQL_LENGTH,
"create database if not exists %s replica 1 days 10 keep 30 rows 1024 cache 2048 "
"ablocks 2 tblocks 32 tables 32 precision 'us'",
tsMonitorDbName);
} else if (cmd == MONITOR_CMD_CREATE_MT_DN) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.dn(ts timestamp"
", cpu_taosd float, cpu_system float, cpu_cores int"
", mem_taosd float, mem_system float, mem_total int"
", disk_used float, disk_total int"
", band_speed float"
", io_read float, io_write float"
", req_http int, req_select int, req_insert int"
") tags (ipaddr binary(%d))",
tsMonitorDbName, IP_LEN_STR + 1);
} else if (cmd == MONITOR_CMD_CREATE_TB_DN) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn_%s using %s.dn tags('%s')", tsMonitorDbName,
tsMonitorConn.privateIpStr, tsMonitorDbName, tsPrivateIp);
} else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.acct(ts timestamp "
", currentPointsPerSecond bigint, maxPointsPerSecond bigint"
", totalTimeSeries bigint, maxTimeSeries bigint"
", totalStorage bigint, maxStorage bigint"
", totalQueryTime bigint, maxQueryTime bigint"
", totalInbound bigint, maxInbound bigint"
", totalOutbound bigint, maxOutbound bigint"
", totalDbs smallint, maxDbs smallint"
", totalUsers smallint, maxUsers smallint"
", totalStreams smallint, maxStreams smallint"
", totalConns smallint, maxConns smallint"
", accessState smallint"
") tags (acctId binary(%d))",
tsMonitorDbName, TSDB_USER_LEN + 1);
} else if (cmd == MONITOR_CMD_CREATE_TB_ACCT_ROOT) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.acct_%s using %s.acct tags('%s')", tsMonitorDbName, "root",
tsMonitorDbName, "root");
} else if (cmd == MONITOR_CMD_CREATE_TB_SLOWQUERY) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.slowquery(ts timestamp, username "
"binary(%d), created_time timestamp, time bigint, sql binary(%d))",
tsMonitorDbName, TSDB_TABLE_ID_LEN, TSDB_SHOW_SQL_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_LOG) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.log(ts timestamp, level tinyint, "
"content binary(%d), ipaddr binary(%d))",
tsMonitorDbName, LOG_LEN_STR, IP_LEN_STR);
}
sql[SQL_LENGTH] = 0;
}
static void monitorInitDatabase() {
if (tsMonitorConn.cmdIndex < MONITOR_CMD_MAX) {
dnodeBuildMonitorSql(tsMonitorConn.sql, tsMonitorConn.cmdIndex);
taos_query_a(tsMonitorConn.conn, tsMonitorConn.sql, monitorInitDatabaseCb, NULL);
} else {
tsMonitorConn.state = MONITOR_STATE_INITIALIZED;
monitorPrint("monitor service init success");
monitorStartTimer();
}
}
static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code) {
if (-code == TSDB_CODE_TABLE_ALREADY_EXIST || -code == TSDB_CODE_DB_ALREADY_EXIST || code >= 0) {
monitorTrace("monitor:%p, sql success, reason:%d, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
if (tsMonitorConn.cmdIndex == MONITOR_CMD_CREATE_TB_LOG) {
monitorPrint("dnode:%s is started", tsPrivateIp);
}
tsMonitorConn.cmdIndex++;
monitorInitDatabase();
} else {
monitorError("monitor:%p, sql failed, reason:%s, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
tsMonitorConn.state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
}
}
void monitorStopSystem() {
monitorPrint("monitor module is stopped");
tsMonitorConn.state = MONITOR_STATE_STOPPED;
if (tsMonitorConn.initTimer != NULL) {
taosTmrStopA(&(tsMonitorConn.initTimer));
}
if (tsMonitorConn.timer != NULL) {
taosTmrStopA(&(tsMonitorConn.timer));
}
}
void monitorCleanUpSystem() {
monitorStopSystem();
monitorPrint("monitor module cleanup");
}
static void monitorStartTimer() {
taosTmrReset(monitorSaveSystemInfo, tsMonitorInterval * 1000, NULL, tscTmr, &tsMonitorConn.timer);
}
static void dnodeMontiorInsertAcctCallback(void *param, TAOS_RES *result, int32_t code) {
if (code < 0) {
monitorError("monitor:%p, save account info failed, code:%s", tsMonitorConn.conn, tstrerror(code));
} else if (code == 0) {
monitorError("monitor:%p, save account info failed, affect rows:%d", tsMonitorConn.conn, code);
} else {
monitorTrace("monitor:%p, save account info success, code:%s", tsMonitorConn.conn, tstrerror(code));
}
}
static void dnodeMontiorInsertSysCallback(void *param, TAOS_RES *result, int32_t code) {
if (code < 0) {
monitorError("monitor:%p, save system info failed, code:%s %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
} else if (code == 0) {
monitorError("monitor:%p, save system info failed, affect rows:%d %s", tsMonitorConn.conn, code, tsMonitorConn.sql);
} else {
monitorTrace("monitor:%p, save system info success, code:%s %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
}
}
static void dnodeMontiorInsertLogCallback(void *param, TAOS_RES *result, int32_t code) {
if (code < 0) {
monitorError("monitor:%p, save log failed, code:%s", tsMonitorConn.conn, tstrerror(code));
} else if (code == 0) {
monitorError("monitor:%p, save log failed, affect rows:%d", tsMonitorConn.conn, code);
} else {
monitorTrace("monitor:%p, save log info success, code:%s", tsMonitorConn.conn, tstrerror(code));
}
}
// unit is MB
static int32_t monitorBuildMemorySql(char *sql) {
float sysMemoryUsedMB = 0;
bool suc = taosGetSysMemory(&sysMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get sys memory info failed.", tsMonitorConn.conn);
}
float procMemoryUsedMB = 0;
suc = taosGetProcMemory(&procMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get proc memory info failed.", tsMonitorConn.conn);
}
return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
}
// unit is %
static int32_t monitorBuildCpuSql(char *sql) {
float sysCpuUsage = 0, procCpuUsage = 0;
bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
if (!suc) {
monitorError("monitor:%p, get cpu usage failed.", tsMonitorConn.conn);
}
if (sysCpuUsage <= procCpuUsage) {
sysCpuUsage = procCpuUsage + (float)0.1;
}
return sprintf(sql, ", %f, %f, %d", procCpuUsage, sysCpuUsage, tsNumOfCores);
}
// unit is GB
static int32_t monitorBuildDiskSql(char *sql) {
return sprintf(sql, ", %f, %d", (tsTotalDataDirGB - tsAvailDataDirGB), (int32_t)tsTotalDataDirGB);
}
// unit is Kb
static int32_t monitorBuildBandSql(char *sql) {
float bandSpeedKb = 0;
bool suc = taosGetBandSpeed(&bandSpeedKb);
if (!suc) {
monitorError("monitor:%p, get bandwidth speed failed.", tsMonitorConn.conn);
}
return sprintf(sql, ", %f", bandSpeedKb);
}
static int32_t monitorBuildReqSql(char *sql) {
SDnodeStatisInfo info = dnodeGetStatisInfo();
return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
}
static int32_t monitorBuildIoSql(char *sql) {
float readKB = 0, writeKB = 0;
bool suc = taosGetProcIO(&readKB, &writeKB);
if (!suc) {
monitorError("monitor:%p, get io info failed.", tsMonitorConn.conn);
}
return sprintf(sql, ", %f, %f", readKB, writeKB);
}
static void monitorSaveSystemInfo() {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) {
monitorStartTimer();
return;
}
int64_t ts = taosGetTimestampUs();
char * sql = tsMonitorConn.sql;
int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn_%s values(%" PRId64, tsMonitorDbName, tsMonitorConn.privateIpStr, ts);
pos += monitorBuildCpuSql(sql + pos);
pos += monitorBuildMemorySql(sql + pos);
pos += monitorBuildDiskSql(sql + pos);
pos += monitorBuildBandSql(sql + pos);
pos += monitorBuildIoSql(sql + pos);
pos += monitorBuildReqSql(sql + pos);
monitorTrace("monitor:%p, save system info, sql:%s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertSysCallback, "log");
if (tsMonitorConn.timer != NULL && tsMonitorConn.state != MONITOR_STATE_STOPPED) {
monitorStartTimer();
}
}
void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
char sql[1024] = {0};
sprintf(sql,
"insert into %s.acct_%s using %s.acct tags('%s') values(now"
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %d)",
tsMonitorDbName, pMon->acctId, tsMonitorDbName, pMon->acctId,
pMon->currentPointsPerSecond, pMon->maxPointsPerSecond,
pMon->totalTimeSeries, pMon->maxTimeSeries,
pMon->totalStorage, pMon->maxStorage,
pMon->totalQueryTime, pMon->maxQueryTime,
pMon->totalInbound, pMon->maxInbound,
pMon->totalOutbound, pMon->maxOutbound,
pMon->totalDbs, pMon->maxDbs,
pMon->totalUsers, pMon->maxUsers,
pMon->totalStreams, pMon->maxStreams,
pMon->totalConns, pMon->maxConns,
pMon->accessState);
monitorTrace("monitor:%p, save account info, sql %s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertAcctCallback, "account");
}
void monitorSaveLog(int32_t level, const char *const format, ...) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
va_list argpointer;
char sql[SQL_LENGTH] = {0};
int32_t max_length = SQL_LENGTH - 30;
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
int32_t len = snprintf(sql, (size_t)max_length, "import into %s.log values(%" PRId64 ", %d,'", tsMonitorDbName,
taosGetTimestampUs(), level);
va_start(argpointer, format);
len += vsnprintf(sql + len, (size_t)(max_length - len), format, argpointer);
va_end(argpointer);
if (len > max_length) len = max_length;
len += sprintf(sql + len, "', '%s')", tsPrivateIp);
sql[len++] = 0;
monitorTrace("monitor:%p, save log, sql: %s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertLogCallback, "log");
}
void monitorExecuteSQL(char *sql) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
monitorTrace("monitor:%p, execute sql: %s", tsMonitorConn.conn, sql);
// bug while insert binary
// taos_query_a(tsMonitorConn.conn, sql, NULL, NULL);
}

461
src/plugins/monitor/src/monitorSystem.c
View File

@ -1,461 +0,0 @@
/*
* 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 "os.h"
#include "tlog.h"
#include "monitor.h"
#include "dnode.h"
#include "tsclient.h"
#include "taosdef.h"
#include "tsystem.h"
#include "ttime.h"
#include "ttimer.h"
#include "tutil.h"
#include "monitorSystem.h"
#define monitorError(...) \
if (monitorDebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR MON ", 255, __VA_ARGS__); \
}
#define monitorWarn(...) \
if (monitorDebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN MON ", monitorDebugFlag, __VA_ARGS__); \
}
#define monitorTrace(...) \
if (monitorDebugFlag & DEBUG_TRACE) { \
taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); \
}
#define monitorPrint(...) \
{ taosPrintLog("MON ", 255, __VA_ARGS__); }
#define monitorLError(...) monitorError(__VA_ARGS__)
#define monitorLWarn(...) monitorWarn(__VA_ARGS__)
#define monitorLPrint(...) monitorPrint(__VA_ARGS__)
#define SQL_LENGTH 1024
#define LOG_LEN_STR 80
#define IP_LEN_STR 15
#define CHECK_INTERVAL 1000
typedef enum {
MONITOR_CMD_CREATE_DB,
MONITOR_CMD_CREATE_TB_LOG,
MONITOR_CMD_CREATE_MT_DN,
MONITOR_CMD_CREATE_MT_ACCT,
MONITOR_CMD_CREATE_TB_DN,
MONITOR_CMD_CREATE_TB_ACCT_ROOT,
MONITOR_CMD_CREATE_TB_SLOWQUERY,
MONITOR_CMD_MAX
} MonitorCommand;
typedef enum {
MONITOR_STATE_UN_INIT,
MONITOR_STATE_INITIALIZING,
MONITOR_STATE_INITIALIZED,
MONITOR_STATE_STOPPED
} MonitorState;
typedef struct {
void * conn;
void * timer;
char privateIpStr[TSDB_IPv4ADDR_LEN];
int8_t cmdIndex;
int8_t state;
char sql[SQL_LENGTH];
void * initTimer;
void * diskTimer;
} MonitorConn;
MonitorConn *monitor = NULL;
TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int),
void *param, void **taos);
void monitorInitConn(void *para, void *unused);
void monitorInitConnCb(void *param, TAOS_RES *result, int code);
void monitorInitDatabase();
void monitorInitDatabaseCb(void *param, TAOS_RES *result, int code);
void monitorStartTimer();
void monitorSaveSystemInfo();
void monitorSaveLog(int level, const char *const format, ...);
void monitorSaveAcctLog(char *acctId, int64_t currentPointsPerSecond, int64_t maxPointsPerSecond,
int64_t totalTimeSeries, int64_t maxTimeSeries, int64_t totalStorage, int64_t maxStorage,
int64_t totalQueryTime, int64_t maxQueryTime, int64_t totalInbound, int64_t maxInbound,
int64_t totalOutbound, int64_t maxOutbound, int64_t totalDbs, int64_t maxDbs,
int64_t totalUsers, int64_t maxUsers, int64_t totalStreams, int64_t maxStreams,
int64_t totalConns, int64_t maxConns, int8_t accessState);
void (*mnodeCountRequestFp)(SDnodeStatisInfo *info) = NULL;
void monitorExecuteSQL(char *sql);
void monitorCheckDiskUsage(void *para, void *unused) {
taosGetDisk();
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &monitor->diskTimer);
}
int monitorInitSystem() {
monitor = (MonitorConn *)malloc(sizeof(MonitorConn));
memset(monitor, 0, sizeof(MonitorConn));
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &monitor->diskTimer);
return 0;
}
int monitorStartSystem() {
if (monitor == NULL) {
monitorInitSystem();
}
taosTmrReset(monitorInitConn, 10, NULL, tscTmr, &monitor->initTimer);
return 0;
}
void monitorStartSystemRetry() {
if (monitor->initTimer != NULL) {
taosTmrReset(monitorInitConn, 3000, NULL, tscTmr, &monitor->initTimer);
}
}
void monitorInitConn(void *para, void *unused) {
monitorPrint("starting to initialize monitor service ..");
monitor->state = MONITOR_STATE_INITIALIZING;
if (monitor->privateIpStr[0] == 0) {
strcpy(monitor->privateIpStr, tsPrivateIp);
for (int i = 0; i < TSDB_IPv4ADDR_LEN; ++i) {
if (monitor->privateIpStr[i] == '.') {
monitor->privateIpStr[i] = '_';
}
}
}
if (monitor->conn == NULL) {
taos_connect_a(NULL, "monitor", tsInternalPass, "", 0, monitorInitConnCb, monitor, &(monitor->conn));
} else {
monitorInitDatabase();
}
}
void monitorInitConnCb(void *param, TAOS_RES *result, int code) {
if (code < 0) {
monitorError("monitor:%p, connect to database failed, code:%d", monitor->conn, code);
taos_close(monitor->conn);
monitor->conn = NULL;
monitor->state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
return;
}
monitorTrace("monitor:%p, connect to database success, code:%d", monitor->conn, code);
monitorInitDatabase();
}
void dnodeBuildMonitorSql(char *sql, int cmd) {
memset(sql, 0, SQL_LENGTH);
if (cmd == MONITOR_CMD_CREATE_DB) {
snprintf(sql, SQL_LENGTH,
"create database if not exists %s replica 1 days 10 keep 30 rows 1024 cache 2048 "
"ablocks 2 tblocks 32 tables 32 precision 'us'",
tsMonitorDbName);
} else if (cmd == MONITOR_CMD_CREATE_MT_DN) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.dn(ts timestamp"
", cpu_taosd float, cpu_system float, cpu_cores int"
", mem_taosd float, mem_system float, mem_total int"
", disk_used float, disk_total int"
", band_speed float"
", io_read float, io_write float"
", req_http int, req_select int, req_insert int"
") tags (ipaddr binary(%d))",
tsMonitorDbName, IP_LEN_STR + 1);
} else if (cmd == MONITOR_CMD_CREATE_TB_DN) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn_%s using %s.dn tags('%s')", tsMonitorDbName,
monitor->privateIpStr, tsMonitorDbName, tsPrivateIp);
} else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.acct(ts timestamp "
", currentPointsPerSecond bigint, maxPointsPerSecond bigint"
", totalTimeSeries bigint, maxTimeSeries bigint"
", totalStorage bigint, maxStorage bigint"
", totalQueryTime bigint, maxQueryTime bigint"
", totalInbound bigint, maxInbound bigint"
", totalOutbound bigint, maxOutbound bigint"
", totalDbs smallint, maxDbs smallint"
", totalUsers smallint, maxUsers smallint"
", totalStreams smallint, maxStreams smallint"
", totalConns smallint, maxConns smallint"
", accessState smallint"
") tags (acctId binary(%d))",
tsMonitorDbName, TSDB_USER_LEN + 1);
} else if (cmd == MONITOR_CMD_CREATE_TB_ACCT_ROOT) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.acct_%s using %s.acct tags('%s')", tsMonitorDbName, "root",
tsMonitorDbName, "root");
} else if (cmd == MONITOR_CMD_CREATE_TB_SLOWQUERY) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.slowquery(ts timestamp, username "
"binary(%d), created_time timestamp, time bigint, sql binary(%d))",
tsMonitorDbName, TSDB_TABLE_ID_LEN, TSDB_SHOW_SQL_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_LOG) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.log(ts timestamp, level tinyint, "
"content binary(%d), ipaddr binary(%d))",
tsMonitorDbName, LOG_LEN_STR, IP_LEN_STR);
}
sql[SQL_LENGTH] = 0;
}
void monitorInitDatabase() {
if (monitor->cmdIndex < MONITOR_CMD_MAX) {
dnodeBuildMonitorSql(monitor->sql, monitor->cmdIndex);
taos_query_a(monitor->conn, monitor->sql, monitorInitDatabaseCb, NULL);
} else {
monitor->state = MONITOR_STATE_INITIALIZED;
monitorPrint("monitor service init success");
monitorStartTimer();
}
}
void monitorInitDatabaseCb(void *param, TAOS_RES *result, int code) {
if (-code == TSDB_CODE_TABLE_ALREADY_EXIST || -code == TSDB_CODE_DB_ALREADY_EXIST || code >= 0) {
monitorTrace("monitor:%p, sql success, code:%d, %s", monitor->conn, code, monitor->sql);
if (monitor->cmdIndex == MONITOR_CMD_CREATE_TB_LOG) {
monitorLPrint("dnode:%s is started", tsPrivateIp);
}
monitor->cmdIndex++;
monitorInitDatabase();
} else {
monitorError("monitor:%p, sql failed, code:%d, %s", monitor->conn, code, monitor->sql);
monitor->state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
}
}
void monitorStopSystem() {
if (monitor == NULL) {
return;
}
monitorLPrint("dnode:%s monitor module is stopped", tsPrivateIp);
monitor->state = MONITOR_STATE_STOPPED;
// taosLogFp = NULL;
if (monitor->initTimer != NULL) {
taosTmrStopA(&(monitor->initTimer));
}
if (monitor->timer != NULL) {
taosTmrStopA(&(monitor->timer));
}
}
void monitorCleanUpSystem() {
monitorPrint("monitor service cleanup");
monitorStopSystem();
}
void monitorStartTimer() {
taosTmrReset(monitorSaveSystemInfo, tsMonitorInterval * 1000, NULL, tscTmr, &monitor->timer);
}
void dnodeMontiorInsertAcctCallback(void *param, TAOS_RES *result, int code) {
if (code < 0) {
monitorError("monitor:%p, save account info failed, code:%d", monitor->conn, code);
} else if (code == 0) {
monitorError("monitor:%p, save account info failed, affect rows:%d", monitor->conn, code);
} else {
monitorTrace("monitor:%p, save account info success, code:%d", monitor->conn, code);
}
}
void dnodeMontiorInsertSysCallback(void *param, TAOS_RES *result, int code) {
if (code < 0) {
monitorError("monitor:%p, save system info failed, code:%d %s", monitor->conn, code, monitor->sql);
} else if (code == 0) {
monitorError("monitor:%p, save system info failed, affect rows:%d %s", monitor->conn, code, monitor->sql);
} else {
monitorTrace("monitor:%p, save system info success, code:%d %s", monitor->conn, code, monitor->sql);
}
}
void dnodeMontiorInsertLogCallback(void *param, TAOS_RES *result, int code) {
if (code < 0) {
monitorError("monitor:%p, save log failed, code:%d", monitor->conn, code);
} else if (code == 0) {
monitorError("monitor:%p, save log failed, affect rows:%d", monitor->conn, code);
} else {
monitorTrace("monitor:%p, save log info success, code:%d", monitor->conn, code);
}
}
// unit is MB
int monitorBuildMemorySql(char *sql) {
float sysMemoryUsedMB = 0;
bool suc = taosGetSysMemory(&sysMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get sys memory info failed.", monitor->conn);
}
float procMemoryUsedMB = 0;
suc = taosGetProcMemory(&procMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get proc memory info failed.", monitor->conn);
}
return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
}
// unit is %
int monitorBuildCpuSql(char *sql) {
float sysCpuUsage = 0, procCpuUsage = 0;
bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
if (!suc) {
monitorError("monitor:%p, get cpu usage failed.", monitor->conn);
}
if (sysCpuUsage <= procCpuUsage) {
sysCpuUsage = procCpuUsage + (float)0.1;
}
return sprintf(sql, ", %f, %f, %d", procCpuUsage, sysCpuUsage, tsNumOfCores);
}
// unit is GB
int monitorBuildDiskSql(char *sql) {
return sprintf(sql, ", %f, %d", (tsTotalDataDirGB - tsAvailDataDirGB), (int32_t)tsTotalDataDirGB);
}
// unit is Kb
int monitorBuildBandSql(char *sql) {
float bandSpeedKb = 0;
bool suc = taosGetBandSpeed(&bandSpeedKb);
if (!suc) {
monitorError("monitor:%p, get bandwidth speed failed.", monitor->conn);
}
return sprintf(sql, ", %f", bandSpeedKb);
}
int monitorBuildReqSql(char *sql) {
SDnodeStatisInfo info;
info.httpReqNum = info.submitReqNum = info.queryReqNum = 0;
(*mnodeCountRequestFp)(&info);
return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
}
int monitorBuildIoSql(char *sql) {
float readKB = 0, writeKB = 0;
bool suc = taosGetProcIO(&readKB, &writeKB);
if (!suc) {
monitorError("monitor:%p, get io info failed.", monitor->conn);
}
return sprintf(sql, ", %f, %f", readKB, writeKB);
}
void monitorSaveSystemInfo() {
if (monitor->state != MONITOR_STATE_INITIALIZED) {
return;
}
if (mnodeCountRequestFp == NULL) {
return;
}
int64_t ts = taosGetTimestampUs();
char * sql = monitor->sql;
int pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn_%s values(%" PRId64, tsMonitorDbName, monitor->privateIpStr, ts);
pos += monitorBuildCpuSql(sql + pos);
pos += monitorBuildMemorySql(sql + pos);
pos += monitorBuildDiskSql(sql + pos);
pos += monitorBuildBandSql(sql + pos);
pos += monitorBuildIoSql(sql + pos);
pos += monitorBuildReqSql(sql + pos);
monitorTrace("monitor:%p, save system info, sql:%s", monitor->conn, sql);
taos_query_a(monitor->conn, sql, dnodeMontiorInsertSysCallback, "log");
if (monitor->timer != NULL && monitor->state != MONITOR_STATE_STOPPED) {
monitorStartTimer();
}
}
void monitorSaveAcctLog(char *acctId, int64_t currentPointsPerSecond, int64_t maxPointsPerSecond,
int64_t totalTimeSeries, int64_t maxTimeSeries, int64_t totalStorage, int64_t maxStorage,
int64_t totalQueryTime, int64_t maxQueryTime, int64_t totalInbound, int64_t maxInbound,
int64_t totalOutbound, int64_t maxOutbound, int64_t totalDbs, int64_t maxDbs,
int64_t totalUsers, int64_t maxUsers, int64_t totalStreams, int64_t maxStreams,
int64_t totalConns, int64_t maxConns, int8_t accessState) {
if (monitor == NULL) return;
if (monitor->state != MONITOR_STATE_INITIALIZED) return;
char sql[1024] = {0};
sprintf(sql,
"insert into %s.acct_%s using %s.acct tags('%s') values(now"
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %" PRId64 ", %" PRId64
", %d)",
tsMonitorDbName, acctId, tsMonitorDbName, acctId,
currentPointsPerSecond, maxPointsPerSecond,
totalTimeSeries, maxTimeSeries,
totalStorage, maxStorage,
totalQueryTime, maxQueryTime,
totalInbound, maxInbound,
totalOutbound, maxOutbound,
totalDbs, maxDbs,
totalUsers, maxUsers,
totalStreams, maxStreams,
totalConns, maxConns,
accessState);
monitorTrace("monitor:%p, save account info, sql %s", monitor->conn, sql);
taos_query_a(monitor->conn, sql, dnodeMontiorInsertAcctCallback, "account");
}
void monitorSaveLog(int level, const char *const format, ...) {
va_list argpointer;
char sql[SQL_LENGTH] = {0};
int max_length = SQL_LENGTH - 30;
if (monitor->state != MONITOR_STATE_INITIALIZED) {
return;
}
int len = snprintf(sql, (size_t)max_length, "import into %s.log values(%" PRId64 ", %d,'", tsMonitorDbName,
taosGetTimestampUs(), level);
va_start(argpointer, format);
len += vsnprintf(sql + len, (size_t)(max_length - len), format, argpointer);
va_end(argpointer);
if (len > max_length) len = max_length;
len += sprintf(sql + len, "', '%s')", tsPrivateIp);
sql[len++] = 0;
monitorTrace("monitor:%p, save log, sql: %s", monitor->conn, sql);
taos_query_a(monitor->conn, sql, dnodeMontiorInsertLogCallback, "log");
}
void monitorExecuteSQL(char *sql) {
monitorTrace("monitor:%p, execute sql: %s", monitor->conn, sql);
taos_query_a(monitor->conn, sql, NULL, NULL);
}

4
src/vnode/inc/vnodeLog.h
View File

@ -39,10 +39,6 @@ extern int32_t ddebugFlag;
#define dPrint(...) \
{ taosPrintLog("DND ", 255, __VA_ARGS__); }
#define dLError(...) taosLogError(__VA_ARGS__) dError(__VA_ARGS__)
#define dLWarn(...) taosLogWarn(__VA_ARGS__) dWarn(__VA_ARGS__)
#define dLPrint(...) taosLogPrint(__VA_ARGS__) dPrint(__VA_ARGS__)
#ifdef __cplusplus
}
#endif

3
src/vnode/src/vnodeMain.c
View File

@ -332,6 +332,9 @@ static void vnodeBuildVloadMsg(char *pNode, void * param) {
SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++];
pLoad->vgId = htonl(pVnode->vgId);
pLoad->totalStorage = htobe64(pLoad->totalStorage);
pLoad->compStorage = htobe64(pLoad->compStorage);
pLoad->pointsWritten = htobe64(pLoad->pointsWritten);
pLoad->status = pVnode->status;
pLoad->role = pVnode->role;
pLoad->replica = pVnode->syncCfg.replica;

2
tests/script/sh/deploy.sh
View File

@ -93,7 +93,7 @@ echo "privateIp $NODE_IP" >> $TAOS_CFG
echo "dDebugFlag 199" >> $TAOS_CFG
echo "mDebugFlag 199" >> $TAOS_CFG
echo "sdbDebugFlag 199" >> $TAOS_CFG
echo "rpcDebugFlag 135" >> $TAOS_CFG
echo "rpcDebugFlag 131" >> $TAOS_CFG
echo "tmrDebugFlag 131" >> $TAOS_CFG
echo "cDebugFlag 135" >> $TAOS_CFG
echo "httpDebugFlag 131" >> $TAOS_CFG

13
tests/script/unique/account/account_delete.sim
View File

@ -59,10 +59,17 @@ endi
sql drop account oroot
print ============= step4
sleep 2000
$x = 0
show4:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show dnodes
if $data02 != 0 then
return -1
if $data03 != 0 then
goto show4
endi
print ============= step5

2
tests/script/unique/account/basic.sim
View File

@ -1,6 +1,6 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -m 192.168.0.1 -i 192.168.0.1
system sh/exec_up_up.sh -n dnode1 -s start
system sh/exec_up.sh -n dnode1 -s start
sql connect
print =============== show accounts

36
tests/script/unique/account/monitor.sim
View File

@ -1,36 +0,0 @@
system sh/stop_dnodes.sh
system sh/ip.sh -i 1 -s up
system sh/deploy.sh -n dnode1 -m 192.168.0.1 -i 192.168.0.1
system sh/cfg.sh -n dnode1 -c monitor -v 0
print ========== step1
system sh/cfg.sh -n dnode1 -c monitor -v 1
system sh/cfg.sh -n dnode1 -c monitorInterval -v 1
system sh/exec_up.sh -n dnode1 -s start
sql connect
sleep 3000
print ========== step2
sql drop database log -x step21
return -1
step21:
sql drop table log.cpu -x step22
return -1
step22:
sql drop user log -x step23
return -1
step23:
print ========== step3
sleep 3000
sql select * from log.dn
if $rows == 0 then
return -1
endi

2
tests/script/unique/account/paras.sim
View File

@ -1,6 +1,6 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -m 192.168.0.1 -i 192.168.0.1
system sh/exec_up_up.sh -n dnode1 -s start
system sh/exec_up.sh -n dnode1 -s start
sql connect
print =============== show accounts

3
tests/script/unique/account/testSuite.sim
View File

@ -6,4 +6,7 @@ run unique/account/pass_len.sim
run unique/account/user_create.sim
run unique/account/user_len.sim
run unique/account/authority.sim
run unique/account/basic.sim
run unique/account/paras.sim
run unique/account/usage.sim
run unique/account/monitor.sim

2
tests/script/unique/account/usage.sim
View File

@ -1,6 +1,6 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -m 192.168.0.1 -i 192.168.0.1
system sh/exec_up_up.sh -n dnode1 -s start
system sh/exec_up.sh -n dnode1 -s start
sql connect
print =============== show accounts