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

rename some funciton pointors

This commit is contained in:
slguan 2020-03-02 22:58:11 +08:00
parent 62d79763de
commit 14ecbb0fe7
25 changed files with 672 additions and 362 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/dnode/inc/dnodeSystem.h
View File

@ -38,6 +38,7 @@ extern int32_t tsMaxQueues;
extern void ** tsRpcQhandle; extern void ** tsRpcQhandle;
extern void *tsQueryQhandle; extern void *tsQueryQhandle;
extern void *tsDnodeMgmtQhandle; extern void *tsDnodeMgmtQhandle;
extern void *tsDnodeTmr;
int32_t dnodeInitSystem(); int32_t dnodeInitSystem();
void dnodeCleanUpSystem(); void dnodeCleanUpSystem();

2
src/dnode/inc/dnodeVnodeMgmt.h
View File

@ -58,6 +58,8 @@ int32_t dnodeDropVnode(int32_t vnode);
//tsdb_repo_t* dnodeGetVnode(int vid); //tsdb_repo_t* dnodeGetVnode(int vid);
void* dnodeGetVnode(int32_t vnode); void* dnodeGetVnode(int32_t vnode);
int32_t dnodeGetVnodesNum();
/* /*
* get the status of vnode * get the status of vnode
*/ */

74
src/dnode/src/dnodeMgmt.c
View File

@ -29,10 +29,14 @@
void (*dnodeInitMgmtIpFp)() = NULL; void (*dnodeInitMgmtIpFp)() = NULL;
int32_t (*dnodeInitMgmtFp)() = NULL; int32_t (*dnodeInitMgmtFp)() = NULL;
void (*dnodeCleanUpMgmtFp)() = NULL;
void (*dnodeProcessStatusRspFp)(int8_t *pCont, int32_t contLen, int8_t msgType, void *pConn) = NULL; void (*dnodeProcessStatusRspFp)(int8_t *pCont, int32_t contLen, int8_t msgType, void *pConn) = NULL;
void (*dnodeSendMsgToMnodeFp)(int8_t msgType, void *pCont, int32_t contLen) = NULL; void (*dnodeSendMsgToMnodeFp)(int8_t msgType, void *pCont, int32_t contLen) = NULL;
void (*dnodeSendRspToMnodeFp)(void *handle, int32_t code, void *pCont, int contLen) = NULL; void (*dnodeSendRspToMnodeFp)(void *handle, int32_t code, void *pCont, int contLen) = NULL;
static void *tsStatusTimer = NULL;
static void (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(void *pCont, int32_t contLen, int8_t msgType, void *pConn); static void (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(void *pCont, int32_t contLen, int8_t msgType, void *pConn);
static void dnodeInitProcessShellMsg(); static void dnodeInitProcessShellMsg();
@ -86,12 +90,71 @@ void dnodeSendRspToMnode(void *pConn, int8_t msgType, int32_t code, void *pCont,
} }
} }
void dnodeSendStatusMsgToMgmt(void *handle, void *tmrId) {
taosTmrReset(dnodeSendStatusMsgToMgmt, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
if (tsStatusTimer == NULL) {
dError("Failed to start status timer");
return;
}
int32_t contLen = sizeof(SStatusMsg) + dnodeGetVnodesNum() * sizeof(SVnodeLoad);
SStatusMsg *pStatus = rpcMallocCont(contLen);
if (pStatus == NULL) {
dError("Failed to malloc status message");
return;
}
int32_t totalVnodes = dnodeGetVnodesNum();
pStatus->version = htonl(tsVersion);
pStatus->privateIp = htonl(inet_addr(tsPrivateIp));
pStatus->publicIp = htonl(inet_addr(tsPublicIp));
pStatus->lastReboot = htonl(tsRebootTime);
pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
pStatus->openVnodes = htons((uint16_t) totalVnodes);
pStatus->numOfCores = htons((uint16_t) tsNumOfCores);
pStatus->diskAvailable = tsAvailDataDirGB;
pStatus->alternativeRole = (uint8_t) tsAlternativeRole;
SVnodeLoad *pLoad = (SVnodeLoad *)pStatus->load;
//TODO loop all vnodes
// for (int32_t vnode = 0, count = 0; vnode <= totalVnodes; ++vnode) {
// if (vnodeList[vnode].cfg.maxSessions <= 0) continue;
//
// SVnodeObj *pVnode = vnodeList + vnode;
// pLoad->vnode = htonl(vnode);
// pLoad->vgId = htonl(pVnode->cfg.vgId);
// pLoad->status = (uint8_t)vnodeList[vnode].vnodeStatus;
// pLoad->syncStatus =(uint8_t)vnodeList[vnode].syncStatus;
// pLoad->accessState = (uint8_t)(pVnode->accessState);
// pLoad->totalStorage = htobe64(pVnode->vnodeStatistic.totalStorage);
// pLoad->compStorage = htobe64(pVnode->vnodeStatistic.compStorage);
// if (pVnode->vnodeStatus == TSDB_VN_STATUS_MASTER) {
// pLoad->pointsWritten = htobe64(pVnode->vnodeStatistic.pointsWritten);
// } else {
// pLoad->pointsWritten = htobe64(0);
// }
// pLoad++;
//
// if (++count >= tsOpenVnodes) {
// break;
// }
// }
dnodeSendMsgToMnode(TSDB_MSG_TYPE_STATUS, pStatus, contLen);
//grantSendMsgToMgmt();
}
int32_t dnodeInitMgmt() { int32_t dnodeInitMgmt() {
if (dnodeInitMgmtFp) { if (dnodeInitMgmtFp) {
dnodeInitMgmtFp(); dnodeInitMgmtFp();
} }
dnodeInitProcessShellMsg(); dnodeInitProcessShellMsg();
taosTmrReset(dnodeSendStatusMsgToMgmt, 500, NULL, tsDnodeTmr, &tsStatusTimer);
return 0; return 0;
} }
@ -101,6 +164,17 @@ void dnodeInitMgmtIp() {
} }
} }
void dnodeCleanUpMgmt() {
if (tsStatusTimer != NULL) {
taosTmrStopA(&tsStatusTimer);
tsStatusTimer = NULL;
}
if (dnodeCleanUpMgmtFp) {
dnodeCleanUpMgmtFp();
}
}
void dnodeProcessMsgFromMgmt(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) { void dnodeProcessMsgFromMgmt(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) { if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) {
dError("invalid msg type:%d", msgType); dError("invalid msg type:%d", msgType);

26
src/dnode/src/dnodeSystem.c
View File

@ -33,12 +33,12 @@
#include "dnodeVnodeMgmt.h" #include "dnodeVnodeMgmt.h"
#ifdef CLUSTER #ifdef CLUSTER
#include "acct.h" //#include "acct.h"
#include "admin.h" //#include "admin.h"
#include "cluster.h" //#include "cluster.h"
#include "grant.h" //#include "grant.h"
#include "replica.h" //#include "replica.h"
#include "storage.h" //#include "storage.h"
#endif #endif
static pthread_mutex_t tsDnodeMutex; static pthread_mutex_t tsDnodeMutex;
@ -48,8 +48,7 @@ static int32_t dnodeInitRpcQHandle();
static int32_t dnodeInitQueryQHandle(); static int32_t dnodeInitQueryQHandle();
static int32_t dnodeInitTmrCtl(); static int32_t dnodeInitTmrCtl();
void *tsStatusTimer = NULL; void *tsDnodeTmr;
void *vnodeTmrCtrl;
void **tsRpcQhandle; void **tsRpcQhandle;
void *tsDnodeMgmtQhandle; void *tsDnodeMgmtQhandle;
void *tsQueryQhandle; void *tsQueryQhandle;
@ -90,10 +89,7 @@ void dnodeCleanUpSystem() {
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED); dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED);
} }
if (tsStatusTimer != NULL) {
taosTmrStopA(&tsStatusTimer);
tsStatusTimer = NULL;
}
dnodeCleanupShell(); dnodeCleanupShell();
dnodeCleanUpModules(); dnodeCleanUpModules();
@ -259,15 +255,15 @@ static int32_t dnodeInitQueryQHandle() {
int32_t maxQueueSize = tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode; int32_t maxQueueSize = tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode;
dTrace("query task queue initialized, max slot:%d, task threads:%d", maxQueueSize, numOfThreads); dTrace("query task queue initialized, max slot:%d, task threads:%d", maxQueueSize, numOfThreads);
tsQueryQhandle = taosInitSchedulerWithInfo(maxQueueSize, numOfThreads, "query", vnodeTmrCtrl); tsQueryQhandle = taosInitSchedulerWithInfo(maxQueueSize, numOfThreads, "query", tsDnodeTmr);
return 0; return 0;
} }
static int32_t dnodeInitTmrCtl() { static int32_t dnodeInitTmrCtl() {
vnodeTmrCtrl = taosTmrInit(TSDB_MAX_VNODES * (tsVnodePeers + 10) + tsSessionsPerVnode + 1000, 200, 60000, tsDnodeTmr = taosTmrInit(TSDB_MAX_VNODES * (tsVnodePeers + 10) + tsSessionsPerVnode + 1000, 200, 60000,
"DND-vnode"); "DND-vnode");
if (vnodeTmrCtrl == NULL) { if (tsDnodeTmr == NULL) {
dError("failed to init timer, exit"); dError("failed to init timer, exit");
return -1; return -1;
} }

5
src/inc/dnode.h
View File

@ -48,17 +48,14 @@ extern int32_t (*dnodeCheckSystem)();
// dnodeSystem // dnodeSystem
extern void *tsDnodeMgmtQhandle; extern void *tsDnodeMgmtQhandle;
void dnodeCheckDataDirOpenned(const char* dir);
void dnodeProcessMsgFromMgmt(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code); void dnodeProcessMsgFromMgmt(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
// dnodeModule // dnodeModule
extern void (*dnodeStartModules)(); extern void (*dnodeStartModules)();
// multilevelStorage
extern int32_t (*dnodeInitStorage)();
extern void (*dnodeCleanupStorage)();
void dnodeCheckDataDirOpenned(const char* dir);
void dnodeLockVnodes(); void dnodeLockVnodes();
void dnodeUnLockVnodes(); void dnodeUnLockVnodes();

4
src/inc/mnode.h
View File

@ -260,12 +260,10 @@ typedef struct {
//mgmtSystem //mgmtSystem
int32_t mgmtStartSystem(); int32_t mgmtStartSystem();
void mgmtCleanUpSystem(); void mgmtCleanUpSystem();
void mgmtProcessMsgFromDnode(char msgType, void *pCont, int contLen, void *pConn, int32_t code);
extern int32_t (*mgmtInitSystem)();
extern void (*mgmtStopSystem)();
extern void (*mgmtCleanUpRedirect)(); extern void (*mgmtCleanUpRedirect)();
void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
src/inc/taoserror.h
View File

@ -163,6 +163,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TABLE_ID_MISMATCH, 0, 118, "table id mismat
TAOS_DEFINE_ERROR(TSDB_CODE_QUERY_CACHE_ERASED, 0, 119, "query cache erased") TAOS_DEFINE_ERROR(TSDB_CODE_QUERY_CACHE_ERASED, 0, 119, "query cache erased")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_MSG, 0, 120, "invalid message") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_MSG, 0, 120, "invalid message")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_TABLE_TYPE, 0, 121, "invalid table typee") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_TABLE_TYPE, 0, 121, "invalid table typee")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_MSG_VERSION, 0, 122, "invalid version of message")
TAOS_DEFINE_ERROR(TSDB_CODE_DNODE_NOT_EXIST, 0, 123, "dnode not exist")
#ifdef TAOS_ERROR_C #ifdef TAOS_ERROR_C
}; };

