未验证 提交 0c971b9a 编写于 作者: S Shengliang Guan 提交者: GitHub

Merge pull request #4418 from taosdata/feature/wal

[TD-2270]<fix>: race condition in balance module while scheduled by timer
...@@ -236,7 +236,7 @@ ...@@ -236,7 +236,7 @@
# httpDebugFlag 131 # httpDebugFlag 131
# debug flag for monitor # debug flag for monitor
# monitorDebugFlag 131 # monDebugFlag 131
# debug flag for query # debug flag for query
# qDebugflag 131 # qDebugflag 131
......
/*
* 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_BALANCE_INT_H
#define TDENGINE_BALANCE_INT_H
#ifdef __cplusplus
extern "C" {
#endif
#include "mnodeInt.h"
#include "mnodeDef.h"
#include "mnodeDnode.h"
typedef struct {
int32_t size;
int32_t maxSize;
SDnodeObj **list;
} SBnDnodes;
typedef struct {
void * timer;
bool stop;
pthread_mutex_t mutex;
pthread_cond_t cond;
pthread_t thread;
} SBnThread;
typedef struct {
pthread_mutex_t mutex;
} SBnMgmt;
int32_t bnInit();
void bnCleanUp();
bool bnStart();
void bnCheckStatus();
void bnCheckModules();
extern SBnDnodes tsBnDnodes;
extern void *tsMnodeTmr;
#ifdef __cplusplus
}
#endif
#endif
/*
* 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_BALANCE_SCORE_H
#define TDENGINE_BALANCE_SCORE_H
#ifdef __cplusplus
extern "C" {
#endif
#include "bnInt.h"
void bnInitDnodes();
void bnCleanupDnodes();
void bnAccquireDnodes();
void bnReleaseDnodes();
float bnTryCalcDnodeScore(SDnodeObj *pDnode, int32_t extraVnode);
#ifdef __cplusplus
}
#endif
#endif
/*
* 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_BALANCE_THREAD_H
#define TDENGINE_BALANCE_THREAD_H
#ifdef __cplusplus
extern "C" {
#endif
#include "bnInt.h"
int32_t bnInitThread();
void bnCleanupThread();
void bnNotify();
void bnStartTimer(int64_t mseconds);
#ifdef __cplusplus
}
#endif
#endif
/*
* 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 "tglobal.h"
#include "mnodeShow.h"
#include "mnodeUser.h"
#include "bnScore.h"
SBnDnodes tsBnDnodes;
static int32_t bnGetScoresMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t bnRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t bnCalcCpuScore(SDnodeObj *pDnode) {
if (pDnode->cpuAvgUsage < 80)
return 0;
else if (pDnode->cpuAvgUsage < 90)
return 10;
else
return 50;
}
static int32_t bnCalcMemoryScore(SDnodeObj *pDnode) {
if (pDnode->memoryAvgUsage < 80)
return 0;
else if (pDnode->memoryAvgUsage < 90)
return 10;
else
return 50;
}
static int32_t bnCalcDiskScore(SDnodeObj *pDnode) {
if (pDnode->diskAvgUsage < 80)
return 0;
else if (pDnode->diskAvgUsage < 90)
return 10;
else
return 50;
}
static int32_t bnCalcBandScore(SDnodeObj *pDnode) {
if (pDnode->bandwidthUsage < 30)
return 0;
else if (pDnode->bandwidthUsage < 80)
return 10;
else
return 50;
}
static float bnCalcModuleScore(SDnodeObj *pDnode) {
if (pDnode->numOfCores <= 0) return 0;
if (pDnode->isMgmt) {
return (float)tsMnodeEqualVnodeNum / pDnode->numOfCores;
}
return 0;
}
static float bnCalcVnodeScore(SDnodeObj *pDnode, int32_t extra) {
if (pDnode->status == TAOS_DN_STATUS_DROPPING || pDnode->status == TAOS_DN_STATUS_OFFLINE) return 100000000;
if (pDnode->numOfCores <= 0) return 0;
return (float)(pDnode->openVnodes + extra) / pDnode->numOfCores;
}
/**
* calc singe score, such as cpu/memory/disk/bandwitdh/vnode
* 1. get the score config
* 2. if the value is out of range, use border data
* 3. otherwise use interpolation method
**/
static void bnCalcDnodeScore(SDnodeObj *pDnode) {
pDnode->score = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) +
bnCalcBandScore(pDnode) + bnCalcModuleScore(pDnode) + bnCalcVnodeScore(pDnode, 0) +
pDnode->customScore;
}
float bnTryCalcDnodeScore(SDnodeObj *pDnode, int32_t extra) {
int32_t systemScore = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) +
bnCalcBandScore(pDnode);
float moduleScore = bnCalcModuleScore(pDnode);
float vnodeScore = bnCalcVnodeScore(pDnode, extra);
float score = systemScore + moduleScore + vnodeScore + pDnode->customScore;
return score;
}
void bnInitDnodes() {
mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_SCORES, bnGetScoresMeta);
mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_SCORES, bnRetrieveScores);
mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_SCORES, mnodeCancelGetNextDnode);
memset(&tsBnDnodes, 0, sizeof(SBnDnodes));
tsBnDnodes.maxSize = 16;
tsBnDnodes.list = calloc(tsBnDnodes.maxSize, sizeof(SDnodeObj *));
}
void bnCleanupDnodes() {
if (tsBnDnodes.list != NULL) {
free(tsBnDnodes.list);
tsBnDnodes.list = NULL;
}
}
static void bnCheckDnodesSize(int32_t dnodesNum) {
if (tsBnDnodes.maxSize <= dnodesNum) {
tsBnDnodes.maxSize = dnodesNum * 2;
tsBnDnodes.list = realloc(tsBnDnodes.list, tsBnDnodes.maxSize * sizeof(SDnodeObj *));
}
}
void bnAccquireDnodes() {
int32_t dnodesNum = mnodeGetDnodesNum();
bnCheckDnodesSize(dnodesNum);
void * pIter = NULL;
SDnodeObj *pDnode = NULL;
int32_t dnodeIndex = 0;
while (1) {
if (dnodeIndex >= dnodesNum) {
mnodeCancelGetNextDnode(pIter);
break;
}
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (pDnode->status == TAOS_DN_STATUS_OFFLINE) {
mnodeDecDnodeRef(pDnode);
continue;
}
bnCalcDnodeScore(pDnode);
int32_t orderIndex = dnodeIndex;
for (; orderIndex > 0; --orderIndex) {
if (pDnode->score > tsBnDnodes.list[orderIndex - 1]->score) {
break;
}
tsBnDnodes.list[orderIndex] = tsBnDnodes.list[orderIndex - 1];
}
tsBnDnodes.list[orderIndex] = pDnode;
dnodeIndex++;
}
tsBnDnodes.size = dnodeIndex;
}
void bnReleaseDnodes() {
for (int32_t i = 0; i < tsBnDnodes.size; ++i) {
SDnodeObj *pDnode = tsBnDnodes.list[i];
if (pDnode != NULL) {
mnodeDecDnodeRef(pDnode);
}
}
}
static int32_t bnGetScoresMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SUserObj *pUser = mnodeGetUserFromConn(pConn);
if (pUser == NULL) return 0;
if (strcmp(pUser->pAcct->user, "root") != 0) {
mnodeDecUserRef(pUser);
return TSDB_CODE_MND_NO_RIGHTS;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
strcpy(pSchema[cols].name, "system scores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
strcpy(pSchema[cols].name, "custom scores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
strcpy(pSchema[cols].name, "module scores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
strcpy(pSchema[cols].name, "vnode scores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
strcpy(pSchema[cols].name, "total scores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "open vnodes");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "cpu cores");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 18 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "balance state");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pMeta->numOfColumns = htons(cols);
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = mnodeGetDnodesNum();
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pIter = NULL;
mnodeDecUserRef(pUser);
return 0;
}
static int32_t bnRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
int32_t numOfRows = 0;
SDnodeObj *pDnode = NULL;
char * pWrite;
int32_t cols = 0;
while (numOfRows < rows) {
pShow->pIter = mnodeGetNextDnode(pShow->pIter, &pDnode);
if (pDnode == NULL) break;
int32_t systemScore = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) + bnCalcBandScore(pDnode);
float moduleScore = bnCalcModuleScore(pDnode);
float vnodeScore = bnCalcVnodeScore(pDnode, 0);
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int16_t *)pWrite = pDnode->dnodeId;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(float *)pWrite = systemScore;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(float *)pWrite = pDnode->customScore;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(float *)pWrite = (int32_t)moduleScore;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(float *)pWrite = (int32_t)vnodeScore;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(float *)pWrite = (int32_t)(vnodeScore + moduleScore + pDnode->customScore + systemScore);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t *)pWrite = pDnode->openVnodes;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t *)pWrite = pDnode->numOfCores;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_TO_VARSTR(pWrite, mnodeGetDnodeStatusStr(pDnode->status));
cols++;
numOfRows++;
mnodeDecDnodeRef(pDnode);
}
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
/*
* 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 "ttimer.h"
#include "tglobal.h"
#include "mnodeSdb.h"
#include "bnThread.h"
static SBnThread tsBnThread;
static void *bnThreadFunc(void *arg) {
while (1) {
pthread_mutex_lock(&tsBnThread.mutex);
if (tsBnThread.stop) {
pthread_mutex_unlock(&tsBnThread.mutex);
break;
}
pthread_cond_wait(&tsBnThread.cond, &tsBnThread.mutex);
bool updateSoon = bnStart();
bnStartTimer(updateSoon ? 1000 : -1);
pthread_mutex_unlock(&(tsBnThread.mutex));
}
return NULL;
}
int32_t bnInitThread() {
memset(&tsBnThread, 0, sizeof(SBnThread));
tsBnThread.stop = false;
pthread_mutex_init(&tsBnThread.mutex, NULL);
pthread_cond_init(&tsBnThread.cond, NULL);
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_DETACHED);
int32_t ret = pthread_create(&tsBnThread.thread, &thattr, bnThreadFunc, NULL);
pthread_attr_destroy(&thattr);
if (ret != 0) {
mError("failed to create balance thread since %s", strerror(errno));
return -1;
}
bnStartTimer(2000);
mDebug("balance thread is created");
return 0;
}
void bnCleanupThread() {
mDebug("balance thread will be cleanup");
if (tsBnThread.timer != NULL) {
taosTmrStopA(&tsBnThread.timer);
tsBnThread.timer = NULL;
mDebug("stop balance timer");
}
pthread_mutex_lock(&tsBnThread.mutex);
tsBnThread.stop = true;
pthread_cond_signal(&tsBnThread.cond);
pthread_mutex_unlock(&(tsBnThread.mutex));
pthread_join(tsBnThread.thread, NULL);
pthread_cond_destroy(&tsBnThread.cond);
pthread_mutex_destroy(&tsBnThread.mutex);
}
static void bnPostSignal() {
pthread_mutex_lock(&tsBnThread.mutex);
pthread_cond_signal(&tsBnThread.cond);
pthread_mutex_unlock(&(tsBnThread.mutex));
}
/*
* once sdb work as mater, then tsAccessSquence reset to zero
* increase tsAccessSquence every balance interval
*/
static void bnProcessTimer(void *handle, void *tmrId) {
if (!sdbIsMaster()) return;
if (tsBnThread.stop) return;
tsBnThread.timer = NULL;
tsAccessSquence++;
bnCheckStatus();
bnStartTimer(-1);
if (handle == NULL) {
if (tsAccessSquence % tsBalanceInterval == 0) {
mDebug("balance function is scheduled by timer");
bnPostSignal();
}
} else {
int64_t mseconds = (int64_t)handle;
mDebug("balance function is scheduled by event for %" PRId64 " mseconds arrived", mseconds);
bnPostSignal();
}
}
void bnStartTimer(int64_t mseconds) {
if (tsBnThread.stop) return;
bool updateSoon = (mseconds != -1);
if (updateSoon) {
taosTmrReset(bnProcessTimer, mseconds, (void *)mseconds, tsMnodeTmr, &tsBnThread.timer);
} else {
taosTmrReset(bnProcessTimer, tsStatusInterval * 1000, NULL, tsMnodeTmr, &tsBnThread.timer);
}
}
void bnNotify() {
bnStartTimer(500);
}
...@@ -5103,7 +5103,7 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) { ...@@ -5103,7 +5103,7 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
const int tokenDebugFlagEnd = 20; const int tokenDebugFlagEnd = 20;
const SDNodeDynConfOption cfgOptions[] = { const SDNodeDynConfOption cfgOptions[] = {
{"resetLog", 8}, {"resetQueryCache", 15}, {"balance", 7}, {"monitor", 7}, {"resetLog", 8}, {"resetQueryCache", 15}, {"balance", 7}, {"monitor", 7},
{"debugFlag", 9}, {"monitorDebugFlag", 16}, {"vDebugFlag", 10}, {"mDebugFlag", 10}, {"debugFlag", 9}, {"monDebugFlag", 12}, {"vDebugFlag", 10}, {"mDebugFlag", 10},
{"cDebugFlag", 10}, {"httpDebugFlag", 13}, {"qDebugflag", 10}, {"sdbDebugFlag", 12}, {"cDebugFlag", 10}, {"httpDebugFlag", 13}, {"qDebugflag", 10}, {"sdbDebugFlag", 12},
{"uDebugFlag", 10}, {"tsdbDebugFlag", 13}, {"sDebugflag", 10}, {"rpcDebugFlag", 12}, {"uDebugFlag", 10}, {"tsdbDebugFlag", 13}, {"sDebugflag", 10}, {"rpcDebugFlag", 12},
{"dDebugFlag", 10}, {"mqttDebugFlag", 13}, {"wDebugFlag", 10}, {"tmrDebugFlag", 12}, {"dDebugFlag", 10}, {"mqttDebugFlag", 13}, {"wDebugFlag", 10}, {"tmrDebugFlag", 12},
......
...@@ -179,7 +179,7 @@ extern int32_t tmrDebugFlag; ...@@ -179,7 +179,7 @@ extern int32_t tmrDebugFlag;
extern int32_t sdbDebugFlag; extern int32_t sdbDebugFlag;
extern int32_t httpDebugFlag; extern int32_t httpDebugFlag;
extern int32_t mqttDebugFlag; extern int32_t mqttDebugFlag;
extern int32_t monitorDebugFlag; extern int32_t monDebugFlag;
extern int32_t uDebugFlag; extern int32_t uDebugFlag;
extern int32_t rpcDebugFlag; extern int32_t rpcDebugFlag;
extern int32_t odbcDebugFlag; extern int32_t odbcDebugFlag;
......
...@@ -209,7 +209,7 @@ int32_t jniDebugFlag = 131; ...@@ -209,7 +209,7 @@ int32_t jniDebugFlag = 131;
int32_t odbcDebugFlag = 131; int32_t odbcDebugFlag = 131;
int32_t httpDebugFlag = 131; int32_t httpDebugFlag = 131;
int32_t mqttDebugFlag = 131; int32_t mqttDebugFlag = 131;
int32_t monitorDebugFlag = 131; int32_t monDebugFlag = 131;
int32_t qDebugFlag = 131; int32_t qDebugFlag = 131;
int32_t rpcDebugFlag = 131; int32_t rpcDebugFlag = 131;
int32_t uDebugFlag = 131; int32_t uDebugFlag = 131;
...@@ -219,9 +219,9 @@ int32_t wDebugFlag = 135; ...@@ -219,9 +219,9 @@ int32_t wDebugFlag = 135;
int32_t tsdbDebugFlag = 131; int32_t tsdbDebugFlag = 131;
int32_t cqDebugFlag = 135; int32_t cqDebugFlag = 135;
int32_t (*monitorStartSystemFp)() = NULL; int32_t (*monStartSystemFp)() = NULL;
void (*monitorStopSystemFp)() = NULL; void (*monStopSystemFp)() = NULL;
void (*monitorExecuteSQLFp)(char *sql) = NULL; void (*monExecuteSQLFp)(char *sql) = NULL;
char *qtypeStr[] = {"rpc", "fwd", "wal", "cq", "query"}; char *qtypeStr[] = {"rpc", "fwd", "wal", "cq", "query"};
...@@ -238,7 +238,7 @@ void taosSetAllDebugFlag() { ...@@ -238,7 +238,7 @@ void taosSetAllDebugFlag() {
odbcDebugFlag = debugFlag; odbcDebugFlag = debugFlag;
httpDebugFlag = debugFlag; httpDebugFlag = debugFlag;
mqttDebugFlag = debugFlag; mqttDebugFlag = debugFlag;
monitorDebugFlag = debugFlag; monDebugFlag = debugFlag;
qDebugFlag = debugFlag; qDebugFlag = debugFlag;
rpcDebugFlag = debugFlag; rpcDebugFlag = debugFlag;
uDebugFlag = debugFlag; uDebugFlag = debugFlag;
...@@ -279,15 +279,15 @@ bool taosCfgDynamicOptions(char *msg) { ...@@ -279,15 +279,15 @@ bool taosCfgDynamicOptions(char *msg) {
if (strncasecmp(cfg->option, "monitor", olen) == 0) { if (strncasecmp(cfg->option, "monitor", olen) == 0) {
if (1 == vint) { if (1 == vint) {
if (monitorStartSystemFp) { if (monStartSystemFp) {
(*monitorStartSystemFp)(); (*monStartSystemFp)();
uInfo("monitor is enabled"); uInfo("monitor is enabled");
} else { } else {
uError("monitor can't be updated, for monitor not initialized"); uError("monitor can't be updated, for monitor not initialized");
} }
} else { } else {
if (monitorStopSystemFp) { if (monStopSystemFp) {
(*monitorStopSystemFp)(); (*monStopSystemFp)();
uInfo("monitor is disabled"); uInfo("monitor is disabled");
} else { } else {
uError("monitor can't be updated, for monitor not initialized"); uError("monitor can't be updated, for monitor not initialized");
...@@ -310,8 +310,8 @@ bool taosCfgDynamicOptions(char *msg) { ...@@ -310,8 +310,8 @@ bool taosCfgDynamicOptions(char *msg) {
} }
if (strncasecmp(option, "resetQueryCache", 15) == 0) { if (strncasecmp(option, "resetQueryCache", 15) == 0) {
if (monitorExecuteSQLFp) { if (monExecuteSQLFp) {
(*monitorExecuteSQLFp)("resetQueryCache"); (*monExecuteSQLFp)("resetQueryCache");
uInfo("resetquerycache is executed"); uInfo("resetquerycache is executed");
} else { } else {
uError("resetquerycache can't be executed, for monitor not started"); uError("resetquerycache can't be executed, for monitor not started");
...@@ -1240,8 +1240,8 @@ static void doInitGlobalConfig(void) { ...@@ -1240,8 +1240,8 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE; cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg); taosInitConfigOption(cfg);
cfg.option = "monitorDebugFlag"; cfg.option = "monDebugFlag";
cfg.ptr = &monitorDebugFlag; cfg.ptr = &monDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32; cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG; cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG;
cfg.minValue = 0; cfg.minValue = 0;
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
#include "tqueue.h" #include "tqueue.h"
#include "tsync.h" #include "tsync.h"
#include "ttimer.h" #include "ttimer.h"
#include "tbalance.h" #include "tbn.h"
#include "tglobal.h" #include "tglobal.h"
#include "dnode.h" #include "dnode.h"
#include "vnode.h" #include "vnode.h"
......
...@@ -78,10 +78,10 @@ static void dnodeAllocModules() { ...@@ -78,10 +78,10 @@ static void dnodeAllocModules() {
tsModule[TSDB_MOD_MONITOR].enable = (tsEnableMonitorModule == 1); tsModule[TSDB_MOD_MONITOR].enable = (tsEnableMonitorModule == 1);
tsModule[TSDB_MOD_MONITOR].name = "monitor"; tsModule[TSDB_MOD_MONITOR].name = "monitor";
tsModule[TSDB_MOD_MONITOR].initFp = monitorInitSystem; tsModule[TSDB_MOD_MONITOR].initFp = monInitSystem;
tsModule[TSDB_MOD_MONITOR].cleanUpFp = monitorCleanUpSystem; tsModule[TSDB_MOD_MONITOR].cleanUpFp = monCleanupSystem;
tsModule[TSDB_MOD_MONITOR].startFp = monitorStartSystem; tsModule[TSDB_MOD_MONITOR].startFp = monStartSystem;
tsModule[TSDB_MOD_MONITOR].stopFp = monitorStopSystem; tsModule[TSDB_MOD_MONITOR].stopFp = monStopSystem;
if (tsEnableMonitorModule) { if (tsEnableMonitorModule) {
dnodeSetModuleStatus(TSDB_MOD_MONITOR); dnodeSetModuleStatus(TSDB_MOD_MONITOR);
} }
......
...@@ -47,13 +47,13 @@ typedef struct { ...@@ -47,13 +47,13 @@ typedef struct {
int8_t accessState; int8_t accessState;
} SAcctMonitorObj; } SAcctMonitorObj;
int32_t monitorInitSystem(); int32_t monInitSystem();
int32_t monitorStartSystem(); int32_t monStartSystem();
void monitorStopSystem(); void monStopSystem();
void monitorCleanUpSystem(); void monCleanupSystem();
void monitorSaveAcctLog(SAcctMonitorObj *pMonObj); void monSaveAcctLog(SAcctMonitorObj *pMonObj);
void monitorSaveLog(int32_t level, const char *const format, ...); void monSaveLog(int32_t level, const char *const format, ...);
void monitorExecuteSQL(char *sql); void monExecuteSQL(char *sql);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -23,14 +23,14 @@ extern "C" { ...@@ -23,14 +23,14 @@ extern "C" {
struct SVgObj; struct SVgObj;
struct SDnodeObj; struct SDnodeObj;
int32_t balanceInit(); int32_t bnInit();
void balanceCleanUp(); void bnCleanUp();
void balanceAsyncNotify(); void bnNotify();
void balanceSyncNotify(); void bnCheckModules();
void balanceReset(); void bnReset();
int32_t balanceAllocVnodes(struct SVgObj *pVgroup); int32_t bnAllocVnodes(struct SVgObj *pVgroup);
int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId); int32_t bnAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId);
int32_t balanceDropDnode(struct SDnodeObj *pDnode); int32_t bnDropDnode(struct SDnodeObj *pDnode);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -41,9 +41,9 @@ extern int32_t sdbDebugFlag; ...@@ -41,9 +41,9 @@ extern int32_t sdbDebugFlag;
#define sdbDebug(...) { if (sdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }} #define sdbDebug(...) { if (sdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
#define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }} #define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
#define mLError(...) { monitorSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) } #define mLError(...) { monSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) }
#define mLWarn(...) { monitorSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__) } #define mLWarn(...) { monSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__) }
#define mLInfo(...) { monitorSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) } #define mLInfo(...) { monSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#include "tgrant.h" #include "tgrant.h"
#include "tglobal.h" #include "tglobal.h"
#include "tname.h" #include "tname.h"
#include "tbalance.h" #include "tbn.h"
#include "tdataformat.h" #include "tdataformat.h"
#include "mnode.h" #include "mnode.h"
#include "mnodeDef.h" #include "mnodeDef.h"
...@@ -1004,7 +1004,7 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) { ...@@ -1004,7 +1004,7 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
mDebug("db:%s, all vgroups is altered", pDb->name); mDebug("db:%s, all vgroups is altered", pDb->name);
mLInfo("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg)); mLInfo("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
balanceAsyncNotify(); bnNotify();
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
...@@ -16,12 +16,12 @@ ...@@ -16,12 +16,12 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "tgrant.h" #include "tgrant.h"
#include "tbalance.h" #include "tbn.h"
#include "tglobal.h" #include "tglobal.h"
#include "tconfig.h" #include "tconfig.h"
#include "tutil.h" #include "tutil.h"
#include "tsocket.h" #include "tsocket.h"
#include "tbalance.h" #include "tbn.h"
#include "tsync.h" #include "tsync.h"
#include "tdataformat.h" #include "tdataformat.h"
#include "mnode.h" #include "mnode.h"
...@@ -115,7 +115,7 @@ static int32_t mnodeDnodeActionDelete(SSdbRow *pRow) { ...@@ -115,7 +115,7 @@ static int32_t mnodeDnodeActionDelete(SSdbRow *pRow) {
mnodeDropAllDnodeVgroups(pDnode); mnodeDropAllDnodeVgroups(pDnode);
#endif #endif
mnodeDropMnodeLocal(pDnode->dnodeId); mnodeDropMnodeLocal(pDnode->dnodeId);
balanceAsyncNotify(); bnNotify();
mnodeUpdateDnodeEps(); mnodeUpdateDnodeEps();
mDebug("dnode:%d, all vgroups is dropped from sdb", pDnode->dnodeId); mDebug("dnode:%d, all vgroups is dropped from sdb", pDnode->dnodeId);
...@@ -347,7 +347,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) { ...@@ -347,7 +347,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_INVALID_DNODE_CFG_OPTION; return TSDB_CODE_MND_INVALID_DNODE_CFG_OPTION;
} }
int32_t code = balanceAlterDnode(pDnode, vnodeId, dnodeId); int32_t code = bnAlterDnode(pDnode, vnodeId, dnodeId);
mnodeDecDnodeRef(pDnode); mnodeDecDnodeRef(pDnode);
return code; return code;
} else { } else {
...@@ -591,8 +591,8 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) { ...@@ -591,8 +591,8 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
mInfo("dnode:%d, from offline to online", pDnode->dnodeId); mInfo("dnode:%d, from offline to online", pDnode->dnodeId);
pDnode->status = TAOS_DN_STATUS_READY; pDnode->status = TAOS_DN_STATUS_READY;
pDnode->offlineReason = TAOS_DN_OFF_ONLINE; pDnode->offlineReason = TAOS_DN_OFF_ONLINE;
balanceSyncNotify(); bnCheckModules();
balanceAsyncNotify(); bnNotify();
} }
if (openVnodes != pDnode->openVnodes) { if (openVnodes != pDnode->openVnodes) {
...@@ -708,7 +708,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) { ...@@ -708,7 +708,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
#ifndef _SYNC #ifndef _SYNC
int32_t code = mnodeDropDnode(pDnode, pMsg); int32_t code = mnodeDropDnode(pDnode, pMsg);
#else #else
int32_t code = balanceDropDnode(pDnode); int32_t code = bnDropDnode(pDnode);
#endif #endif
mnodeDecDnodeRef(pDnode); mnodeDecDnodeRef(pDnode);
return code; return code;
...@@ -1182,12 +1182,12 @@ static char* mnodeGetDnodeAlternativeRoleStr(int32_t alternativeRole) { ...@@ -1182,12 +1182,12 @@ static char* mnodeGetDnodeAlternativeRoleStr(int32_t alternativeRole) {
#ifndef _SYNC #ifndef _SYNC
int32_t balanceInit() { return TSDB_CODE_SUCCESS; } int32_t bnInit() { return TSDB_CODE_SUCCESS; }
void balanceCleanUp() {} void bnCleanUp() {}
void balanceAsyncNotify() {} void bnNotify() {}
void balanceSyncNotify() {} void bnCheckModules() {}
void balanceReset() {} void bnReset() {}
int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; } int32_t bnAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; }
char* syncRole[] = { char* syncRole[] = {
"offline", "offline",
...@@ -1197,7 +1197,7 @@ char* syncRole[] = { ...@@ -1197,7 +1197,7 @@ char* syncRole[] = {
"master" "master"
}; };
int32_t balanceAllocVnodes(SVgObj *pVgroup) { int32_t bnAllocVnodes(SVgObj *pVgroup) {
void * pIter = NULL; void * pIter = NULL;
SDnodeObj *pDnode = NULL; SDnodeObj *pDnode = NULL;
SDnodeObj *pSelDnode = NULL; SDnodeObj *pSelDnode = NULL;
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
#include "os.h" #include "os.h"
#include "taosdef.h" #include "taosdef.h"
#include "tsched.h" #include "tsched.h"
#include "tbalance.h" #include "tbn.h"
#include "tgrant.h" #include "tgrant.h"
#include "ttimer.h" #include "ttimer.h"
#include "tglobal.h" #include "tglobal.h"
...@@ -58,7 +58,7 @@ static const SMnodeComponent tsMnodeComponents[] = { ...@@ -58,7 +58,7 @@ static const SMnodeComponent tsMnodeComponents[] = {
{"tables", mnodeInitTables, mnodeCleanupTables}, {"tables", mnodeInitTables, mnodeCleanupTables},
{"mnodes", mnodeInitMnodes, mnodeCleanupMnodes}, {"mnodes", mnodeInitMnodes, mnodeCleanupMnodes},
{"sdb", sdbInit, sdbCleanUp}, {"sdb", sdbInit, sdbCleanUp},
{"balance", balanceInit, balanceCleanUp}, {"balance", bnInit, bnCleanUp},
{"grant", grantInit, grantCleanUp}, {"grant", grantInit, grantCleanUp},
{"show", mnodeInitShow, mnodeCleanUpShow} {"show", mnodeInitShow, mnodeCleanUpShow}
}; };
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#include "tglobal.h" #include "tglobal.h"
#include "trpc.h" #include "trpc.h"
#include "tsync.h" #include "tsync.h"
#include "tbalance.h" #include "tbn.h"
#include "tutil.h" #include "tutil.h"
#include "tsocket.h" #include "tsocket.h"
#include "tdataformat.h" #include "tdataformat.h"
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#include "tsystem.h" #include "tsystem.h"
#include "tutil.h" #include "tutil.h"
#include "tgrant.h" #include "tgrant.h"
#include "tbalance.h" #include "tbn.h"
#include "tglobal.h" #include "tglobal.h"
#include "mnode.h" #include "mnode.h"
#include "dnode.h" #include "dnode.h"
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
#include "os.h" #include "os.h"
#include "taosdef.h" #include "taosdef.h"
#include "tsched.h" #include "tsched.h"
#include "tbalance.h" #include "tbn.h"
#include "tgrant.h" #include "tgrant.h"
#include "ttimer.h" #include "ttimer.h"
#include "tglobal.h" #include "tglobal.h"
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#include "hash.h" #include "hash.h"
#include "tutil.h" #include "tutil.h"
#include "tref.h" #include "tref.h"
#include "tbalance.h" #include "tbn.h"
#include "tqueue.h" #include "tqueue.h"
#include "twal.h" #include "twal.h"
#include "tsync.h" #include "tsync.h"
...@@ -244,7 +244,7 @@ static void sdbNotifyRole(int32_t vgId, int8_t role) { ...@@ -244,7 +244,7 @@ static void sdbNotifyRole(int32_t vgId, int8_t role) {
sdbInfo("vgId:1, mnode role changed from %s to %s", syncRole[tsSdbMgmt.role], syncRole[role]); sdbInfo("vgId:1, mnode role changed from %s to %s", syncRole[tsSdbMgmt.role], syncRole[role]);
if (role == TAOS_SYNC_ROLE_MASTER && tsSdbMgmt.role != TAOS_SYNC_ROLE_MASTER) { if (role == TAOS_SYNC_ROLE_MASTER && tsSdbMgmt.role != TAOS_SYNC_ROLE_MASTER) {
balanceReset(); bnReset();
} }
tsSdbMgmt.role = role; tsSdbMgmt.role = role;
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#include "tsocket.h" #include "tsocket.h"
#include "tidpool.h" #include "tidpool.h"
#include "tsync.h" #include "tsync.h"
#include "tbalance.h" #include "tbn.h"
#include "tglobal.h" #include "tglobal.h"
#include "tdataformat.h" #include "tdataformat.h"
#include "dnode.h" #include "dnode.h"
...@@ -563,7 +563,7 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg) { ...@@ -563,7 +563,7 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg) {
pVgroup->numOfVnodes = pDb->cfg.replications; pVgroup->numOfVnodes = pDb->cfg.replications;
pVgroup->createdTime = taosGetTimestampMs(); pVgroup->createdTime = taosGetTimestampMs();
pVgroup->accessState = TSDB_VN_ALL_ACCCESS; pVgroup->accessState = TSDB_VN_ALL_ACCCESS;
int32_t code = balanceAllocVnodes(pVgroup); int32_t code = bnAllocVnodes(pVgroup);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mError("db:%s, no enough dnode to alloc %d vnodes to vgroup, reason:%s", pDb->name, pVgroup->numOfVnodes, mError("db:%s, no enough dnode to alloc %d vnodes to vgroup, reason:%s", pDb->name, pVgroup->numOfVnodes,
tstrerror(code)); tstrerror(code));
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
#include "os.h" #include "os.h"
#include "taosdef.h" #include "taosdef.h"
#include "tsched.h" #include "tsched.h"
#include "tbalance.h" #include "tbn.h"
#include "tgrant.h" #include "tgrant.h"
#include "tglobal.h" #include "tglobal.h"
#include "trpc.h" #include "trpc.h"
......
...@@ -27,12 +27,12 @@ ...@@ -27,12 +27,12 @@
#include "monitor.h" #include "monitor.h"
#include "taoserror.h" #include "taoserror.h"
#define mnFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }} #define monFatal(...) { if (monDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
#define mnError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }} #define monError(...) { if (monDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
#define mnWarn(...) { if (monitorDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }} #define monWarn(...) { if (monDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }}
#define mnInfo(...) { if (monitorDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }} #define monInfo(...) { if (monDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }}
#define mnDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} #define monDebug(...) { if (monDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monDebugFlag, __VA_ARGS__); }}
#define mnTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} #define monTrace(...) { if (monDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monDebugFlag, __VA_ARGS__); }}
#define SQL_LENGTH 1030 #define SQL_LENGTH 1030
#define LOG_LEN_STR 100 #define LOG_LEN_STR 100
...@@ -48,12 +48,12 @@ typedef enum { ...@@ -48,12 +48,12 @@ typedef enum {
MON_CMD_CREATE_TB_ACCT_ROOT, MON_CMD_CREATE_TB_ACCT_ROOT,
MON_CMD_CREATE_TB_SLOWQUERY, MON_CMD_CREATE_TB_SLOWQUERY,
MON_CMD_MAX MON_CMD_MAX
} EMonitorCommand; } EMonCmd;
typedef enum { typedef enum {
MON_STATE_NOT_INIT, MON_STATE_NOT_INIT,
MON_STATE_INITED MON_STATE_INITED
} EMonitorState; } EMonState;
typedef struct { typedef struct {
pthread_t thread; pthread_t thread;
...@@ -64,17 +64,17 @@ typedef struct { ...@@ -64,17 +64,17 @@ typedef struct {
int8_t start; // enable/disable by mnode int8_t start; // enable/disable by mnode
int8_t quiting; // taosd is quiting int8_t quiting; // taosd is quiting
char sql[SQL_LENGTH + 1]; char sql[SQL_LENGTH + 1];
} SMonitorConn; } SMonConn;
static SMonitorConn tsMonitor = {0}; static SMonConn tsMonitor = {0};
static void monitorSaveSystemInfo(); static void monSaveSystemInfo();
static void *monitorThreadFunc(void *param); static void *monThreadFunc(void *param);
static void monitorBuildMonitorSql(char *sql, int32_t cmd); static void monBuildMonitorSql(char *sql, int32_t cmd);
extern int32_t (*monitorStartSystemFp)(); extern int32_t (*monStartSystemFp)();
extern void (*monitorStopSystemFp)(); extern void (*monStopSystemFp)();
extern void (*monitorExecuteSQLFp)(char *sql); extern void (*monExecuteSQLFp)(char *sql);
int32_t monitorInitSystem() { int32_t monInitSystem() {
if (tsMonitor.ep[0] == 0) { if (tsMonitor.ep[0] == 0) {
strcpy(tsMonitor.ep, tsLocalEp); strcpy(tsMonitor.ep, tsLocalEp);
} }
...@@ -90,29 +90,29 @@ int32_t monitorInitSystem() { ...@@ -90,29 +90,29 @@ int32_t monitorInitSystem() {
pthread_attr_init(&thAttr); pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) { if (pthread_create(&tsMonitor.thread, &thAttr, monThreadFunc, NULL)) {
mnError("failed to create thread to for monitor module, reason:%s", strerror(errno)); monError("failed to create thread to for monitor module, reason:%s", strerror(errno));
return -1; return -1;
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
mnDebug("monitor thread is launched"); monDebug("monitor thread is launched");
monitorStartSystemFp = monitorStartSystem; monStartSystemFp = monStartSystem;
monitorStopSystemFp = monitorStopSystem; monStopSystemFp = monStopSystem;
return 0; return 0;
} }
int32_t monitorStartSystem() { int32_t monStartSystem() {
taos_init(); taos_init();
tsMonitor.start = 1; tsMonitor.start = 1;
monitorExecuteSQLFp = monitorExecuteSQL; monExecuteSQLFp = monExecuteSQL;
mnInfo("monitor module start"); monInfo("monitor module start");
return 0; return 0;
} }
static void *monitorThreadFunc(void *param) { static void *monThreadFunc(void *param) {
mnDebug("starting to initialize monitor module ..."); monDebug("starting to initialize monitor module ...");
while (1) { while (1) {
static int32_t accessTimes = 0; static int32_t accessTimes = 0;
...@@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) { ...@@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) {
if (tsMonitor.quiting) { if (tsMonitor.quiting) {
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
mnInfo("monitor thread will quit, for taosd is quiting"); monInfo("monitor thread will quit, for taosd is quiting");
break; break;
} else { } else {
taosGetDisk(); taosGetDisk();
...@@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) { ...@@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) {
} }
if (dnodeGetDnodeId() <= 0) { if (dnodeGetDnodeId() <= 0) {
mnDebug("dnode not initialized, waiting for 3000 ms to start monitor module"); monDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
continue; continue;
} }
...@@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) { ...@@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) {
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0); tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0);
if (tsMonitor.conn == NULL) { if (tsMonitor.conn == NULL) {
mnError("failed to connect to database, reason:%s", tstrerror(terrno)); monError("failed to connect to database, reason:%s", tstrerror(terrno));
continue; continue;
} else { } else {
mnDebug("connect to database success"); monDebug("connect to database success");
} }
} }
...@@ -151,16 +151,16 @@ static void *monitorThreadFunc(void *param) { ...@@ -151,16 +151,16 @@ static void *monitorThreadFunc(void *param) {
int code = 0; int code = 0;
for (; tsMonitor.cmdIndex < MON_CMD_MAX; ++tsMonitor.cmdIndex) { for (; tsMonitor.cmdIndex < MON_CMD_MAX; ++tsMonitor.cmdIndex) {
monitorBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex); monBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex);
void *res = taos_query(tsMonitor.conn, tsMonitor.sql); void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
code = taos_errno(res); code = taos_errno(res);
taos_free_result(res); taos_free_result(res);
if (code != 0) { if (code != 0) {
mnError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code)); monError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
break; break;
} else { } else {
mnDebug("successfully to exec sql:%s", tsMonitor.sql); monDebug("successfully to exec sql:%s", tsMonitor.sql);
} }
} }
...@@ -171,16 +171,16 @@ static void *monitorThreadFunc(void *param) { ...@@ -171,16 +171,16 @@ static void *monitorThreadFunc(void *param) {
if (tsMonitor.state == MON_STATE_INITED) { if (tsMonitor.state == MON_STATE_INITED) {
if (accessTimes % tsMonitorInterval == 0) { if (accessTimes % tsMonitorInterval == 0) {
monitorSaveSystemInfo(); monSaveSystemInfo();
} }
} }
} }
mnInfo("monitor thread is stopped"); monInfo("monitor thread is stopped");
return NULL; return NULL;
} }
static void monitorBuildMonitorSql(char *sql, int32_t cmd) { static void monBuildMonitorSql(char *sql, int32_t cmd) {
memset(sql, 0, SQL_LENGTH); memset(sql, 0, SQL_LENGTH);
if (cmd == MON_CMD_CREATE_DB) { if (cmd == MON_CMD_CREATE_DB) {
...@@ -236,47 +236,47 @@ static void monitorBuildMonitorSql(char *sql, int32_t cmd) { ...@@ -236,47 +236,47 @@ static void monitorBuildMonitorSql(char *sql, int32_t cmd) {
sql[SQL_LENGTH] = 0; sql[SQL_LENGTH] = 0;
} }
void monitorStopSystem() { void monStopSystem() {
tsMonitor.start = 0; tsMonitor.start = 0;
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
monitorExecuteSQLFp = NULL; monExecuteSQLFp = NULL;
mnInfo("monitor module stopped"); monInfo("monitor module stopped");
} }
void monitorCleanUpSystem() { void monCleanupSystem() {
tsMonitor.quiting = 1; tsMonitor.quiting = 1;
monitorStopSystem(); monStopSystem();
pthread_join(tsMonitor.thread, NULL); pthread_join(tsMonitor.thread, NULL);
if (tsMonitor.conn != NULL) { if (tsMonitor.conn != NULL) {
taos_close(tsMonitor.conn); taos_close(tsMonitor.conn);
tsMonitor.conn = NULL; tsMonitor.conn = NULL;
} }
mnInfo("monitor module is cleaned up"); monInfo("monitor module is cleaned up");
} }
// unit is MB // unit is MB
static int32_t monitorBuildMemorySql(char *sql) { static int32_t monBuildMemorySql(char *sql) {
float sysMemoryUsedMB = 0; float sysMemoryUsedMB = 0;
bool suc = taosGetSysMemory(&sysMemoryUsedMB); bool suc = taosGetSysMemory(&sysMemoryUsedMB);
if (!suc) { if (!suc) {
mnDebug("failed to get sys memory info"); monDebug("failed to get sys memory info");
} }
float procMemoryUsedMB = 0; float procMemoryUsedMB = 0;
suc = taosGetProcMemory(&procMemoryUsedMB); suc = taosGetProcMemory(&procMemoryUsedMB);
if (!suc) { if (!suc) {
mnDebug("failed to get proc memory info"); monDebug("failed to get proc memory info");
} }
return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB); return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
} }
// unit is % // unit is %
static int32_t monitorBuildCpuSql(char *sql) { static int32_t monBuildCpuSql(char *sql) {
float sysCpuUsage = 0, procCpuUsage = 0; float sysCpuUsage = 0, procCpuUsage = 0;
bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage); bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
if (!suc) { if (!suc) {
mnDebug("failed to get cpu usage"); monDebug("failed to get cpu usage");
} }
if (sysCpuUsage <= procCpuUsage) { if (sysCpuUsage <= procCpuUsage) {
...@@ -287,72 +287,72 @@ static int32_t monitorBuildCpuSql(char *sql) { ...@@ -287,72 +287,72 @@ static int32_t monitorBuildCpuSql(char *sql) {
} }
// unit is GB // unit is GB
static int32_t monitorBuildDiskSql(char *sql) { static int32_t monBuildDiskSql(char *sql) {
return sprintf(sql, ", %f, %d", (tsTotalDataDirGB - tsAvailDataDirGB), (int32_t)tsTotalDataDirGB); return sprintf(sql, ", %f, %d", (tsTotalDataDirGB - tsAvailDataDirGB), (int32_t)tsTotalDataDirGB);
} }
// unit is Kb // unit is Kb
static int32_t monitorBuildBandSql(char *sql) { static int32_t monBuildBandSql(char *sql) {
float bandSpeedKb = 0; float bandSpeedKb = 0;
bool suc = taosGetBandSpeed(&bandSpeedKb); bool suc = taosGetBandSpeed(&bandSpeedKb);
if (!suc) { if (!suc) {
mnDebug("failed to get bandwidth speed"); monDebug("failed to get bandwidth speed");
} }
return sprintf(sql, ", %f", bandSpeedKb); return sprintf(sql, ", %f", bandSpeedKb);
} }
static int32_t monitorBuildReqSql(char *sql) { static int32_t monBuildReqSql(char *sql) {
SStatisInfo info = dnodeGetStatisInfo(); SStatisInfo info = dnodeGetStatisInfo();
return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum); return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
} }
static int32_t monitorBuildIoSql(char *sql) { static int32_t monBuildIoSql(char *sql) {
float readKB = 0, writeKB = 0; float readKB = 0, writeKB = 0;
bool suc = taosGetProcIO(&readKB, &writeKB); bool suc = taosGetProcIO(&readKB, &writeKB);
if (!suc) { if (!suc) {
mnDebug("failed to get io info"); monDebug("failed to get io info");
} }
return sprintf(sql, ", %f, %f", readKB, writeKB); return sprintf(sql, ", %f, %f", readKB, writeKB);
} }
static void monitorSaveSystemInfo() { static void monSaveSystemInfo() {
int64_t ts = taosGetTimestampUs(); int64_t ts = taosGetTimestampUs();
char * sql = tsMonitor.sql; char * sql = tsMonitor.sql;
int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn%d values(%" PRId64, tsMonitorDbName, dnodeGetDnodeId(), ts); int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn%d values(%" PRId64, tsMonitorDbName, dnodeGetDnodeId(), ts);
pos += monitorBuildCpuSql(sql + pos); pos += monBuildCpuSql(sql + pos);
pos += monitorBuildMemorySql(sql + pos); pos += monBuildMemorySql(sql + pos);
pos += monitorBuildDiskSql(sql + pos); pos += monBuildDiskSql(sql + pos);
pos += monitorBuildBandSql(sql + pos); pos += monBuildBandSql(sql + pos);
pos += monitorBuildIoSql(sql + pos); pos += monBuildIoSql(sql + pos);
pos += monitorBuildReqSql(sql + pos); pos += monBuildReqSql(sql + pos);
void *res = taos_query(tsMonitor.conn, tsMonitor.sql); void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
int code = taos_errno(res); int code = taos_errno(res);
taos_free_result(res); taos_free_result(res);
if (code != 0) { if (code != 0) {
mnError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql); monError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
} else { } else {
mnDebug("successfully to save system info, sql:%s", tsMonitor.sql); monDebug("successfully to save system info, sql:%s", tsMonitor.sql);
} }
} }
static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) { static void monExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
int32_t c = taos_errno(result); int32_t c = taos_errno(result);
if (c != TSDB_CODE_SUCCESS) { if (c != TSDB_CODE_SUCCESS) {
mnError("save %s failed, reason:%s", (char *)param, tstrerror(c)); monError("save %s failed, reason:%s", (char *)param, tstrerror(c));
} else { } else {
int32_t rows = taos_affected_rows(result); int32_t rows = taos_affected_rows(result);
mnDebug("save %s succ, rows:%d", (char *)param, rows); monDebug("save %s succ, rows:%d", (char *)param, rows);
} }
taos_free_result(result); taos_free_result(result);
} }
void monitorSaveAcctLog(SAcctMonitorObj *pMon) { void monSaveAcctLog(SAcctMonitorObj *pMon) {
if (tsMonitor.state != MON_STATE_INITED) return; if (tsMonitor.state != MON_STATE_INITED) return;
char sql[1024] = {0}; char sql[1024] = {0};
...@@ -382,11 +382,11 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) { ...@@ -382,11 +382,11 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
pMon->totalConns, pMon->maxConns, pMon->totalConns, pMon->maxConns,
pMon->accessState); pMon->accessState);
mnDebug("save account info, sql:%s", sql); monDebug("save account info, sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info"); taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "account info");
} }
void monitorSaveLog(int32_t level, const char *const format, ...) { void monSaveLog(int32_t level, const char *const format, ...) {
if (tsMonitor.state != MON_STATE_INITED) return; if (tsMonitor.state != MON_STATE_INITED) return;
va_list argpointer; va_list argpointer;
...@@ -403,13 +403,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) { ...@@ -403,13 +403,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
len += sprintf(sql + len, "', '%s')", tsLocalEp); len += sprintf(sql + len, "', '%s')", tsLocalEp);
sql[len++] = 0; sql[len++] = 0;
mnDebug("save log, sql: %s", sql); monDebug("save log, sql: %s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log"); taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "log");
} }
void monitorExecuteSQL(char *sql) { void monExecuteSQL(char *sql) {
if (tsMonitor.state != MON_STATE_INITED) return; if (tsMonitor.state != MON_STATE_INITED) return;
mnDebug("execute sql:%s", sql); monDebug("execute sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql"); taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "sql");
} }
...@@ -24,7 +24,7 @@ sql alter dnode 1 debugFlag 135 ...@@ -24,7 +24,7 @@ sql alter dnode 1 debugFlag 135
sql alter dnode 1 debugFlag 131 sql alter dnode 1 debugFlag 131
sql alter dnode 1 monitor 0 sql alter dnode 1 monitor 0
sql alter dnode 1 debugFlag 135 sql alter dnode 1 debugFlag 135
sql alter dnode 1 monitorDebugFlag 135 sql alter dnode 1 monDebugFlag 135
sql alter dnode 1 vDebugFlag 135 sql alter dnode 1 vDebugFlag 135
sql alter dnode 1 mDebugFlag 135 sql alter dnode 1 mDebugFlag 135
sql alter dnode 1 cDebugFlag 135 sql alter dnode 1 cDebugFlag 135
...@@ -44,15 +44,15 @@ sql_error alter dnode 2 tmrDebugFlag 135 ...@@ -44,15 +44,15 @@ sql_error alter dnode 2 tmrDebugFlag 135
print ======== step3 print ======== step3
sql_error alter $hostname1 debugFlag 135 sql_error alter $hostname1 debugFlag 135
sql_error alter $hostname1 monitorDebugFlag 135 sql_error alter $hostname1 monDebugFlag 135
sql_error alter $hostname1 vDebugFlag 135 sql_error alter $hostname1 vDebugFlag 135
sql_error alter $hostname1 mDebugFlag 135 sql_error alter $hostname1 mDebugFlag 135
sql_error alter dnode $hostname2 debugFlag 135 sql_error alter dnode $hostname2 debugFlag 135
sql_error alter dnode $hostname2 monitorDebugFlag 135 sql_error alter dnode $hostname2 monDebugFlag 135
sql_error alter dnode $hostname2 vDebugFlag 135 sql_error alter dnode $hostname2 vDebugFlag 135
sql_error alter dnode $hostname2 mDebugFlag 135 sql_error alter dnode $hostname2 mDebugFlag 135
sql alter dnode $hostname1 debugFlag 135 sql alter dnode $hostname1 debugFlag 135
sql alter dnode $hostname1 monitorDebugFlag 135 sql alter dnode $hostname1 monDebugFlag 135
sql alter dnode $hostname1 vDebugFlag 135 sql alter dnode $hostname1 vDebugFlag 135
sql alter dnode $hostname1 tmrDebugFlag 131 sql alter dnode $hostname1 tmrDebugFlag 131
......
...@@ -120,7 +120,7 @@ echo "cDebugFlag 143" >> $TAOS_CFG ...@@ -120,7 +120,7 @@ echo "cDebugFlag 143" >> $TAOS_CFG
echo "jnidebugFlag 143" >> $TAOS_CFG echo "jnidebugFlag 143" >> $TAOS_CFG
echo "odbcdebugFlag 143" >> $TAOS_CFG echo "odbcdebugFlag 143" >> $TAOS_CFG
echo "httpDebugFlag 143" >> $TAOS_CFG echo "httpDebugFlag 143" >> $TAOS_CFG
echo "monitorDebugFlag 143" >> $TAOS_CFG echo "monDebugFlag 143" >> $TAOS_CFG
echo "mqttDebugFlag 143" >> $TAOS_CFG echo "mqttDebugFlag 143" >> $TAOS_CFG
echo "qdebugFlag 143" >> $TAOS_CFG echo "qdebugFlag 143" >> $TAOS_CFG
echo "rpcDebugFlag 143" >> $TAOS_CFG echo "rpcDebugFlag 143" >> $TAOS_CFG
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册