42
src/inc/taosmsg.h
View File

@ -572,26 +572,6 @@ typedef struct {
char reserved[64]; char reserved[64];
} SVnodeStatisticInfo; } SVnodeStatisticInfo;
typedef struct {
uint32_t version;
uint32_t publicIp;
uint32_t lastReboot; // time stamp for last reboot
uint16_t numOfCores;
uint8_t alternativeRole;
uint8_t reserve;
uint16_t numOfTotalVnodes; // from config file
uint16_t unused;
float diskAvailable; // GB
uint32_t openVnodes;
char reserved[16];
SVnodeLoad load[];
} SStatusMsg;
typedef struct {
int32_t code;
SRpcIpSet ipList;
} SStatusRsp;
typedef struct { typedef struct {
uint32_t moduleStatus; uint32_t moduleStatus;
uint32_t createdTime; uint32_t createdTime;
@ -599,6 +579,28 @@ typedef struct {
uint32_t reserved; uint32_t reserved;
} SDnodeState; } SDnodeState;
typedef struct {
uint32_t version;
uint32_t privateIp;
uint32_t publicIp;
uint32_t lastReboot; // time stamp for last reboot
uint16_t numOfTotalVnodes; // from config file
uint16_t openVnodes;
uint16_t numOfCores;
float diskAvailable; // GB
uint8_t alternativeRole;
uint8_t reserve[15];
SVnodeLoad load[];
} SStatusMsg;
typedef struct {
int32_t code;
int32_t numOfVnodes;
SDnodeState dnodeState;
SRpcIpSet ipList;
SVnodeAccess vnodeAccess[];
} SStatusRsp;
// internal message // internal message
typedef struct { typedef struct {
uint32_t destId; uint32_t destId;

19
src/mnode/inc/mgmtAcct.h
View File

@ -27,14 +27,17 @@ int32_t mgmtRemoveDbFromAcct(SAcctObj *pAcct, SDbObj *pDb);
int32_t mgmtAddUserIntoAcct(SAcctObj *pAcct, SUserObj *pUser); int32_t mgmtAddUserIntoAcct(SAcctObj *pAcct, SUserObj *pUser);
int32_t mgmtRemoveUserFromAcct(SAcctObj *pAcct, SUserObj *pUser); int32_t mgmtRemoveUserFromAcct(SAcctObj *pAcct, SUserObj *pUser);
extern int32_t (*mgmtInitAccts)(); int32_t mgmtInitAccts();
extern void (*mgmtCleanUpAccts)(); void mgmtCleanUpAccts();
extern SAcctObj* (*mgmtGetAcct)(char *acctName); SAcctObj* mgmtGetAcct(char *acctName);
extern int32_t (*mgmtCheckUserLimit)(SAcctObj *pAcct);
extern int32_t (*mgmtCheckDbLimit)(SAcctObj *pAcct); int32_t mgmtCheckUserLimit(SAcctObj *pAcct);
extern int32_t (*mgmtCheckTableLimit)(SAcctObj *pAcct, SCreateTableMsg *pCreate); int32_t mgmtCheckDbLimit(SAcctObj *pAcct);
extern int32_t (*mgmtGetAcctMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn); int32_t mgmtCheckTableLimit(SAcctObj *pAcct, SCreateTableMsg *pCreate);
extern int32_t (*mgmtRetrieveAccts)(SShowObj *pShow, char *data, int32_t rows, void *pConn); int32_t mgmtGetAcctMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveAccts(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtDoStatistic(void *handle, void *tmrId);
#ifdef __cplusplus #ifdef __cplusplus
} }

26
src/mnode/inc/mgmtDnode.h
View File

@ -24,10 +24,6 @@ extern "C" {
#include <stdbool.h> #include <stdbool.h>
#include "mnode.h" #include "mnode.h"
int32_t mgmtCreateDnode(uint32_t ip);
int32_t mgmtDropDnode(SDnodeObj *pDnode);
int32_t mgmtDropDnodeByIp(uint32_t ip);
int32_t mgmtGetNextVnode(SVnodeGid *pVnodeGid);
void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId); void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId);
void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes); void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes);
int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
@ -44,19 +40,17 @@ int32_t mgmtRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pCo
int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn); int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
extern int32_t (*mgmtInitDnodes)(); int32_t mgmtGetScoresMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern void (*mgmtCleanUpDnodes)(); int32_t mgmtRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn);
extern SDnodeObj* (*mgmtGetDnode)(uint32_t ip);
extern int32_t (*mgmtGetDnodesNum)();
extern void* (*mgmtGetNextDnode)(SShowObj *pShow, SDnodeObj **pDnode);
extern int32_t (*mgmtUpdateDnode)(SDnodeObj *pDnode);
extern void (*mgmtSetDnodeUnRemove)(SDnodeObj *pDnode);
extern int32_t (*mgmtGetScoresMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveScores)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
extern bool (*mgmtCheckConfigShow)(SGlobalConfig *cfg);
extern SDnodeObj tsDnodeObj;
int32_t mgmtInitDnodes();
void mgmtCleanUpDnodes();
int32_t mgmtGetDnodesNum();
int32_t mgmtUpdateDnode(SDnodeObj *pDnode);
void* mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode);
bool mgmtCheckConfigShow(SGlobalConfig *cfg);
void mgmtSetDnodeUnRemove(SDnodeObj *pDnode);
SDnodeObj* mgmtGetDnode(uint32_t ip);
#ifdef __cplusplus #ifdef __cplusplus
} }

6
src/mnode/inc/mgmtDnodeInt.h
View File

@ -34,12 +34,12 @@ void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
void mgmtSendOneFreeVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle); void mgmtSendOneFreeVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
void mgmtSendRemoveVgroupMsg(SVgObj *pVgroup, void *ahandle); void mgmtSendRemoveVgroupMsg(SVgObj *pVgroup, void *ahandle);
extern int32_t (*mgmtInitDnodeInt)(); int32_t mgmtInitDnodeInt();
extern void (*mgmtCleanUpDnodeInt)(); void mgmtCleanUpDnodeInt();
extern void (*mgmtProcessDnodeStatus)(void *handle, void *tmrId);
void mgmtSendMsgToDnode(SRpcIpSet *ipSet, int8_t msgType, void *pCont, int32_t contLen, void *ahandle); void mgmtSendMsgToDnode(SRpcIpSet *ipSet, int8_t msgType, void *pCont, int32_t contLen, void *ahandle);
void mgmtSendRspToDnode(void *pConn, int8_t msgType, int32_t code, void *pCont, int32_t contLen); void mgmtSendRspToDnode(void *pConn, int8_t msgType, int32_t code, void *pCont, int32_t contLen);
void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
#ifdef __cplusplus #ifdef __cplusplus
} }

18
src/mnode/inc/mgmtGrant.h
View File

@ -17,21 +17,21 @@
#define TDENGINE_MGMT_GTANT_H #define TDENGINE_MGMT_GTANT_H
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { "C" {
#endif #endif
#include <stdint.h> #include <stdint.h>
#include <stdbool.h> #include <stdbool.h>
#include "mnode.h" #include "mnode.h"
extern bool (*mgmtCheckExpired)(); bool mgmtCheckExpired();
extern void (*mgmtAddTimeSeries)(SAcctObj *pAcct, uint32_t timeSeriesNum); void mgmtAddTimeSeries(SAcctObj *pAcct, uint32_t timeSeriesNum);
extern void (*mgmtRestoreTimeSeries)(SAcctObj *pAcct, uint32_t timeseries); void mgmtRestoreTimeSeries(SAcctObj *pAcct, uint32_t timeseries);
extern int32_t (*mgmtCheckTimeSeries)(uint32_t timeseries); int32_t mgmtCheckTimeSeries(uint32_t timeseries);
extern int32_t (*mgmtCheckUserGrant)(); int32_t mgmtCheckUserGrant();
extern int32_t (*mgmtCheckDbGrant)(); int32_t mgmtCheckDbGrant();
extern int32_t (*mgmtGetGrantsMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn); int32_t mgmtGetGrantsMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveGrants)(SShowObj *pShow, char *data, int rows, void *pConn); int32_t mgmtRetrieveGrants(SShowObj *pShow, char *data, int rows, void *pConn);
#ifdef __cplusplus #ifdef __cplusplus
} }

4
src/mnode/inc/mgmtMnode.h
View File

@ -24,8 +24,8 @@ extern "C" {
#include <stdbool.h> #include <stdbool.h>
#include "mnode.h" #include "mnode.h"
extern int32_t (*mgmtGetMnodeMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn); int32_t mgmtGetMnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveMnodes)(SShowObj *pShow, char *data, int32_t rows, void *pConn); int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
#ifdef __cplusplus #ifdef __cplusplus
} }

8
src/mnode/inc/mgmtSystem.h
View File

@ -22,13 +22,13 @@ extern "C" {
#include <stdint.h> #include <stdint.h>
int32_t mgmtInitSystem();
int32_t mgmtStartSystem(); int32_t mgmtStartSystem();
void mgmtCleanUpSystem(); void mgmtCleanUpSystem();
void mgmtStopSystem();
extern int32_t (*mgmtInitSystem)();
extern int32_t (*mgmtCheckMgmtRunning)();
extern void (*mgmtDoStatistic)(void *handle, void *tmrId);
extern void (*mgmtStopSystem)();
extern void (*mgmtCleanUpRedirect)(); extern void (*mgmtCleanUpRedirect)();
#ifdef __cplusplus #ifdef __cplusplus

126
src/mnode/src/mgmtAcct.c
View File

@ -26,6 +26,16 @@ extern void *tsUserSdb;
extern void *tsDbSdb; extern void *tsDbSdb;
static SAcctObj tsAcctObj; static SAcctObj tsAcctObj;
int32_t (*mgmtInitAcctsFp)() = NULL;
void (*mgmtCleanUpAcctsFp)() = NULL;
SAcctObj *(*mgmtGetAcctFp)(char *acctName) = NULL;
int32_t (*mgmtCheckUserLimitFp)(SAcctObj *pAcct) = NULL;
int32_t (*mgmtCheckDbLimitFp)(SAcctObj *pAcct) = NULL;
int32_t (*mgmtCheckTableLimitFp)(SAcctObj *pAcct, SCreateTableMsg *pCreate) = NULL;
int32_t (*mgmtGetAcctMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = NULL;
int32_t (*mgmtRetrieveAcctsFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn) = NULL;
void (*mgmtDoStatisticFp)(void *handle, void *tmrId) = NULL;
int32_t mgmtAddDbIntoAcct(SAcctObj *pAcct, SDbObj *pDb) { int32_t mgmtAddDbIntoAcct(SAcctObj *pAcct, SDbObj *pDb) {
pthread_mutex_lock(&pAcct->mutex); pthread_mutex_lock(&pAcct->mutex);
pDb->next = pAcct->pHead; pDb->next = pAcct->pHead;
@ -97,73 +107,95 @@ int32_t mgmtRemoveUserFromAcct(SAcctObj *pAcct, SUserObj *pUser) {
return 0; return 0;
} }
int32_t mgmtInitAcctsImp() { int32_t mgmtInitAccts() {
SAcctObj *pAcct = &tsAcctObj; if (mgmtInitAcctsFp) {
pAcct->acctId = 0; return mgmtInitAcctsFp();
strcpy(pAcct->user, "root"); } else {
return 0; SAcctObj *pAcct = &tsAcctObj;
} pAcct->acctId = 0;
strcpy(pAcct->user, "root");
int32_t (*mgmtInitAccts)() = mgmtInitAcctsImp; return 0;
static SAcctObj *mgmtGetAcctImp(char *acctName) {
return &tsAcctObj;
}
SAcctObj *(*mgmtGetAcct)(char *acctName) = mgmtGetAcctImp;
static int32_t mgmtCheckUserLimitImp(SAcctObj *pAcct) {
int32_t numOfUsers = sdbGetNumOfRows(tsUserSdb);
if (numOfUsers >= tsMaxUsers) {
mWarn("numOfUsers:%d, exceed tsMaxUsers:%d", numOfUsers, tsMaxUsers);
return TSDB_CODE_TOO_MANY_USERS;
} }
return 0;
} }
int32_t (*mgmtCheckUserLimit)(SAcctObj *pAcct) = mgmtCheckUserLimitImp; SAcctObj *mgmtGetAcct(char *acctName) {
if (mgmtGetAcctFp) {
static int32_t mgmtCheckDbLimitImp(SAcctObj *pAcct) { return mgmtGetAcctFp(acctName);
int32_t numOfDbs = sdbGetNumOfRows(tsDbSdb); } else {
if (numOfDbs >= tsMaxDbs) { return &tsAcctObj;
mWarn("numOfDbs:%d, exceed tsMaxDbs:%d", numOfDbs, tsMaxDbs);
return TSDB_CODE_TOO_MANY_DATABASES;
} }
return 0;
} }
int32_t (*mgmtCheckDbLimit)(SAcctObj *pAcct) = mgmtCheckDbLimitImp; int32_t mgmtCheckUserLimit(SAcctObj *pAcct) {
if (mgmtCheckUserLimitFp) {
static int32_t mgmtCheckTableLimitImp(SAcctObj *pAcct, SCreateTableMsg *pCreate) { return mgmtCheckUserLimitFp(pAcct);
return 0; } else {
int32_t numOfUsers = sdbGetNumOfRows(tsUserSdb);
if (numOfUsers >= tsMaxUsers) {
mWarn("numOfUsers:%d, exceed tsMaxUsers:%d", numOfUsers, tsMaxUsers);
return TSDB_CODE_TOO_MANY_USERS;
}
return 0;
}
} }
int32_t (*mgmtCheckTableLimit)(SAcctObj *pAcct, SCreateTableMsg *pCreate) = mgmtCheckTableLimitImp; int32_t mgmtCheckDbLimit(SAcctObj *pAcct) {
if (mgmtCheckDbLimitFp) {
static void mgmtCleanUpAcctsImp() { return mgmtCheckDbLimitFp(pAcct);
} else {
int32_t numOfDbs = sdbGetNumOfRows(tsDbSdb);
if (numOfDbs >= tsMaxDbs) {
mWarn("numOfDbs:%d, exceed tsMaxDbs:%d", numOfDbs, tsMaxDbs);
return TSDB_CODE_TOO_MANY_DATABASES;
}
return 0;
}
} }
void (*mgmtCleanUpAccts)() = mgmtCleanUpAcctsImp; int32_t mgmtCheckTableLimit(SAcctObj *pAcct, SCreateTableMsg *pCreate) {
if (mgmtCheckTableLimitFp) {
static int32_t mgmtGetAcctMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) { return mgmtCheckTableLimitFp(pAcct, pCreate);
return TSDB_CODE_OPS_NOT_SUPPORT; } else {
return 0;
}
} }
int32_t (*mgmtGetAcctMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetAcctMetaImp; void mgmtCleanUpAccts() {
if (mgmtCleanUpAcctsFp) {
static int32_t mgmtRetrieveAcctsImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) { mgmtCleanUpAcctsFp();
return 0; }
} }
int32_t (*mgmtRetrieveAccts)(SShowObj *pShow, char *data, int32_t rows, void *pConn) = mgmtRetrieveAcctsImp; int32_t mgmtGetAcctMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
if (mgmtGetAcctMetaFp) {
return mgmtGetAcctMetaFp(pMeta, pShow, pConn);
} else {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
}
int32_t mgmtRetrieveAccts(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
if (mgmtRetrieveAcctsFp) {
return mgmtRetrieveAcctsFp(pShow, data, rows, pConn);
} else {
return 0;
}
}
SAcctObj *mgmtGetAcctFromConn(void *pConn) { SAcctObj *mgmtGetAcctFromConn(void *pConn) {
SRpcConnInfo connInfo; SRpcConnInfo connInfo;
rpcGetConnInfo(pConn, &connInfo); rpcGetConnInfo(pConn, &connInfo);
SUserObj *pUser = mgmtGetUser(connInfo.user);
if(pUser != NULL) { SUserObj *pUser = mgmtGetUser(connInfo.user);
if (pUser != NULL) {
return pUser->pAcct; return pUser->pAcct;
} }
return NULL; return NULL;
} }
void mgmtDoStatistic(void *handle, void *tmrId) {
if (mgmtDoStatisticFp) {
mgmtDoStatisticFp(handle, tmrId);
}
}

1
src/mnode/src/mgmtBalance.c
View File

@ -32,7 +32,6 @@ void (*mgmtCleanupBalance)() = mgmtCleanupBalanceImp;
int32_t mgmtAllocVnodesImp(SVgObj *pVgroup) { int32_t mgmtAllocVnodesImp(SVgObj *pVgroup) {
int selectedVnode = -1; int selectedVnode = -1;
SDnodeObj *pDnode = &tsDnodeObj;
int lastAllocVode = pDnode->lastAllocVnode; int lastAllocVode = pDnode->lastAllocVnode;
for (int i = 0; i < pDnode->numOfVnodes; i++) { for (int i = 0; i < pDnode->numOfVnodes; i++) {

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

@ -24,7 +24,17 @@
#include "mgmtUser.h" #include "mgmtUser.h"
#include "mgmtVgroup.h" #include "mgmtVgroup.h"
SDnodeObj tsDnodeObj; int32_t (*mgmtInitDnodesFp)() = NULL;
void (*mgmtCleanUpDnodesFp)() = NULL;
SDnodeObj *(*mgmtGetDnodeFp)(uint32_t ip) = NULL;
int32_t (*mgmtGetDnodesNumFp)() = NULL;
int32_t (*mgmtUpdateDnodeFp)(SDnodeObj *pDnode) = NULL;
void * (*mgmtGetNextDnodeFp)(SShowObj *pShow, SDnodeObj **pDnode) = NULL;
int32_t (*mgmtGetScoresMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = NULL;
int32_t (*mgmtRetrieveScoresFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn) = NULL;
void (*mgmtSetDnodeUnRemoveFp)(SDnodeObj *pDnode) = NULL;
static SDnodeObj tsDnodeObj;
void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) { void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) {
int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore; int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore;
@ -154,7 +164,9 @@ int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
pShow->numOfColumns = cols; pShow->numOfColumns = cols;
pShow->offset[0] = 0; pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1]; for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = mgmtGetDnodesNum(); pShow->numOfRows = mgmtGetDnodesNum();
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1]; pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
@ -165,9 +177,9 @@ int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) { int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
int32_t numOfRows = 0; int32_t numOfRows = 0;
int32_t cols = 0;
SDnodeObj *pDnode = NULL; SDnodeObj *pDnode = NULL;
char *pWrite; char *pWrite;
int32_t cols = 0;
char ipstr[20]; char ipstr[20];
while (numOfRows < rows) { while (numOfRows < rows) {
@ -517,85 +529,94 @@ int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pCon
return numOfRows; return numOfRows;
} }
SDnodeObj *mgmtGetDnodeImp(uint32_t ip) { int32_t mgmtInitDnodes() {
return &tsDnodeObj; if (mgmtInitDnodesFp) {
} return mgmtInitDnodesFp();
SDnodeObj *(*mgmtGetDnode)(uint32_t ip) = mgmtGetDnodeImp;
int32_t mgmtUpdateDnodeImp(SDnodeObj *pDnode) {
return 0;
}
int32_t (*mgmtUpdateDnode)(SDnodeObj *pDnode) = mgmtUpdateDnodeImp;
void mgmtCleanUpDnodesImp() {
}
void (*mgmtCleanUpDnodes)() = mgmtCleanUpDnodesImp;
int32_t mgmtInitDnodesImp() {
tsDnodeObj.privateIp = inet_addr(tsPrivateIp);;
tsDnodeObj.createdTime = taosGetTimestampMs();
tsDnodeObj.lastReboot = taosGetTimestampSec();
tsDnodeObj.numOfCores = (uint16_t) tsNumOfCores;
tsDnodeObj.status = TSDB_DN_STATUS_READY;
tsDnodeObj.alternativeRole = TSDB_DNODE_ROLE_ANY;
tsDnodeObj.numOfTotalVnodes = tsNumOfTotalVnodes;
tsDnodeObj.thandle = (void *) (1); //hack way
if (tsDnodeObj.numOfVnodes == TSDB_INVALID_VNODE_NUM) {
mgmtSetDnodeMaxVnodes(&tsDnodeObj);
mPrint("dnode first access, set total vnodes:%d", tsDnodeObj.numOfVnodes);
}
tsDnodeObj.status = TSDB_DN_STATUS_READY;
return 0;
}
int32_t (*mgmtInitDnodes)() = mgmtInitDnodesImp;
int32_t mgmtGetDnodesNumImp() {
return 1;
}
int32_t (*mgmtGetDnodesNum)() = mgmtGetDnodesNumImp;
void *mgmtGetNextDnodeImp(SShowObj *pShow, SDnodeObj **pDnode) {
if (*pDnode == NULL) {
*pDnode = &tsDnodeObj;
} else { } else {
*pDnode = NULL; tsDnodeObj.privateIp = inet_addr(tsPrivateIp);;
tsDnodeObj.createdTime = taosGetTimestampMs();
tsDnodeObj.lastReboot = taosGetTimestampSec();
tsDnodeObj.numOfCores = (uint16_t) tsNumOfCores;
tsDnodeObj.status = TSDB_DN_STATUS_READY;
tsDnodeObj.alternativeRole = TSDB_DNODE_ROLE_ANY;
tsDnodeObj.numOfTotalVnodes = tsNumOfTotalVnodes;
tsDnodeObj.thandle = (void *) (1); //hack way
tsDnodeObj.status = TSDB_DN_STATUS_READY;
mgmtSetDnodeMaxVnodes(&tsDnodeObj);
return 0;
}
}
void mgmtCleanUpDnodes() {
if (mgmtCleanUpDnodesFp) {
mgmtCleanUpDnodesFp();
}
}
SDnodeObj *mgmtGetDnode(uint32_t ip) {
if (mgmtGetDnodeFp) {
return mgmtGetDnodeFp(ip);
} else {
return &tsDnodeObj;
}
}
int32_t mgmtGetDnodesNum() {
if (mgmtGetDnodesNumFp) {
return mgmtGetDnodesNumFp();
} else {
return 1;
}
}
int32_t mgmtUpdateDnode(SDnodeObj *pDnode) {
if (mgmtUpdateDnodeFp) {
return mgmtUpdateDnodeFp(pDnode);
} else {
return 0;
}
}
void *mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode) {
if (mgmtGetNextDnodeFp) {
return mgmtGetNextDnodeFp(pShow, pDnode);
} else {
if (*pDnode == NULL) {
*pDnode = &tsDnodeObj;
} else {
*pDnode = NULL;
}
} }
return *pDnode; return *pDnode;
} }
void *(*mgmtGetNextDnode)(SShowObj *pShow, SDnodeObj **pDnode) = mgmtGetNextDnodeImp; int32_t mgmtGetScoresMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
if (mgmtGetScoresMetaFp) {
int32_t mgmtGetScoresMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) { return mgmtGetScoresMetaFp(pMeta, pShow, pConn);
return TSDB_CODE_OPS_NOT_SUPPORT; } else {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
} }
int32_t (*mgmtGetScoresMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetScoresMetaImp; int32_t mgmtRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
if (mgmtRetrieveScoresFp) {
int32_t mgmtRetrieveScoresImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) { return mgmtRetrieveScoresFp(pShow, data, rows, pConn);
return 0; } else {
return 0;
}
} }
int32_t (*mgmtRetrieveScores)(SShowObj *pShow, char *data, int32_t rows, void *pConn) = mgmtRetrieveScoresImp; void mgmtSetDnodeUnRemove(SDnodeObj *pDnode) {
if (mgmtSetDnodeUnRemoveFp) {
void mgmtSetDnodeUnRemoveImp(SDnodeObj *pDnode) { mgmtSetDnodeUnRemoveFp(pDnode);
}
} }
void (*mgmtSetDnodeUnRemove)(SDnodeObj *pDnode) = mgmtSetDnodeUnRemoveImp; bool mgmtCheckConfigShow(SGlobalConfig *cfg) {
bool mgmtCheckConfigShowImp(SGlobalConfig *cfg) {
if (cfg->cfgType & TSDB_CFG_CTYPE_B_CLUSTER) if (cfg->cfgType & TSDB_CFG_CTYPE_B_CLUSTER)
return false; return false;
if (cfg->cfgType & TSDB_CFG_CTYPE_B_NOT_PRINT) if (cfg->cfgType & TSDB_CFG_CTYPE_B_NOT_PRINT)
return false; return false;
return true; return true;
} }
bool (*mgmtCheckConfigShow)(SGlobalConfig *cfg) = mgmtCheckConfigShowImp;

212
src/mnode/src/mgmtDnodeInt.c
View File

@ -31,10 +31,15 @@
#include "mgmtTable.h" #include "mgmtTable.h"
#include "mgmtVgroup.h" #include "mgmtVgroup.h"
int32_t (*mgmtInitDnodeIntFp)() = NULL;
void (*mgmtCleanUpDnodeIntFp)() = NULL;
void (*mgmtSendMsgToDnodeFp)(SRpcIpSet *ipSet, int8_t msgType, void *pCont, int32_t contLen, void *ahandle) = NULL; void (*mgmtSendMsgToDnodeFp)(SRpcIpSet *ipSet, int8_t msgType, void *pCont, int32_t contLen, void *ahandle) = NULL;
void (*mgmtSendRspToDnodeFp)(void *handle, int32_t code, void *pCont, int32_t contLen) = NULL; void (*mgmtSendRspToDnodeFp)(void *handle, int32_t code, void *pCont, int32_t contLen) = NULL;
void *mgmtStatusTimer = NULL; void *mgmtStatusTimer = NULL;
static void mgmtProcessDnodeStatus(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
static void mgmtSendMsgToDnodeQueueFp(SSchedMsg *sched) { static void mgmtSendMsgToDnodeQueueFp(SSchedMsg *sched) {
int32_t contLen = *(int32_t *) (sched->msg - 4); int32_t contLen = *(int32_t *) (sched->msg - 4);
int32_t code = *(int32_t *) (sched->msg - 8); int32_t code = *(int32_t *) (sched->msg - 8);
@ -249,6 +254,10 @@ void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *p
mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code); mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code);
} else if (msgType == TSDB_MSG_TYPE_DNODE_CFG_RSP) { } else if (msgType == TSDB_MSG_TYPE_DNODE_CFG_RSP) {
} else if (msgType == TSDB_MSG_TYPE_ALTER_STREAM_RSP) { } else if (msgType == TSDB_MSG_TYPE_ALTER_STREAM_RSP) {
} else if (msgType == TSDB_MSG_TYPE_STATUS) {
mgmtProcessDnodeStatus(msgType, pConn, contLen, pConn, code);
} else if (msgType == TSDB_MSG_TYPE_GRANT) {
mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code);
} else { } else {
mError("%s from dnode is not processed", taosMsg[(int8_t)msgType]); mError("%s from dnode is not processed", taosMsg[(int8_t)msgType]);
} }
@ -256,8 +265,6 @@ void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *p
//rpcFreeCont(pCont); //rpcFreeCont(pCont);
} }
void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) { void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) {
mTrace("table:%s, sid:%d send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle); mTrace("table:%s, sid:%d send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle);
} }
@ -317,108 +324,113 @@ int32_t mgmtCfgDynamicOptions(SDnodeObj *pDnode, char *msg) {
} }
int32_t mgmtSendCfgDnodeMsg(char *cont) { int32_t mgmtSendCfgDnodeMsg(char *cont) {
#ifdef CLUSTER //#ifdef CLUSTER
char * pMsg, *pStart; // char * pMsg, *pStart;
int32_t msgLen = 0; // int32_t msgLen = 0;
#endif //#endif
//
SDnodeObj *pDnode; // SDnodeObj *pDnode;
SCfgDnodeMsg * pCfg = (SCfgDnodeMsg *)cont; // SCfgDnodeMsg * pCfg = (SCfgDnodeMsg *)cont;
uint32_t ip; // uint32_t ip;
//
ip = inet_addr(pCfg->ip); // ip = inet_addr(pCfg->ip);
pDnode = mgmtGetDnode(ip); // pDnode = mgmtGetDnode(ip);
if (pDnode == NULL) { // if (pDnode == NULL) {
mError("dnode ip:%s not configured", pCfg->ip); // mError("dnode ip:%s not configured", pCfg->ip);
return TSDB_CODE_NOT_CONFIGURED; // return TSDB_CODE_NOT_CONFIGURED;
} // }
//
mTrace("dnode:%s, dynamic option received, content:%s", taosIpStr(pDnode->privateIp), pCfg->config); // mTrace("dnode:%s, dynamic option received, content:%s", taosIpStr(pDnode->privateIp), pCfg->config);
int32_t code = mgmtCfgDynamicOptions(pDnode, pCfg->config); // int32_t code = mgmtCfgDynamicOptions(pDnode, pCfg->config);
if (code != -1) { // if (code != -1) {
return code; // return code;
} // }
//
#ifdef CLUSTER //#ifdef CLUSTER
pStart = taosBuildReqMsg(pDnode->thandle, TSDB_MSG_TYPE_DNODE_CFG); // pStart = taosBuildReqMsg(pDnode->thandle, TSDB_MSG_TYPE_DNODE_CFG);
if (pStart == NULL) return TSDB_CODE_NODE_OFFLINE; // if (pStart == NULL) return TSDB_CODE_NODE_OFFLINE;
pMsg = pStart; // pMsg = pStart;
//
memcpy(pMsg, cont, sizeof(SCfgDnodeMsg)); // memcpy(pMsg, cont, sizeof(SCfgDnodeMsg));
pMsg += sizeof(SCfgDnodeMsg); // pMsg += sizeof(SCfgDnodeMsg);
//
msgLen = pMsg - pStart; // msgLen = pMsg - pStart;
mgmtSendMsgToDnode(pDnode, pStart, msgLen); // mgmtSendMsgToDnode(pDnode, pStart, msgLen);
#else //#else
(void)tsCfgDynamicOptions(pCfg->config); // (void)tsCfgDynamicOptions(pCfg->config);
#endif //#endif
return 0; // return 0;
} }
int32_t mgmtInitDnodeIntImp() { return 0; } int32_t mgmtInitDnodeInt() {
int32_t (*mgmtInitDnodeInt)() = mgmtInitDnodeIntImp; if (mgmtInitDnodeIntFp) {
return mgmtInitDnodeIntFp();
} else {
return 0;
}
}
void mgmtCleanUpDnodeIntImp() {} void mgmtCleanUpDnodeInt() {
void (*mgmtCleanUpDnodeInt)() = mgmtCleanUpDnodeIntImp; if (mgmtCleanUpDnodeIntFp) {
mgmtCleanUpDnodeIntFp();
}
}
void mgmtProcessDnodeStatus(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
SStatusMsg *pStatus = (SStatusMsg *)pCont;
SDnodeObj *pObj = mgmtGetDnode(htonl(pStatus->privateIp));
if (pObj == NULL) {
mError("dnode:%s not exist", taosIpStr(pObj->privateIp));
mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_DNODE_NOT_EXIST, NULL, 0);
return;
}
pObj->lastReboot = htonl(pStatus->lastReboot);
pObj->numOfTotalVnodes = htons(pStatus->numOfTotalVnodes);
pObj->openVnodes = htons(pStatus->openVnodes);
pObj->numOfCores = htons(pStatus->numOfCores);
pObj->diskAvailable = pStatus->diskAvailable;
pObj->alternativeRole = pStatus->alternativeRole;
//
// if (mgmtProcessDnodeStatusFp) {
// mgmtProcessDnodeStatusFp(pStatus, pObj, pConn);
// return;
// }
void mgmtProcessDnodeStatusImp(void *handle, void *tmrId) {
/*
SDnodeObj *pObj = &tsDnodeObj;
pObj->openVnodes = tsOpenVnodes;
pObj->status = TSDB_DN_STATUS_READY; pObj->status = TSDB_DN_STATUS_READY;
float memoryUsedMB = 0; // // wait vnode dropped
taosGetSysMemory(&memoryUsedMB); // for (int32_t vnode = 0; vnode < pObj->numOfVnodes; ++vnode) {
pObj->diskAvailable = tsAvailDataDirGB; // SVnodeLoad *pVload = &(pObj->vload[vnode]);
// if (pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) {
for (int32_t vnode = 0; vnode < pObj->numOfVnodes; ++vnode) { // bool existInDnode = false;
SVnodeLoad *pVload = &(pObj->vload[vnode]); // for (int32_t j = 0; j < pObj->openVnodes; ++j) {
SVnodeObj * pVnode = vnodeList + vnode; // if (htonl(pStatus->load[j].vnode) == vnode) {
// existInDnode = true;
// wait vnode dropped // break;
if (pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) { // }
if (vnodeList[vnode].cfg.maxSessions <= 0) { // }
pVload->dropStatus = TSDB_VN_DROP_STATUS_READY; //
pVload->status = TSDB_VN_STATUS_OFFLINE; // if (!existInDnode) {
mPrint("dnode:%s, vid:%d, drop finished", taosIpStr(pObj->privateIp), vnode); // pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
taosTmrStart(mgmtMonitorDbDrop, 10000, NULL, tsMgmtTmr); // pVload->status = TSDB_VN_STATUS_OFFLINE;
} // mgmtUpdateDnode(pObj);
} // mPrint("dnode:%s, vid:%d, drop finished", taosIpStr(pObj->privateIp), vnode);
// taosTmrStart(mgmtMonitorDbDrop, 10000, NULL, tsMgmtTmr);
if (vnodeList[vnode].cfg.maxSessions <= 0) { // }
continue; // } else if (pVload->vgId == 0) {
} // /*
// * In some cases, vnode information may be reported abnormally, recover it
pVload->vnode = vnode; // */
pVload->status = TSDB_VN_STATUS_MASTER; // if (pVload->dropStatus != TSDB_VN_DROP_STATUS_READY || pVload->status != TSDB_VN_STATUS_OFFLINE) {
pVload->totalStorage = pVnode->vnodeStatistic.totalStorage; // mPrint("dnode:%s, vid:%d, vgroup:%d status:%s dropStatus:%s, set it to avail status",
pVload->compStorage = pVnode->vnodeStatistic.compStorage; // taosIpStr(pObj->privateIp), vnode, pVload->vgId, taosGetVnodeStatusStr(pVload->status),
pVload->pointsWritten = pVnode->vnodeStatistic.pointsWritten; // taosGetVnodeDropStatusStr(pVload->dropStatus));
uint32_t vgId = pVnode->cfg.vgId; // pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
// pVload->status = TSDB_VN_STATUS_OFFLINE;
SVgObj *pVgroup = mgmtGetVgroup(vgId); // mgmtUpdateDnode(pObj);
if (pVgroup == NULL) { // }
mError("vgroup:%d is not there, but associated with vnode %d", vgId, vnode); // }
pVload->dropStatus = TSDB_VN_DROP_STATUS_DROPPING; // }
continue;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("vgroup:%d not belongs to any database, vnode:%d", vgId, vnode);
continue;
}
if (pVload->vgId == 0 || pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) {
mError("vid:%d, mgmt not exist, drop it", vnode);
pVload->dropStatus = TSDB_VN_DROP_STATUS_DROPPING;
}
}
taosTmrReset(mgmtProcessDnodeStatus, tsStatusInterval * 1000, NULL, tsMgmtTmr, &mgmtStatusTimer);
if (mgmtStatusTimer == NULL) {
mError("Failed to start status timer");
}
*/
} }
void (*mgmtProcessDnodeStatus)(void *handle, void *tmrId) = mgmtProcessDnodeStatusImp;

75
src/mnode/src/mgmtGrant.c
View File

@ -19,31 +19,74 @@
#include "mgmtAcct.h" #include "mgmtAcct.h"
#include "mgmtGrant.h" #include "mgmtGrant.h"
int32_t mgmtCheckUserGrantImp() { return 0; } int32_t (*mgmtCheckUserGrantFp)() = NULL;
int32_t (*mgmtCheckUserGrant)() = mgmtCheckUserGrantImp; int32_t (*mgmtCheckDbGrantFp)() = NULL;
void (*mgmtAddTimeSeriesFp)(uint32_t timeSeriesNum) = NULL;
void (*mgmtRestoreTimeSeriesFp)(uint32_t timeSeriesNum) = NULL;
int32_t (*mgmtCheckTimeSeriesFp)(uint32_t timeseries) = NULL;
bool (*mgmtCheckExpiredFp)() = NULL;
int32_t (*mgmtGetGrantsMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = NULL;
int32_t (*mgmtRetrieveGrantsFp)(SShowObj *pShow, char *data, int rows, void *pConn) = NULL;
int32_t mgmtCheckDbGrantImp() { return 0; } int32_t mgmtCheckUserGrant() {
int32_t (*mgmtCheckDbGrant)() = mgmtCheckDbGrantImp; if (mgmtCheckUserGrantFp) {
return mgmtCheckUserGrantFp();
} else {
return 0;
}
}
void mgmtAddTimeSeriesImp(SAcctObj *pAcct, uint32_t timeSeriesNum) { int32_t mgmtCheckDbGrant() {
if (mgmtCheckDbGrantFp) {
return mgmtCheckDbGrantFp();
} else {
return 0;
}
}
void mgmtAddTimeSeries(SAcctObj *pAcct, uint32_t timeSeriesNum) {
pAcct->acctInfo.numOfTimeSeries += timeSeriesNum; pAcct->acctInfo.numOfTimeSeries += timeSeriesNum;
if (mgmtAddTimeSeriesFp) {
mgmtAddTimeSeriesFp(timeSeriesNum);
}
} }
void (*mgmtAddTimeSeries)(SAcctObj *pAcct, uint32_t timeSeriesNum) = mgmtAddTimeSeriesImp;
void mgmtRestoreTimeSeriesImp(SAcctObj *pAcct, uint32_t timeSeriesNum) { void mgmtRestoreTimeSeries(SAcctObj *pAcct, uint32_t timeSeriesNum) {
pAcct->acctInfo.numOfTimeSeries -= timeSeriesNum; pAcct->acctInfo.numOfTimeSeries -= timeSeriesNum;
if (mgmtRestoreTimeSeriesFp) {
mgmtRestoreTimeSeriesFp(timeSeriesNum);
}
} }
void (*mgmtRestoreTimeSeries)(SAcctObj *pAcct, uint32_t timeSeriesNum) = mgmtRestoreTimeSeriesImp;
int32_t mgmtCheckTimeSeriesImp(uint32_t timeseries) { return 0; } int32_t mgmtCheckTimeSeries(uint32_t timeseries) {
int32_t (*mgmtCheckTimeSeries)(uint32_t timeseries) = mgmtCheckTimeSeriesImp; if (mgmtCheckTimeSeriesFp) {
return mgmtCheckTimeSeriesFp(timeseries);
} else {
return 0;
}
}
bool mgmtCheckExpiredImp() { return false; } bool mgmtCheckExpired() {
bool (*mgmtCheckExpired)() = mgmtCheckExpiredImp; if (mgmtCheckExpiredFp) {
return mgmtCheckExpiredFp();
} else {
return false;
}
}
int32_t mgmtGetGrantsMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) { return TSDB_CODE_OPS_NOT_SUPPORT; } int32_t mgmtGetGrantsMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t (*mgmtGetGrantsMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetGrantsMetaImp; if (mgmtGetGrantsMetaFp) {
return mgmtGetGrantsMetaFp(pMeta, pShow, pConn);
} else {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
}
int32_t mgmtRetrieveGrantsImp(SShowObj *pShow, char *data, int rows, void *pConn) { return 0; } int32_t mgmtRetrieveGrants(SShowObj *pShow, char *data, int rows, void *pConn) {
int32_t (*mgmtRetrieveGrants)(SShowObj *pShow, char *data, int rows, void *pConn) = mgmtRetrieveGrantsImp; if (mgmtRetrieveGrantsFp) {
return mgmtRetrieveGrantsFp(pShow, data, rows, pConn);
} else {
return 0;
}
}

130
src/mnode/src/mgmtMnode.c
View File

@ -14,16 +14,134 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tschemautil.h"
#include "mgmtMnode.h" #include "mgmtMnode.h"
#include "mgmtUser.h"
int32_t mgmtGetMnodeMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) { void *(*mgmtGetNextMnodeFp)(SShowObj *pShow, SSdbPeer **pMnode) = NULL;
return TSDB_CODE_OPS_NOT_SUPPORT; int32_t (*mgmtInitMnodesFp)() = NULL;
} int32_t (*mgmtGetMnodesNumFp)() = NULL;
int32_t (*mgmtGetMnodeMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetMnodeMetaImp; static int32_t mgmtGetMnodesNum();
static void *mgmtGetNextMnode(SShowObj *pShow, SSdbPeer **pMnode);
int32_t mgmtGetMnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SUserObj *pUser = mgmtGetUserFromConn(pConn);
if (pUser == NULL) return 0;
if (strcmp(pUser->user, "root") != 0) return TSDB_CODE_NO_RIGHTS;
SSchema *pSchema = tsGetSchema(pMeta);
pShow->bytes[cols] = 16;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "IP");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created time");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 10;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "status");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 10;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "role");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 16;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "public ip");
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 = mgmtGetMnodesNum();
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
pShow->pNode = NULL;
int32_t mgmtRetrieveMnodesImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
return 0; return 0;
} }
int32_t (*mgmtRetrieveMnodes)(SShowObj *pShow, char *data, int32_t rows, void *pConn) = mgmtRetrieveMnodesImp; int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
int32_t numOfRows = 0;
int32_t cols = 0;
SSdbPeer *pMnode = NULL;
char *pWrite;
char ipstr[20];
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextMnode(pShow, (SDnodeObj **)&pMnode);
pShow->pNode = sdbFetchRow(mnodeSdb, pShow->pNode, (void **)&pMnode);
if (pMnode == NULL) break;
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, pMnode->ipstr);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = pMnode->createdTime;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, sdbStatusStr[(uint8_t)pMnode->status]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, sdbRoleStr[(uint8_t)pMnode->role]);
cols++;
tinet_ntoa(ipstr, pMnode->publicIp);
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, ipstr);
cols++;
numOfRows++;
}
pShow->numOfReads += numOfRows;
return numOfRows;
}
static int32_t mgmtGetMnodesNum() {
if (mgmtGetMnodesNumFp) {
return mgmtGetMnodesNumFp();
} else {
return 1;
}
}
static void *mgmtGetNextMnode(SShowObj *pShow, SSdbPeer **pMnode) {
if (mgmtGetNextMnodeFp) {
return mgmtGetNextMnodeFp(pShow, pMnode);
} else {
if (*pMnode == NULL) {
*pMnode = &tsMnodeObj;
} else {
*pMnode = NULL;
}
}
return *pMnode;
}

1
src/mnode/src/mgmtShell.c
View File

@ -51,6 +51,7 @@ static void (*mgmtProcessShellMsg[TSDB_MSG_TYPE_MAX])(void *pCont, int32_t contL
static void mgmtProcessUnSupportMsg(void *pCont, int32_t contLen, void *ahandle); static void mgmtProcessUnSupportMsg(void *pCont, int32_t contLen, void *ahandle);
static int mgmtRetriveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey); static int mgmtRetriveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
uint32_t mgmtAccessSquence = 0;
void *tsShellConnServer = NULL; void *tsShellConnServer = NULL;
void mgmtProcessTranRequest(SSchedMsg *sched) { void mgmtProcessTranRequest(SSchedMsg *sched) {

71
src/mnode/src/mgmtSystem.c
View File

@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "taosdef.h" #include "taosdef.h"
#include "tmodule.h"
#include "tsched.h" #include "tsched.h"
#include "mnode.h" #include "mnode.h"
#include "mgmtAcct.h" #include "mgmtAcct.h"
@ -55,6 +56,20 @@ void mgmtCleanUpSystem() {
mPrint("mgmt is cleaned up"); mPrint("mgmt is cleaned up");
} }
int32_t mgmtCheckMgmtRunning() {
if (tsModuleStatus & (1 << TSDB_MOD_MGMT)) {
return -1;
}
tsetModuleStatus(TSDB_MOD_MGMT);
// strcpy(sdbMasterIp, mgmtIpStr[0]);
strcpy(sdbPrivateIp, tsPrivateIp);
sdbPublicIp = inet_addr(tsPublicIp);
return 0;
}
int32_t mgmtStartSystem() { int32_t mgmtStartSystem() {
mPrint("starting to initialize TDengine mgmt ..."); mPrint("starting to initialize TDengine mgmt ...");
@ -111,10 +126,10 @@ int32_t mgmtStartSystem() {
return -1; return -1;
} }
if (mgmtInitShell() < 0) { // if (mgmtInitShell() < 0) {
mError("failed to init shell"); // mError("failed to init shell");
return -1; // return -1;
} // }
if (sdbInitPeers(tsMgmtDirectory) < 0) { if (sdbInitPeers(tsMgmtDirectory) < 0) {
mError("failed to init peers"); mError("failed to init peers");
@ -125,39 +140,41 @@ int32_t mgmtStartSystem() {
mError("failed to init dnode balance") mError("failed to init dnode balance")
} }
taosTmrReset(mgmtDoStatistic, tsStatusInterval * 30000, NULL, tsMgmtTmr, &tsMgmtStatisTimer); if (mgmtDoStatistic) {
taosTmrReset(mgmtDoStatistic, tsStatusInterval * 30000, NULL, tsMgmtTmr, &tsMgmtStatisTimer);
}
mPrint("TDengine mgmt is initialized successfully"); mPrint("TDengine mgmt is initialized successfully");
return 0; return 0;
} }
int32_t mgmtInitSystemImp() { int32_t mgmtInitSystem() {
int32_t code = mgmtStartSystem(); struct stat dirstat;
if (code != 0) { bool directoryExist = (stat(tsMgmtDirectory, &dirstat) == 0);
return code; bool equalWithMaster = (strcmp(tsMasterIp, tsPrivateIp) == 0);
if (equalWithMaster || directoryExist) {
if (mgmtStartSystem() != 0) {
return -1;
}
} }
taosTmrReset(mgmtProcessDnodeStatus, 500, NULL, tsMgmtTmr, &mgmtStatusTimer); if (mgmtInitShell() < 0) {
return code; mError("failed to init shell");
} return -1;
}
int32_t (*mgmtInitSystem)() = mgmtInitSystemImp;
int32_t mgmtCheckMgmtRunningImp() {
return 0; return 0;
} }
int32_t (*mgmtCheckMgmtRunning)() = mgmtCheckMgmtRunningImp; void mgmtStopSystem() {
if (sdbMaster) {
mTrace("it is a master mgmt node, it could not be stopped");
return;
}
void mgmtDoStatisticImp(void *handle, void *tmrId) {} mgmtCleanUpSystem();
remove(tsMgmtDirectory);
void (*mgmtDoStatistic)(void *handle, void *tmrId) = mgmtDoStatisticImp; // mgmtInitRedirect();
}
void mgmtStopSystemImp() {}
void (*mgmtStopSystem)() = mgmtStopSystemImp;
void mgmtCleanUpRedirectImp() {}
void (*mgmtCleanUpRedirect)() = mgmtCleanUpRedirectImp;

1
src/os/linux/src/tsystem.c
View File

@ -585,6 +585,7 @@ void tsPrintOsInfo() {
pPrint(" os release: %s", buf.release); pPrint(" os release: %s", buf.release);
pPrint(" os version: %s", buf.version); pPrint(" os version: %s", buf.version);
pPrint(" os machine: %s", buf.machine); pPrint(" os machine: %s", buf.machine);
pPrint("==================================");
} }
void taosKillSystem() { void taosKillSystem() {

1
src/util/inc/tglobalcfg.h
View File

@ -193,7 +193,6 @@ extern char tsCharset[64]; // default encode string
// //
void tsReadGlobalLogConfig(); void tsReadGlobalLogConfig();
bool tsReadGlobalConfig(); bool tsReadGlobalConfig();
bool tsReadGlobalConfigSpec();
int tsCfgDynamicOptions(char *msg); int tsCfgDynamicOptions(char *msg);
void tsPrintGlobalConfig(); void tsPrintGlobalConfig();
void tsPrintGlobalConfigSpec(); void tsPrintGlobalConfigSpec();

24
src/util/src/tglobalcfg.c
View File

@ -24,6 +24,9 @@
#include "tsystem.h" #include "tsystem.h"
#include "tutil.h" #include "tutil.h"
void (*tsReadStorageConfig)() = NULL;
void (*tsPrintStorageConfig)() = NULL;
// monitor module api // monitor module api
int (*startMonitor)() = NULL; int (*startMonitor)() = NULL;
void (*stopMonitor)() = NULL; void (*stopMonitor)() = NULL;
@ -942,7 +945,9 @@ bool tsReadGlobalConfig() {
fclose(fp); fclose(fp);
} }
tsReadGlobalConfigSpec(); if (tsReadStorageConfig) {
tsReadStorageConfig();
}
if (tsPrivateIp[0] == 0) { if (tsPrivateIp[0] == 0) {
taosGetPrivateIp(tsPrivateIp); taosGetPrivateIp(tsPrivateIp);
@ -1111,11 +1116,13 @@ void tsPrintGlobalConfig() {
} }
} }
tsPrintGlobalConfigSpec(); if (tsPrintStorageConfig) {
tsPrintStorageConfig();
} else {
pPrint(" dataDir: %s", dataDir);
}
tsPrintOsInfo(); tsPrintOsInfo();
pPrint("==================================");
} }
void tsSetAllDebugFlag() { void tsSetAllDebugFlag() {
@ -1206,12 +1213,3 @@ void tsSetTimeZone() {
pPrint("timezone format changed to %s", tsTimezone); pPrint("timezone format changed to %s", tsTimezone);
} }
#ifndef CLUSTER
bool tsReadGlobalConfigSpec() { return true; }
void tsPrintGlobalConfigSpec() {
pPrint(" dataDir: %s", dataDir);
}
#endif