Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
未验证 提交 fcf29f14 编写于 作者: H hzcheng 提交者： GitHub
浏览文件
操作

Merge pull request #1387 from taosdata/refact/slguan 

Refact/slguan
上级 cb1602f7 4662cff5
隐藏空白更改
内联 并排
Showing with 922 addition and 769 deletion
src/dnode/inc/dnodeMClient.h
浏览文件 @ fcf29f14
......@@ -20,8 +20,10 @@
extern "C" {
#endif
int32_t dnodeInitMClient();
void dnodeCleanupMClient();
int32_t dnodeInitMClient();
void dnodeCleanupMClient();
void dnodeSendMsgToMnode(SRpcMsg *rpcMsg);
uint32_t dnodeGetMnodeMasteIp();
#ifdef __cplusplus
}
......
src/dnode/inc/dnodeMgmt.h
浏览文件 @ fcf29f14
......@@ -23,6 +23,7 @@ extern "C" {
int32_t dnodeInitMgmt();
void dnodeCleanupMgmt();
void dnodeMgmt(SRpcMsg *rpcMsg);
void dnodeUpdateDnodeId(int32_t dnodeId);
void* dnodeGetVnode(int32_t vgId);
int32_t dnodeGetVnodeStatus(void *pVnode);
......
src/dnode/inc/dnodeModule.h
浏览文件 @ fcf29f14
......@@ -23,6 +23,7 @@ extern "C" {
int32_t dnodeInitModules();
void dnodeCleanUpModules();
void dnodeStartModules();
void dnodeProcessModuleStatus(uint32_t moduleStatus);
#ifdef __cplusplus
}
......
src/dnode/src/dnodeMClient.c
浏览文件 @ fcf29f14
......@@ -18,16 +18,35 @@
#include "taosmsg.h"
#include "tlog.h"
#include "trpc.h"
#include "tutil.h"
#include "dnode.h"
#include "dnodeMClient.h"
#include "dnodeModule.h"
#include "dnodeMgmt.h"
static void (*dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static bool dnodeReadMnodeIpList();
static void dnodeSaveMnodeIpList();
static void dnodeProcessRspFromMnode(SRpcMsg *pMsg);
static void dnodeProcessStatusRsp(SRpcMsg *pMsg);
static void *tsDnodeMClientRpc;
static void (*tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void *tsDnodeMClientRpc = NULL;
static SRpcIpSet tsDnodeMnodeIpList = {0};
int32_t dnodeInitMClient() {
dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
if (!dnodeReadMnodeIpList()) {
dTrace("failed to read mnode iplist, set it from cfg file");
memset(&tsDnodeMnodeIpList, 0, sizeof(SRpcIpSet));
tsDnodeMnodeIpList.port = tsMnodeDnodePort;
tsDnodeMnodeIpList.numOfIps = 1;
tsDnodeMnodeIpList.ip[0] = inet_addr(tsMasterIp);
if (tsSecondIp[0]) {
tsDnodeMnodeIpList.numOfIps = 2;
tsDnodeMnodeIpList.ip[1] = inet_addr(tsSecondIp);
}
}
tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
......@@ -35,17 +54,20 @@ int32_t dnodeInitMClient() {
rpcInit.label = "DND-MC";
rpcInit.numOfThreads = 1;
rpcInit.cfp = dnodeProcessRspFromMnode;
rpcInit.sessions = TSDB_SESSIONS_PER_DNODE;
rpcInit.sessions = 100;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.idleTime = tsShellActivityTimer * 2000;
rpcInit.user = "t";
rpcInit.ckey = "key";
rpcInit.secret = "secret";
tsDnodeMClientRpc = rpcOpen(&rpcInit);
if (tsDnodeMClientRpc == NULL) {
dError("failed to init connection from mgmt");
dError("failed to init mnode rpc client");
return -1;
}
dPrint("client connection to mgmt is opened");
dPrint("mnode rpc client is opened");
return 0;
}
......@@ -53,18 +75,122 @@ void dnodeCleanupMClient() {
if (tsDnodeMClientRpc) {
rpcClose(tsDnodeMClientRpc);
tsDnodeMClientRpc = NULL;
dPrint("mnode rpc client is closed");
}
}
static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) {
if (dnodeProcessMgmtRspFp[pMsg->msgType]) {
(*dnodeProcessMgmtRspFp[pMsg->msgType])(pMsg);
if (tsDnodeProcessMgmtRspFp[pMsg->msgType]) {
(*tsDnodeProcessMgmtRspFp[pMsg->msgType])(pMsg);
} else {
dError("%s is not processed in mclient", taosMsg[pMsg->msgType]);
dError("%s is not processed in mnode rpc client", taosMsg[pMsg->msgType]);
}
rpcFreeCont(pMsg->pCont);
}
static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
if (pMsg->code != TSDB_CODE_SUCCESS) {
dError("status rsp is received, error:%s", tstrerror(pMsg->code));
return;
}
SDMStatusRsp *pStatusRsp = pMsg->pCont;
if (pStatusRsp->ipList.numOfIps <= 0) {
dError("status msg is invalid, num of ips is %d", pStatusRsp->ipList.numOfIps);
return;
}
pStatusRsp->ipList.port = htons(pStatusRsp->ipList.port);
for (int32_t i = 0; i < pStatusRsp->ipList.numOfIps; ++i) {
pStatusRsp->ipList.ip[i] = htonl(pStatusRsp->ipList.ip[i]);
}
//dTrace("status msg is received, result:%s", tstrerror(pMsg->code));
if (memcmp(&(pStatusRsp->ipList), &tsDnodeMnodeIpList, sizeof(SRpcIpSet)) != 0) {
dPrint("mnode ip list is changed, numOfIps:%d inUse:%d", pStatusRsp->ipList.numOfIps, pStatusRsp->ipList.inUse);
memcpy(&tsDnodeMnodeIpList, &pStatusRsp->ipList, sizeof(SRpcIpSet));
for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; ++i) {
dPrint("mnode index:%d ip:%s", i, taosIpStr(tsDnodeMnodeIpList.ip[i]));
}
dnodeSaveMnodeIpList();
}
SDnodeState *pState = &pStatusRsp->dnodeState;
pState->numOfVnodes = htonl(pState->numOfVnodes);
pState->moduleStatus = htonl(pState->moduleStatus);
pState->createdTime = htonl(pState->createdTime);
pState->dnodeId = htonl(pState->dnodeId);
dnodeProcessModuleStatus(pState->moduleStatus);
dnodeUpdateDnodeId(pState->dnodeId);
}
void dnodeSendMsgToMnode(SRpcMsg *rpcMsg) {
rpcSendRequest(tsDnodeMClientRpc, &tsDnodeMnodeIpList, rpcMsg);
}
static bool dnodeReadMnodeIpList() {
char ipFile[TSDB_FILENAME_LEN] = {0};
sprintf(ipFile, "%s/iplist", tsDnodeDir);
FILE *fp = fopen(ipFile, "r");
if (!fp) return false;
char option[32] = {0};
int32_t value = 0;
int32_t num = 0;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return false;
if (strcmp(option, "inUse") != 0) return false;
tsDnodeMnodeIpList.inUse = (int8_t)value;;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return false;
if (strcmp(option, "numOfIps") != 0) return false;
tsDnodeMnodeIpList.numOfIps = (int8_t)value;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return false;
if (strcmp(option, "port") != 0) return false;
tsDnodeMnodeIpList.port = (uint16_t)value;
for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return false;
if (strncmp(option, "ip", 2) != 0) return false;
tsDnodeMnodeIpList.ip[i] = (uint32_t)value;
}
fclose(fp);
dPrint("read mnode iplist successed");
for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
dPrint("mnode index:%d ip:%s", i, taosIpStr(tsDnodeMnodeIpList.ip[i]));
}
return true;
}
static void dnodeSaveMnodeIpList() {
char ipFile[TSDB_FILENAME_LEN] = {0};
sprintf(ipFile, "%s/iplist", tsDnodeDir);
FILE *fp = fopen(ipFile, "w");
if (!fp) return;
fprintf(fp, "inUse %d\n", tsDnodeMnodeIpList.inUse);
fprintf(fp, "numOfIps %d\n", tsDnodeMnodeIpList.numOfIps);
fprintf(fp, "port %u\n", tsDnodeMnodeIpList.port);
for (int32_t i = 0; i < tsDnodeMnodeIpList.numOfIps; i++) {
fprintf(fp, "ip%d %u\n", i, tsDnodeMnodeIpList.ip[i]);
}
fclose(fp);
dPrint("save mnode iplist successed");
}
uint32_t dnodeGetMnodeMasteIp() {
return tsDnodeMnodeIpList.ip[0];
}
\ No newline at end of file
src/dnode/src/dnodeMain.c
浏览文件 @ fcf29f14
......@@ -175,9 +175,9 @@ static int32_t dnodeInitSystem() {
if (dnodeInitModules() != 0) return -1;
if (dnodeInitRead() != 0) return -1;
if (dnodeInitWrite() != 0) return -1;
if (dnodeInitMgmt() != 0) return -1;
if (dnodeInitMnode() != 0) return -1;
if (dnodeInitMClient() != 0) return -1;
if (dnodeInitMnode() != 0) return -1;
if (dnodeInitMgmt() != 0) return -1;
if (dnodeInitShell() != 0) return -1;
dnodeStartModules();
......
src/dnode/src/dnodeMgmt.c
浏览文件 @ fcf29f14
......@@ -22,16 +22,18 @@
#include "trpc.h"
#include "tstatus.h"
#include "tsdb.h"
#include "ttime.h"
#include "ttimer.h"
#include "dnodeMClient.h"
#include "dnodeMgmt.h"
#include "dnodeRead.h"
#include "dnodeWrite.h"
typedef struct {
int32_t vgId; // global vnode group ID
int32_t vnode;
int32_t status; // status: master, slave, notready, deleting
int32_t refCount; // reference count
int8_t dirty;
int8_t status; // status: master, slave, notready, deleting
int64_t version;
void *wworker;
void *rworker;
......@@ -48,6 +50,7 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir);
static void dnodeCleanupVnode(SVnodeObj *pVnode);
static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *cfg);
static void dnodeDropVnode(SVnodeObj *pVnode);
static void dnodeDoDropVnode(SVnodeObj *pVnode);
static void dnodeProcessCreateVnodeMsg(SRpcMsg *pMsg);
static void dnodeProcessDropVnodeMsg(SRpcMsg *pMsg);
static void dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
......@@ -55,17 +58,23 @@ static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
static void dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
static void (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
static void dnodeSendStatusMsg(void *handle, void *tmrId);
static void dnodeReadDnodeId();
static void * tsDnodeVnodesHash = NULL;
static void *tsDnodeTmr = NULL;
static void *tsStatusTimer = NULL;
static void *tsDnodeVnodesHash = NULL;
static void *tsDnodeTmr = NULL;
static void *tsStatusTimer = NULL;
static uint32_t tsRebootTime;
static int32_t tsDnodeId = 0;
static char tsDnodeName[TSDB_DNODE_NAME_LEN];
int32_t dnodeInitMgmt() {
dnodeReadDnodeId();
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeProcessDropVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeProcessAlterVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj), taosHashInt);
if (tsDnodeVnodesHash == NULL) {
......@@ -73,6 +82,8 @@ int32_t dnodeInitMgmt() {
return -1;
}
tsRebootTime = taosGetTimestampSec();
tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
if (tsDnodeTmr == NULL) {
dError("failed to init dnode timer");
......@@ -80,13 +91,6 @@ int32_t dnodeInitMgmt() {
}
taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
SMDCreateVnodeMsg cfg;
cfg.cfg.vgId = 1;
cfg.cfg.precision = 0;
cfg.vnode = 1;
cfg.cfg.maxSessions = 1000;
cfg.cfg.daysPerFile = 10;
return dnodeOpenVnodes();
}
......@@ -96,8 +100,16 @@ void dnodeCleanupMgmt() {
tsStatusTimer = NULL;
}
if (tsDnodeTmr != NULL) {
taosTmrCleanUp(tsDnodeTmr);
tsDnodeTmr = NULL;
}
dnodeCleanupVnodes();
taosCleanUpIntHash(tsDnodeVnodesHash);
if (tsDnodeVnodesHash == NULL) {
taosCleanUpIntHash(tsDnodeVnodesHash);
tsDnodeVnodesHash = NULL;
}
}
void dnodeMgmt(SRpcMsg *pMsg) {
......@@ -113,7 +125,7 @@ void dnodeMgmt(SRpcMsg *pMsg) {
rpcSendResponse(&rsp);
}
rpcFreeCont(pMsg->pCont); // free the received message
rpcFreeCont(pMsg->pCont);
}
void *dnodeGetVnode(int32_t vgId) {
......@@ -152,8 +164,13 @@ void *dnodeGetVnodeTsdb(void *pVnode) {
return ((SVnodeObj *)pVnode)->tsdb;
}
void dnodeReleaseVnode(void *pVnode) {
atomic_sub_fetch_32(&((SVnodeObj *) pVnode)->refCount, 1);
void dnodeReleaseVnode(void *pVnodeRaw) {
SVnodeObj *pVnode = pVnodeRaw;
int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
if (count == 0 && pVnode->dirty) {
dnodeDoDropVnode(pVnode);
}
}
static int32_t dnodeOpenVnodes() {
......@@ -171,8 +188,8 @@ static int32_t dnodeOpenVnodes() {
int32_t vnode = atoi(de->d_name + 5);
if (vnode == 0) continue;
char vnodeDir[TSDB_FILENAME_LEN];
sprintf(vnodeDir, "%s/%s", tsVnodeDir, de->d_name);
char vnodeDir[TSDB_FILENAME_LEN * 3];
snprintf(vnodeDir, TSDB_FILENAME_LEN * 3, "%s/%s", tsVnodeDir, de->d_name);
int32_t code = dnodeOpenVnode(vnode, vnodeDir);
if (code == 0) {
numOfVnodes++;
......@@ -181,7 +198,7 @@ static int32_t dnodeOpenVnodes() {
}
closedir(dir);
dPrint("all vnodes is opened, num:%d", numOfVnodes);
dPrint("dnode mgmt is opened, vnodes:%d", numOfVnodes);
return TSDB_CODE_SUCCESS;
}
......@@ -189,7 +206,7 @@ typedef void (*CleanupFp)(char *);
static void dnodeCleanupVnodes() {
int32_t num = taosGetIntHashSize(tsDnodeVnodesHash);
taosCleanUpIntHashWithFp(tsDnodeVnodesHash, (CleanupFp)dnodeCleanupVnode);
dPrint("all vnodes is opened, num:%d", num);
dPrint("dnode mgmt is closed, vnodes:%d", num);
}
static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
......@@ -203,9 +220,8 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
//STsdbRepoInfo *tsdbInfo = tsdbGetStatus(pTsdb);
SVnodeObj vnodeObj;
vnodeObj.vgId = vnode; //tsdbInfo->tsdbCfg.vgId;
vnodeObj.vnode = vnode; //tsdbInfo->tsdbCfg.tsdbId;
SVnodeObj vnodeObj = {0};
vnodeObj.vgId = vnode;//tsdbInfo->tsdbCfg.tsdbId;
vnodeObj.status = TSDB_VN_STATUS_NOT_READY;
vnodeObj.refCount = 1;
vnodeObj.version = 0;
......@@ -219,7 +235,7 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
dTrace("open vnode:%d in %s", vnodeObj.vnode, rootDir);
dTrace("open vnode:%d in %s", vnodeObj.vgId, rootDir);
return TSDB_CODE_SUCCESS;
}
......@@ -248,14 +264,13 @@ static void dnodeCleanupVnode(SVnodeObj *pVnode) {
pVnode->tsdb = NULL;
}
dTrace("cleanup vnode:%d", pVnode->vnode);
dTrace("cleanup vnode:%d", pVnode->vgId);
}
static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
STsdbCfg tsdbCfg = {0};
tsdbCfg.vgId = pVnodeCfg->cfg.vgId;
tsdbCfg.precision = pVnodeCfg->cfg.precision;
tsdbCfg.tsdbId = pVnodeCfg->vnode;
tsdbCfg.tsdbId = pVnodeCfg->cfg.vgId;
tsdbCfg.maxTables = pVnodeCfg->cfg.maxSessions;
tsdbCfg.daysPerFile = pVnodeCfg->cfg.daysPerFile;
tsdbCfg.minRowsPerFileBlock = -1;
......@@ -290,13 +305,12 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
void *pTsdb = tsdbCreateRepo(rootDir, &tsdbCfg, NULL);
if (pTsdb == NULL) {
dError("vgroup:%d, failed to create tsdb in vnode:%d, reason:%s", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode, tstrerror(terrno));
dError("vgroup:%d, failed to create tsdb in vnode, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(terrno));
return terrno;
}
SVnodeObj vnodeObj;
SVnodeObj vnodeObj = {0};
vnodeObj.vgId = pVnodeCfg->cfg.vgId;
vnodeObj.vnode = pVnodeCfg->vnode;
vnodeObj.status = TSDB_VN_STATUS_NOT_READY;
vnodeObj.refCount = 1;
vnodeObj.version = 0;
......@@ -310,18 +324,11 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
dPrint("vgroup:%d, vnode:%d is created", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode);
dPrint("vgroup:%d, vnode:%d is created", vnodeObj.vgId, vnodeObj.vgId);
return TSDB_CODE_SUCCESS;
}
static void dnodeDropVnode(SVnodeObj *pVnode) {
pVnode->status = TSDB_VN_STATUS_NOT_READY;
int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
if (count > 0) {
// wait refcount
}
static void dnodeDoDropVnode(SVnodeObj *pVnode) {
if (pVnode->tsdb) {
tsdbDropRepo(pVnode->tsdb);
pVnode->tsdb = NULL;
......@@ -331,21 +338,33 @@ static void dnodeDropVnode(SVnodeObj *pVnode) {
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
}
static void dnodeDropVnode(SVnodeObj *pVnode) {
pVnode->status = TSDB_VN_STATUS_NOT_READY;
pVnode->dirty = true;
int32_t count = atomic_sub_fetch_32(&pVnode->refCount, 1);
if (count > 0) {
dTrace("vgroup:%d, vnode will be dropped until refcount:%d is 0", pVnode->vgId, count);
return;
}
dnodeDoDropVnode(pVnode);
}
static void dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
pCreate->vnode = htonl(pCreate->vnode);
pCreate->cfg.vgId = htonl(pCreate->cfg.vgId);
pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
dTrace("vgroup:%d, start to create vnode:%d", pCreate->cfg.vgId, pCreate->vnode);
dTrace("vgroup:%d, start to create vnode:%d in dnode", pCreate->cfg.vgId, pCreate->cfg.vgId);
SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
if (pVnodeObj != NULL) {
rpcRsp.code = TSDB_CODE_SUCCESS;
dPrint("vgroup:%d, vnode:%d is already exist", pCreate->cfg.vgId, pCreate->vnode);
dPrint("vgroup:%d, vnode is already exist", pCreate->cfg.vgId);
} else {
rpcRsp.code = dnodeCreateVnode(pCreate);
}
......@@ -359,6 +378,8 @@ static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
SMDDropVnodeMsg *pDrop = rpcMsg->pCont;
pDrop->vgId = htonl(pDrop->vgId);
dTrace("vgroup:%d, start to drop vnode in dnode", pDrop->vgId);
SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pDrop->vgId);
if (pVnodeObj != NULL) {
dnodeDropVnode(pVnodeObj);
......@@ -374,11 +395,12 @@ static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
pCreate->vnode = htonl(pCreate->vnode);
pCreate->cfg.vgId = htonl(pCreate->cfg.vgId);
pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
dTrace("vgroup:%d, start to alter vnode in dnode", pCreate->cfg.vgId);
SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
if (pVnodeObj != NULL) {
rpcRsp.code = TSDB_CODE_SUCCESS;
......@@ -401,66 +423,109 @@ static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
}
static void dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
// SCfgDnodeMsg *pCfg = (SCfgDnodeMsg *)pCont;
//
// int32_t code = tsCfgDynamicOptions(pCfg->config);
// dnodeSendRspToMnode(pConn, msgType + 1, code, NULL, 0);
SMDCfgDnodeMsg *pCfg = (SMDCfgDnodeMsg *)pMsg->pCont;
int32_t code = tsCfgDynamicOptions(pCfg->config);
SRpcMsg rpcRsp = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0, .code = code, .msgType = 0};
rpcSendResponse(&rpcRsp);
}
static void dnodeBuildVloadMsg(char *pNode, void * param) {
SVnodeObj *pVnode = (SVnodeObj *) pNode;
if (pVnode->dirty) return;
SDMStatusMsg *pStatus = param;
if (pStatus->openVnodes >= TSDB_MAX_VNODES) return;
SVnodeLoad *pLoad = &pStatus->load[pStatus->openVnodes++];
pLoad->vgId = htonl(pVnode->vgId);
pLoad->vnode = htonl(pVnode->vgId);
pLoad->status = pVnode->status;
}
static void dnodeSendStatusMsg(void *handle, void *tmrId) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
if (tsDnodeTmr == NULL) {
dError("dnode timer is already released");
return;
}
if (tsStatusTimer == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to start status timer");
return;
}
// int32_t contLen = sizeof(SDMStatusMsg) + dnodeGetVnodesNum() * sizeof(SVnodeLoad);
// SDMStatusMsg *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);
int32_t contLen = sizeof(SDMStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
SDMStatusMsg *pStatus = rpcMallocCont(contLen);
if (pStatus == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to malloc status message");
return;
}
strcpy(pStatus->dnodeName, tsDnodeName);
pStatus->version = htonl(tsVersion);
pStatus->dnodeId = htonl(tsDnodeId);
pStatus->privateIp = htonl(inet_addr(tsPrivateIp));
pStatus->publicIp = htonl(inet_addr(tsPublicIp));
pStatus->lastReboot = htonl(tsRebootTime);
pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
pStatus->numOfCores = htons((uint16_t) tsNumOfCores);
pStatus->diskAvailable = tsAvailDataDirGB;
pStatus->alternativeRole = (uint8_t) tsAlternativeRole;
taosVisitIntHashWithFp(tsDnodeVnodesHash, dnodeBuildVloadMsg, pStatus);
contLen = sizeof(SDMStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
pStatus->openVnodes = htons(pStatus->openVnodes);
SRpcMsg rpcMsg = {
.pCont = pStatus,
.contLen = contLen,
.msgType = TSDB_MSG_TYPE_DM_STATUS
};
dnodeSendMsgToMnode(&rpcMsg);
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
}
static void dnodeReadDnodeId() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "r");
if (!fp) return;
char option[32] = {0};
int32_t value = 0;
int32_t num = 0;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return;
if (strcmp(option, "dnodeId") != 0) return;
tsDnodeId = value;;
fclose(fp);
dPrint("read dnodeId:%d successed", tsDnodeId);
}
static void dnodeSaveDnodeId() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "w");
if (!fp) return;
fprintf(fp, "dnodeId %d\n", tsDnodeId);
fclose(fp);
dPrint("save dnodeId successed");
}
void dnodeUpdateDnodeId(int32_t dnodeId) {
if (tsDnodeId == 0) {
dPrint("dnodeId is set to %d", dnodeId);
tsDnodeId = dnodeId;
dnodeSaveDnodeId();
}
}
src/dnode/src/dnodeMnode.c
浏览文件 @ fcf29f14
......@@ -40,41 +40,42 @@ int32_t dnodeInitMnode() {
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
rpcInit.localPort = tsDnodeMnodePort;
rpcInit.label = "DND-mgmt";
rpcInit.label = "DND-MS";
rpcInit.numOfThreads = 1;
rpcInit.cfp = dnodeProcessMsgFromMnode;
rpcInit.sessions = TSDB_SESSIONS_PER_DNODE;
rpcInit.sessions = 100;
rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1500;
rpcInit.idleTime = tsShellActivityTimer * 2000;
tsDnodeMnodeRpc = rpcOpen(&rpcInit);
if (tsDnodeMnodeRpc == NULL) {
dError("failed to init connection from mgmt");
dError("failed to init mnode rpc server");
return -1;
}
dPrint("connection to mgmt is opened");
dPrint("mnode rpc server is opened");
return 0;
}
void dnodeCleanupMnode() {
if (tsDnodeMnodeRpc) {
rpcClose(tsDnodeMnodeRpc);
tsDnodeMnodeRpc = NULL;
dPrint("mnode rpc server is closed");
}
}
static void dnodeProcessMsgFromMnode(SRpcMsg *pMsg) {
SRpcMsg rspMsg;
rspMsg.handle = pMsg->handle;
rspMsg.pCont = NULL;
rspMsg.handle = pMsg->handle;
rspMsg.pCont = NULL;
rspMsg.contLen = 0;
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) {
rspMsg.code = TSDB_CODE_NOT_READY;
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
dTrace("conn:%p, query msg is ignored since dnode not running", pMsg->handle);
dTrace("thandle:%p, query msg is ignored since dnode not running", pMsg->handle);
return;
}
......
src/dnode/src/dnodeModule.c
浏览文件 @ fcf29f14
......@@ -74,7 +74,7 @@ int32_t dnodeInitModules() {
for (int mod = 0; mod < TSDB_MOD_MAX; ++mod) {
if (tsModule[mod].num != 0 && tsModule[mod].initFp) {
if ((*tsModule[mod].initFp)() != 0) {
dError("TDengine initialization failed");
dError("failed to init modules");
return -1;
}
}
......@@ -84,11 +84,44 @@ int32_t dnodeInitModules() {
}
void dnodeStartModules() {
for (int mod = 1; mod < TSDB_MOD_MAX; ++mod) {
if (tsModule[mod].num != 0 && tsModule[mod].startFp) {
if ((*tsModule[mod].startFp)() != 0) {
dError("failed to start module:%d", mod);
// for (int mod = 1; mod < TSDB_MOD_MAX; ++mod) {
// if (tsModule[mod].num != 0 && tsModule[mod].startFp) {
// if ((*tsModule[mod].startFp)() != 0) {
// dError("failed to start module:%d", mod);
// }
// }
// }
}
void dnodeProcessModuleStatus(uint32_t moduleStatus) {
if (moduleStatus == tsModuleStatus) return;
dPrint("module status is received, old:%d, new:%d", tsModuleStatus, moduleStatus);
int news = moduleStatus;
int olds = tsModuleStatus;
for (int moduleType = 0; moduleType < TSDB_MOD_MAX; ++moduleType) {
int newStatus = news & (1 << moduleType);
int oldStatus = olds & (1 << moduleType);
if (oldStatus > 0) {
if (newStatus == 0) {
if (tsModule[moduleType].stopFp) {
dPrint("module:%s is stopped on this node", tsModule[moduleType].name);
(*tsModule[moduleType].stopFp)();
}
}
} else if (oldStatus == 0) {
if (newStatus > 0) {
if (tsModule[moduleType].startFp) {
dPrint("module:%s is started on this node", tsModule[moduleType].name);
(*tsModule[moduleType].startFp)();
}
}
} else {
}
}
tsModuleStatus = moduleStatus;
}
src/dnode/src/dnodeShell.c
浏览文件 @ fcf29f14
......@@ -53,15 +53,15 @@ int32_t dnodeInitShell() {
rpcInit.sessions = TSDB_SESSIONS_PER_DNODE;
rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1500;
rpcInit.afp =
rpcInit.afp = dnodeRetrieveUserAuthInfo;
tsDnodeShellRpc = rpcOpen(&rpcInit);
if (tsDnodeShellRpc == NULL) {
dError("failed to init connection from shell");
dError("failed to init shell rpc server");
return -1;
}
dPrint("connection to shell is opened");
dPrint("shell rpc server is opened");
return 0;
}
......
src/dnode/src/dnodeWrite.c
浏览文件 @ fcf29f14
......@@ -94,7 +94,7 @@ void dnodeWrite(SRpcMsg *pMsg) {
SRpcContext *pRpcContext = NULL;
if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT || pMsg->msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
SMsgDesc *pDesc = pCont;
SMsgDesc *pDesc = (SMsgDesc *)pCont;
pDesc->numOfVnodes = htonl(pDesc->numOfVnodes);
pCont += sizeof(SMsgDesc);
if (pDesc->numOfVnodes > 1) {
......
src/inc/mnode.h
浏览文件 @ fcf29f14
......@@ -40,47 +40,62 @@ extern "C" {
#include "tutil.h"
typedef struct {
int32_t mnodeId;
uint32_t privateIp;
uint32_t publicIp;
int64_t createdTime;
int64_t lostTime;
uint64_t dbVersion;
uint32_t rack;
uint16_t idc;
uint16_t slot;
int8_t role;
int8_t status;
int8_t numOfMnodes;
int32_t numOfDnodes;
char mnodeName[TSDB_DNODE_NAME_LEN + 1];
char reserved[7];
char updateEnd[1];
int syncFd;
void *hbTimer;
void *pSync;
} SMnodeObj;
typedef struct {
int32_t dnodeId;
uint32_t privateIp;
int32_t sid;
uint32_t publicIp;
uint32_t moduleStatus;
int32_t openVnodes;
int32_t numOfVnodes;
int32_t numOfFreeVnodes;
int64_t createdTime;
uint32_t publicIp;
int32_t status;
uint32_t lastAccess;
uint32_t rebootTimes;
uint32_t lastReboot; // time stamp for last reboot
int32_t openVnodes;
int32_t numOfTotalVnodes; // from dnode status msg, config information
uint32_t rack;
uint16_t idc;
uint16_t slot;
uint16_t numOfCores; // from dnode status msg
uint8_t alternativeRole; // from dnode status msg, 0-any, 1-mgmt, 2-dnode
uint8_t reserveStatus;
uint16_t numOfTotalVnodes; // from dnode status msg, config information
uint16_t unused;
int8_t alternativeRole; // from dnode status msg, 0-any, 1-mgmt, 2-dnode
int8_t lbStatus; // set in balance function
float lbScore; // calc in balance function
int32_t customScore; // config by user
char dnodeName[TSDB_DNODE_NAME_LEN + 1];
char reserved[7];
char updateEnd[1];
SVnodeLoad vload[TSDB_MAX_VNODES];
int32_t status;
uint32_t lastReboot; // time stamp for last reboot
float diskAvailable; // from dnode status msg
int32_t bandwidthMb; // config by user
int16_t diskAvgUsage; // calc from sys.disk
int16_t cpuAvgUsage; // calc from sys.cpu
int16_t memoryAvgUsage; // calc from sys.mem
int16_t diskAvgUsage; // calc from sys.disk
int16_t bandwidthUsage; // calc from sys.band
uint32_t rack;
uint16_t idc;
uint16_t slot;
int32_t customScore; // config by user
float lbScore; // calc in balance function
int16_t lbStatus; // set in balance function
int16_t lastAllocVnode; // increase while create vnode
SVnodeLoad vload[TSDB_MAX_VNODES];
char reserved[16];
char updateEnd[1];
void * thandle;
} SDnodeObj;
typedef struct {
int32_t dnodeId;
uint32_t ip;
uint32_t publicIp;
int32_t vnode;
uint32_t privateIp;
uint32_t publicIp;
} SVnodeGid;
typedef struct {
......@@ -149,15 +164,13 @@ typedef struct _vg_obj {
uint32_t vgId;
char dbName[TSDB_DB_NAME_LEN + 1];
int64_t createdTime;
uint64_t lastCreate;
uint64_t lastRemove;
int32_t numOfVnodes;
SVnodeGid vnodeGid[TSDB_VNODES_SUPPORT];
int32_t numOfVnodes;
int32_t numOfTables;
int32_t lbIp;
int32_t lbTime;
int8_t lbStatus;
int8_t reserved[16];
int8_t reserved[14];
int8_t updateEnd[1];
struct _vg_obj *prev, *next;
void * idPool;
......@@ -169,8 +182,7 @@ typedef struct _db_obj {
int8_t dirty;
int64_t createdTime;
SDbCfg cfg;
int8_t dropStatus;
char reserved[16];
char reserved[15];
char updateEnd[1];
struct _db_obj *prev, *next;
int32_t numOfVgroups;
......
src/inc/taosdef.h
浏览文件 @ fcf29f14
......@@ -176,6 +176,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_MAX_COLUMNS 256
#define TSDB_MIN_COLUMNS 2 //PRIMARY COLUMN(timestamp) + other columns
#define TSDB_DNODE_NAME_LEN 63
#define TSDB_TABLE_NAME_LEN 64
#define TSDB_DB_NAME_LEN 32
#define TSDB_COL_NAME_LEN 64
......
src/inc/taosmsg.h
浏览文件 @ fcf29f14
......@@ -25,7 +25,6 @@ extern "C" {
#include "taosdef.h"
#include "taoserror.h"
#include "taosdef.h"
#include "trpc.h"
// message type
......@@ -520,21 +519,21 @@ typedef struct {
} SRetrieveTableRsp;
typedef struct {
uint32_t vnode;
uint32_t vgId;
uint8_t status;
uint8_t dropStatus;
uint8_t accessState;
int64_t totalStorage;
int64_t compStorage;
int64_t pointsWritten;
uint8_t syncStatus;
uint8_t reserved[15];
int32_t vgId;
int32_t vnode;
int64_t totalStorage;
int64_t compStorage;
int64_t pointsWritten;
uint8_t status;
uint8_t syncStatus;
uint8_t accessState;
uint8_t reserved[5];
} SVnodeLoad;
typedef struct {
uint32_t vnode;
char accessState;
uint8_t accessState;
uint8_t reserved[3];
} SVnodeAccess;
/*
......@@ -582,14 +581,16 @@ typedef struct {
} SVnodeStatisticInfo;
typedef struct {
int32_t dnodeId;
uint32_t moduleStatus;
uint32_t createdTime;
uint32_t numOfVnodes;
uint32_t reserved;
} SDnodeState;
typedef struct {
uint32_t version;
int32_t dnodeId;
char dnodeName[TSDB_DNODE_NAME_LEN];
uint32_t privateIp;
uint32_t publicIp;
uint32_t lastReboot; // time stamp for last reboot
......@@ -603,14 +604,12 @@ typedef struct {
} SDMStatusMsg;
typedef struct {
int32_t code;
SDnodeState dnodeState;
SRpcIpSet ipList;
SDnodeState dnodeState;
SVnodeAccess vnodeAccess[];
} SDMStatusRsp;
typedef struct {
int32_t vnode;
SVnodeCfg cfg;
SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
} SMDCreateVnodeMsg;
......
src/mnode/inc/mgmtBalance.h
浏览文件 @ fcf29f14
......@@ -23,6 +23,7 @@ extern "C" {
int32_t mgmtInitBalance();
void mgmtCleanupBalance();
void mgmtStartBalanceTimer(int32_t afterMs) ;
int32_t mgmtAllocVnodes(SVgObj *pVgroup);
#ifdef __cplusplus
......
src/mnode/inc/mgmtDServer.h
浏览文件 @ fcf29f14
......@@ -24,21 +24,6 @@ int32_t mgmtInitDServer();
void mgmtCleanupDServer();
void mgmtAddDServerMsgHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg));
//extern void *mgmtStatusTimer;
//
//void mgmtSendCreateTableMsg(SMDCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle);
//void mgmtSendDropTableMsg(SMDDropTableMsg *pRemove, SRpcIpSet *ipSet, void *ahandle);
//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle);
//void mgmtSendDropVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
//
//int32_t mgmtInitDnodeInt();
//void mgmtCleanUpDnodeInt();
//
//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 mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code);
#ifdef __cplusplus
}
#endif
......
src/mnode/inc/mgmtDnode.h
浏览文件 @ fcf29f14
......@@ -25,7 +25,8 @@ int32_t mgmtInitDnodes();
void mgmtCleanUpDnodes();
int32_t mgmtGetDnodesNum();
int32_t mgmtUpdateDnode(SDnodeObj *pDnode);
SDnodeObj* mgmtGetDnode(uint32_t ip);
SDnodeObj* mgmtGetDnode(int32_t dnodeId);
SDnodeObj* mgmtGetDnodeByIp(uint32_t ip);
bool mgmtCheckDnodeInRemoveState(SDnodeObj *pDnode);
bool mgmtCheckDnodeInOfflineState(SDnodeObj *pDnode);
......
src/mnode/inc/mgmtMnode.h
浏览文件 @ fcf29f14
......@@ -22,6 +22,11 @@ extern "C" {
bool mgmtCheckRedirect(void *handle);
void mgmtGetMnodeIpList(SRpcIpSet *ipSet);
int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp);
int32_t mgmtRemoveMnode(uint32_t privateIp);
#ifdef __cplusplus
}
#endif
......
src/mnode/inc/mgmtVgroup.h
浏览文件 @ fcf29f14
......@@ -32,15 +32,15 @@ SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
void mgmtCreateVgroup(SQueuedMsg *pMsg);
void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle);
void mgmtUpdateVgroup(SVgObj *pVgroup);
void mgmtUpdateVgroupIp(SDnodeObj *pDnode);
void mgmtSetVgroupIdPool();
SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb);
void mgmtAddTableIntoVgroup(SVgObj *pVgroup, STableInfo *pTable);
void mgmtRemoveTableFromVgroup(SVgObj *pVgroup, STableInfo *pTable);
SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode);
void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle);
void mgmtSendDropVnodeMsg(int32_t vgId, SRpcIpSet *ipSet, void *ahandle);
SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup);
SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip);
......
src/mnode/src/mgmtAcct.c
浏览文件 @ fcf29f14
......@@ -24,7 +24,7 @@ void (*mgmtCleanUpAcctsFp)() = NULL;
SAcctObj *(*mgmtGetAcctFp)(char *acctName) = NULL;
int32_t (*mgmtCheckUserLimitFp)(SAcctObj *pAcct) = NULL;
int32_t (*mgmtCheckDbLimitFp)(SAcctObj *pAcct) = NULL;
int32_t (*mgmtCheckTimeSeriesLimitFp)(SAcctObj *pAcct, int32_t numOfTimeSeries) = NULL;
int32_t (*mgmtCheckTableLimitFp)(SAcctObj *pAcct, int32_t numOfTimeSeries) = NULL;
int32_t mgmtAddDbIntoAcct(SAcctObj *pAcct, SDbObj *pDb) {
pthread_mutex_lock(&pAcct->mutex);
......@@ -137,8 +137,8 @@ int32_t mgmtCheckDbLimit(SAcctObj *pAcct) {
}
int32_t mgmtCheckTableLimit(SAcctObj *pAcct, int32_t numOfTimeSeries) {
if (mgmtCheckTimeSeriesLimitFp) {
return (*mgmtCheckTimeSeriesLimitFp)(pAcct, numOfTimeSeries);
if (mgmtCheckTableLimitFp) {
return (*mgmtCheckTableLimitFp)(pAcct, numOfTimeSeries);
} else {
return 0;
}
......
src/mnode/src/mgmtBalance.c
浏览文件 @ fcf29f14
......@@ -20,6 +20,7 @@
int32_t (*mgmtInitBalanceFp)() = NULL;
void (*mgmtCleanupBalanceFp)() = NULL;
void (*mgmtStartBalanceTimerFp)(int32_t afterMs) = NULL;
int32_t (*mgmtAllocVnodesFp)(SVgObj *pVgroup) = NULL;
int32_t mgmtInitBalance() {
......@@ -36,33 +37,28 @@ void mgmtCleanupBalance() {
}
}
void mgmtStartBalanceTimer(int32_t afterMs) {
if (mgmtStartBalanceTimerFp) {
(*mgmtStartBalanceTimerFp)(afterMs);
}
}
int32_t mgmtAllocVnodes(SVgObj *pVgroup) {
if (mgmtAllocVnodesFp) {
return mgmtAllocVnodesFp(pVgroup);
return (*mgmtAllocVnodesFp)(pVgroup);
}
SDnodeObj *pDnode = mgmtGetDnode(0);
SDnodeObj *pDnode = mgmtGetDnode(1);
if (pDnode == NULL) return TSDB_CODE_OTHERS;
int32_t selectedVnode = -1;
int32_t lastAllocVode = pDnode->lastAllocVnode;
for (int32_t i = 0; i < pDnode->numOfVnodes; i++) {
int32_t vnode = (i + lastAllocVode) % pDnode->numOfVnodes;
if (pDnode->vload[vnode].vgId == 0 && pDnode->vload[vnode].status == TSDB_VN_STATUS_OFFLINE) {
selectedVnode = vnode;
break;
}
}
if (selectedVnode == -1) {
mError("alloc vnode failed, free vnodes:%d", pDnode->numOfFreeVnodes);
return -1;
if (pDnode->openVnodes < pDnode->numOfTotalVnodes) {
pVgroup->vnodeGid[0].dnodeId = pDnode->dnodeId;
pVgroup->vnodeGid[0].privateIp = pDnode->privateIp;
pVgroup->vnodeGid[0].publicIp = pDnode->publicIp;
mTrace("dnode:%d, alloc one vnode to vgroup", pDnode->dnodeId);
return TSDB_CODE_SUCCESS;
} else {
mTrace("allocate vnode:%d, last allocated vnode:%d", selectedVnode, lastAllocVode);
pVgroup->vnodeGid[0].vnode = selectedVnode;
pDnode->lastAllocVnode = selectedVnode + 1;
if (pDnode->lastAllocVnode >= pDnode->numOfVnodes) pDnode->lastAllocVnode = 0;
return 0;
mError("dnode:%d, failed to alloc vnode to vgroup", pDnode->dnodeId);
return TSDB_CODE_NO_ENOUGH_DNODES;
}
}
src/mnode/src/mgmtChildTable.c
浏览文件 @ fcf29f14
......@@ -34,7 +34,7 @@
#include "mgmtTable.h"
#include "mgmtVgroup.h"
void *tsChildTableSdb;
void *tsChildTableSdb;
int32_t tsChildTableUpdateSize;
void *(*mgmtChildTableActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
......@@ -51,12 +51,12 @@ static void mgmtDestroyChildTable(SChildTableObj *pTable) {
}
static void mgmtChildTableActionInit() {
mgmtChildTableActionFp[SDB_TYPE_INSERT] = mgmtChildTableActionInsert;
mgmtChildTableActionFp[SDB_TYPE_DELETE] = mgmtChildTableActionDelete;
mgmtChildTableActionFp[SDB_TYPE_UPDATE] = mgmtChildTableActionUpdate;
mgmtChildTableActionFp[SDB_TYPE_ENCODE] = mgmtChildTableActionEncode;
mgmtChildTableActionFp[SDB_TYPE_DECODE] = mgmtChildTableActionDecode;
mgmtChildTableActionFp[SDB_TYPE_RESET] = mgmtChildTableActionReset;
mgmtChildTableActionFp[SDB_TYPE_INSERT] = mgmtChildTableActionInsert;
mgmtChildTableActionFp[SDB_TYPE_DELETE] = mgmtChildTableActionDelete;
mgmtChildTableActionFp[SDB_TYPE_UPDATE] = mgmtChildTableActionUpdate;
mgmtChildTableActionFp[SDB_TYPE_ENCODE] = mgmtChildTableActionEncode;
mgmtChildTableActionFp[SDB_TYPE_DECODE] = mgmtChildTableActionDecode;
mgmtChildTableActionFp[SDB_TYPE_RESET] = mgmtChildTableActionReset;
mgmtChildTableActionFp[SDB_TYPE_DESTROY] = mgmtChildTableActionDestroy;
}
......@@ -77,26 +77,26 @@ void *mgmtChildTableActionInsert(void *row, char *str, int32_t size, int32_t *ss
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
mError("ctable:%s, not in vgroup:%d", pTable->tableId, pTable->vgId);
return NULL;
}
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
mError("ctable:%s, vgroup:%d not in db:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return NULL;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("account not exists");
mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
return NULL;
}
if (!sdbMaster) {
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid != pTable->sid) {
mError("sid:%d is not matched from the master:%d", sid, pTable->sid);
mError("ctable:%s, sid:%d is not matched from the master:%d", pTable->tableId, sid, pTable->sid);
return NULL;
}
}
......@@ -128,13 +128,13 @@ void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ss
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) {
mError("vgroup:%d not in DB:%s", pVgroup->vgId, pVgroup->dbName);
mError("ctable:%s, vgroup:%d not in DB:%s", pTable->tableId, pVgroup->vgId, pVgroup->dbName);
return NULL;
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct == NULL) {
mError("account not exists");
mError("ctable:%s, account:%s not exists", pTable->tableId, pDb->cfg.acct);
return NULL;
}
......@@ -313,7 +313,7 @@ void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTab
void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
if (numOfTables >= tsMaxTables) {
mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
mError("ctable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
terrno = TSDB_CODE_TOO_MANY_TABLES;
return NULL;
}
......@@ -321,14 +321,14 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
char *pTagData = (char *) pCreate->schema; // it is a tag key
SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData);
if (pSuperTable == NULL) {
mError("table:%s, corresponding super table does not exist", pCreate->tableId);
mError("ctable:%s, corresponding super table does not exist", pCreate->tableId);
terrno = TSDB_CODE_INVALID_TABLE;
return NULL;
}
SChildTableObj *pTable = (SChildTableObj *) calloc(sizeof(SChildTableObj), 1);
if (pTable == NULL) {
mError("table:%s, failed to alloc memory", pCreate->tableId);
mError("ctable:%s, failed to alloc memory", pCreate->tableId);
terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL;
}
......@@ -345,25 +345,25 @@ void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t
if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
free(pTable);
mError("table:%s, update sdb error", pCreate->tableId);
mError("ctable:%s, update sdb error", pCreate->tableId);
terrno = TSDB_CODE_SDB_ERROR;
return NULL;
}
mTrace("table:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
mTrace("ctable:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
return pTable;
}
int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable) {
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
mError("table:%s, failed to drop child table, vgroup not exist", pTable->tableId);
mError("ctable:%s, failed to drop child table, vgroup not exist", pTable->tableId);
return TSDB_CODE_OTHERS;
}
SMDDropTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropTableMsg));
if (pDrop == NULL) {
mError("table:%s, failed to drop child table, no enough memory", pTable->tableId);
mError("ctable:%s, failed to drop child table, no enough memory", pTable->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
......@@ -375,7 +375,7 @@ int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable) {
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
mTrace("table:%s, send drop table msg", pDrop->tableId);
mTrace("ctable:%s, send drop table msg", pDrop->tableId);
SRpcMsg rpcMsg = {
.handle = newMsg,
.pCont = pDrop,
......@@ -395,6 +395,7 @@ void* mgmtGetChildTable(char *tableId) {
}
int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent) {
// TODO: send message to dnode
// int32_t col = mgmtFindSuperTableTagIndex(pTable->superTable, tagName);
// if (col < 0 || col > pTable->superTable->numOfTags) {
// return TSDB_CODE_APP_ERROR;
......@@ -462,7 +463,7 @@ int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *p
if (usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
} else {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].privateIp;
}
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
pMeta->vpeerDesc[i].vgId = htonl(pVgroup->vgId);
......
src/mnode/src/mgmtDClient.c
浏览文件 @ fcf29f14
......@@ -83,90 +83,3 @@ static void mgmtProcessRspFromDnode(SRpcMsg *rpcMsg) {
rpcFreeCont(rpcMsg->pCont);
}
//static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
// mTrace("drop table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg) {
// mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//
//static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
// mTrace("drop vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//static void mgmtProcessAlterVnodeRsp(SRpcMsg *rpcMsg) {
// mTrace("alter vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
// mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//static void mgmtProcessAlterStreamRsp(SRpcMsg *rpcMsg) {
// mTrace("alter stream rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//static void mgmtProcessConfigDnodeRsp(SRpcMsg *rpcMsg) {
// mTrace("config dnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//}
//
//
//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("table:%s, send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle);
//}
//
//void mgmtSendDropVnodeMsg(int32_t vgId, int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("vnode:%d send free vnode msg, ahandle:%p", vnode, ahandle);
// SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
// SRpcMsg rpcMsg = {
// .handle = ahandle,
// .pCont = pDrop,
// .contLen = pDrop ? sizeof(SMDDropVnodeMsg) : 0,
// .code = 0,
// .msgType = TSDB_MSG_TYPE_MD_DROP_VNODE
// };
// rpcSendRequest(tsMgmtDClientRpc, ipSet, &rpcMsg);
//}
//
////
////int32_t mgmtCfgDynamicOptions(SDnodeObj *pDnode, char *msg) {
//// char *option, *value;
//// int32_t olen, valen;
////
//// paGetToken(msg, &option, &olen);
//// if (strncasecmp(option, "unremove", 8) == 0) {
//// mgmtSetDnodeUnRemove(pDnode);
//// return TSDB_CODE_SUCCESS;
//// } else if (strncasecmp(option, "score", 5) == 0) {
//// paGetToken(option + olen + 1, &value, &valen);
//// if (valen > 0) {
//// int32_t score = atoi(value);
//// mTrace("dnode:%s, custom score set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->customScore, score);
//// pDnode->customScore = score;
//// mgmtUpdateDnode(pDnode);
//// //mgmtStartBalanceTimer(15);
//// }
//// return TSDB_CODE_INVALID_SQL;
//// } else if (strncasecmp(option, "bandwidth", 9) == 0) {
//// paGetToken(msg, &value, &valen);
//// if (valen > 0) {
//// int32_t bandwidthMb = atoi(value);
//// if (bandwidthMb >= 0 && bandwidthMb < 10000000) {
//// mTrace("dnode:%s, bandwidth(Mb) set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->bandwidthMb, bandwidthMb);
//// pDnode->bandwidthMb = bandwidthMb;
//// mgmtUpdateDnode(pDnode);
//// return TSDB_CODE_SUCCESS;
//// }
//// }
//// return TSDB_CODE_INVALID_SQL;
//// }
////
//// return -1;
////}
////
src/mnode/src/mgmtDServer.c
浏览文件 @ fcf29f14
......@@ -85,276 +85,3 @@ static void mgmtProcessMsgFromDnode(SRpcMsg *rpcMsg) {
static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
return TSDB_CODE_SUCCESS;
}
//
//
//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
// SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
// pCfg->dnode = htonl(pCfg->dnode);
// pCfg->vnode = htonl(pCfg->vnode);
// pCfg->sid = htonl(pCfg->sid);
// mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
//
// if (!sdbMaster) {
// mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
// return;
// }
//
// STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
// if (pTable == NULL) {
// mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
// return;
// }
//
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
//
// //TODO
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
// mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
//}
//
//static void mgmtProcessVnodeCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *pConn) {
// if (!sdbMaster) {
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
// return;
// }
//
// SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) pCont;
// pCfg->dnode = htonl(pCfg->dnode);
// pCfg->vnode = htonl(pCfg->vnode);
//
// SVgObj *pVgroup = mgmtGetVgroupByVnode(pCfg->dnode, pCfg->vnode);
// if (pVgroup == NULL) {
// mTrace("dnode:%s, vnode:%d, no vgroup info", taosIpStr(pCfg->dnode), pCfg->vnode);
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_NOT_ACTIVE_VNODE, NULL, 0);
// return;
// }
//
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
//
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
// mgmtSendCreateVnodeMsg(pVgroup, pCfg->vnode, &ipSet, NULL);
//}
//
//static void mgmtProcessCreateTableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
// mTrace("create table rsp received, thandle:%p code:%d", thandle, code);
// if (thandle == NULL) return;
//
// SProcessInfo *info = thandle;
// assert(info->type == TSDB_PROCESS_CREATE_TABLE || info->type == TSDB_PROCESS_CREATE_TABLE_GET_META);
// STableInfo *pTable = info->ahandle;
//
// if (code != TSDB_CODE_SUCCESS) {
// mError("table:%s, failed to create in dnode, code:%d, set it dirty", pTable->tableId);
// mgmtSetTableDirty(pTable, true);
// } else {
// mTrace("table:%s, created in dnode", pTable->tableId);
// mgmtSetTableDirty(pTable, false);
// }
//
// if (code != TSDB_CODE_SUCCESS) {
// SRpcMsg rpcMsg = {0};
// rpcMsg.code = code;
// rpcMsg.handle = info->thandle;
// rpcSendResponse(&rpcMsg);
// } else {
// if (info->type == TSDB_PROCESS_CREATE_TABLE_GET_META) {
// mTrace("table:%s, start to process get meta", pTable->tableId);
// mgmtProcessGetTableMeta(pTable, thandle);
// } else {
// SRpcMsg rpcMsg = {0};
// rpcMsg.code = code;
// rpcMsg.handle = info->thandle;
// rpcSendResponse(&rpcMsg);
// }
// }
//
// free(info);
//}
//
//static void mgmtProcessRemoveTableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
// mTrace("remove table rsp received, thandle:%p code:%d", thandle, code);
//}
//
//
//static void mgmtProcessDropVnodeRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
// mTrace("free vnode rsp received, thandle:%p code:%d", thandle, code);
//}
//
//static void mgmtProcessDropStableRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
// mTrace("drop stable rsp received, thandle:%p code:%d", thandle, code);
//}
//
//static void mgmtProcessCreateVnodeRsp(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle, int32_t code) {
// mTrace("create vnode rsp received, thandle:%p code:%d", thandle, code);
// if (thandle == NULL) return;
//
// SProcessInfo *info = thandle;
// assert(info->type == TSDB_PROCESS_CREATE_VGROUP || info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META);
// info->received++;
// SVgObj *pVgroup = info->ahandle;
//
// bool isGetMeta = false;
// if (info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META) {
// isGetMeta = true;
// }
//
// mTrace("vgroup:%d, received:%d numOfVnodes:%d", pVgroup->vgId, info->received, pVgroup->numOfVnodes);
// if (info->received == pVgroup->numOfVnodes) {
// mgmtProcessCreateTable(pVgroup, info->cont, info->contLen, info->thandle, isGetMeta);
// free(info);
// }
//}
//
//void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
// if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) {
// mError("invalid msg type:%d", msgType);
// return;
// }
//
// mTrace("msg:%d:%s is received from dnode, pConn:%p", msgType, taosMsg[(int8_t)msgType], pConn);
//
// if (msgType == TSDB_MSG_TYPE_DM_CONFIG_TABLE) {
// mgmtProcessTableCfgMsg(msgType, pCont, contLen, pConn);
// } else if (msgType == TSDB_MSG_TYPE_DM_CONFIG_VNODE) {
// mgmtProcessVnodeCfgMsg(msgType, pCont, contLen, pConn);
// } else if (msgType == TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP) {
// mgmtProcessCreateTableRsp(msgType, pCont, contLen, pConn, code);
// } else if (msgType == TSDB_MSG_TYPE_MD_DROP_TABLE_RSP) {
// mgmtProcessRemoveTableRsp(msgType, pCont, contLen, pConn, code);
// } else if (msgType == TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP) {
// mgmtProcessCreateVnodeRsp(msgType, pCont, contLen, pConn, code);
// } else if (msgType == TSDB_MSG_TYPE_MD_DROP_VNODE_RSP) {
// mgmtProcessDropVnodeRsp(msgType, pCont, contLen, pConn, code);
// } else if (msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
// mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code);
// } else if (msgType == TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP) {
// } else if (msgType == TSDB_MSG_TYPE_CM_ALTER_STREAM_RSP) {
// } else if (msgType == TSDB_MSG_TYPE_DM_STATUS) {
// mgmtProcessDnodeStatus(msgType, pCont, contLen, pConn, code);
// } else {
// mError("%s from dnode is not processed", taosMsg[(int8_t)msgType]);
// }
//
// //rpcFreeCont(pCont);
//}
//
//void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("table:%s, sid:%d send alter stream msg, ahandle:%p", pTable->tableId, pTable->sid, ahandle);
//}
//
//void mgmtSendDropVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("vnode:%d send free vnode msg, ahandle:%p", vnode, ahandle);
//
// SMDDropVnodeMsg *pFreeVnode = rpcMallocCont(sizeof(SMDDropVnodeMsg));
// if (pFreeVnode != NULL) {
// pFreeVnode->vnode = htonl(vnode);
// mgmtSendMsgToDnode(ipSet, TSDB_MSG_TYPE_MD_DROP_VNODE, pFreeVnode, sizeof(SMDDropVnodeMsg), ahandle);
// }
//}
//
//int32_t mgmtCfgDynamicOptions(SDnodeObj *pDnode, char *msg) {
// char *option, *value;
// int32_t olen, valen;
//
// paGetToken(msg, &option, &olen);
// if (strncasecmp(option, "unremove", 8) == 0) {
// mgmtSetDnodeUnRemove(pDnode);
// return TSDB_CODE_SUCCESS;
// } else if (strncasecmp(option, "score", 5) == 0) {
// paGetToken(option + olen + 1, &value, &valen);
// if (valen > 0) {
// int32_t score = atoi(value);
// mTrace("dnode:%s, custom score set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->customScore, score);
// pDnode->customScore = score;
// mgmtUpdateDnode(pDnode);
// //mgmtStartBalanceTimer(15);
// }
// return TSDB_CODE_INVALID_SQL;
// } else if (strncasecmp(option, "bandwidth", 9) == 0) {
// paGetToken(msg, &value, &valen);
// if (valen > 0) {
// int32_t bandwidthMb = atoi(value);
// if (bandwidthMb >= 0 && bandwidthMb < 10000000) {
// mTrace("dnode:%s, bandwidth(Mb) set from:%d to:%d", taosIpStr(pDnode->privateIp), pDnode->bandwidthMb, bandwidthMb);
// pDnode->bandwidthMb = bandwidthMb;
// mgmtUpdateDnode(pDnode);
// return TSDB_CODE_SUCCESS;
// }
// }
// return TSDB_CODE_INVALID_SQL;
// }
//
// return -1;
//}
//
//
//void mgmtCleanUpDnodeInt() {
// if (mgmtCleanUpDnodeIntFp) {
// mgmtCleanUpDnodeIntFp();
// }
//}
//
//void mgmtProcessDnodeStatus(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
// SDMStatusMsg *pStatus = (SDMStatusMsg *)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;
//// }
//
// pObj->status = TSDB_DN_STATUS_READY;
//
//// // wait vnode dropped
//// for (int32_t vnode = 0; vnode < pObj->numOfVnodes; ++vnode) {
//// SVnodeLoad *pVload = &(pObj->vload[vnode]);
//// if (pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) {
//// bool existInDnode = false;
//// for (int32_t j = 0; j < pObj->openVnodes; ++j) {
//// if (htonl(pStatus->load[j].vnode) == vnode) {
//// existInDnode = true;
//// break;
//// }
//// }
////
//// if (!existInDnode) {
//// pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
//// pVload->status = TSDB_VN_STATUS_OFFLINE;
//// mgmtUpdateDnode(pObj);
//// mPrint("dnode:%s, vid:%d, drop finished", taosIpStr(pObj->privateIp), vnode);
//// taosTmrStart(mgmtMonitorDbDrop, 10000, NULL, tsMgmtTmr);
//// }
//// } else if (pVload->vgId == 0) {
//// /*
//// * In some cases, vnode information may be reported abnormally, recover it
//// */
//// if (pVload->dropStatus != TSDB_VN_DROP_STATUS_READY || pVload->status != TSDB_VN_STATUS_OFFLINE) {
//// mPrint("dnode:%s, vid:%d, vgroup:%d status:%s dropStatus:%s, set it to avail status",
//// taosIpStr(pObj->privateIp), vnode, pVload->vgId, taosGetVnodeStatusStr(pVload->status),
//// taosGetVnodeDropStatusStr(pVload->dropStatus));
//// pVload->dropStatus = TSDB_VN_DROP_STATUS_READY;
//// pVload->status = TSDB_VN_STATUS_OFFLINE;
//// mgmtUpdateDnode(pObj);
//// }
//// }
//// }
//}
src/mnode/src/mgmtDb.c
浏览文件 @ fcf29f14
......@@ -34,7 +34,7 @@
#include "mgmtUser.h"
#include "mgmtVgroup.h"
static void *tsDbSdb = NULL;
static void *tsDbSdb = NULL;
static int32_t tsDbUpdateSize;
static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate);
......@@ -679,7 +679,7 @@ static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, pDb->dropStatus != TSDB_DB_STATUS_READY ? "dropping" : "ready");
strcpy(pWrite, pDb->dirty != TSDB_DB_STATUS_READY ? "dropping" : "ready");
cols++;
numOfRows++;
......
src/mnode/src/mgmtDnode.c
浏览文件 @ fcf29f14
......@@ -23,12 +23,14 @@
#include "mgmtDClient.h"
#include "mgmtMnode.h"
#include "mgmtShell.h"
#include "mgmtDServer.h"
#include "mgmtUser.h"
#include "mgmtVgroup.h"
int32_t (*mgmtInitDnodesFp)() = NULL;
void (*mgmtCleanUpDnodesFp)() = NULL;
SDnodeObj *(*mgmtGetDnodeFp)(uint32_t ip) = NULL;
SDnodeObj *(*mgmtGetDnodeByIpFp)(int32_t dnodeId) = NULL;
int32_t (*mgmtGetDnodesNumFp)() = NULL;
int32_t (*mgmtUpdateDnodeFp)(SDnodeObj *pDnode) = NULL;
void * (*mgmtGetNextDnodeFp)(SShowObj *pShow, SDnodeObj **pDnode) = NULL;
......@@ -44,80 +46,29 @@ static int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, vo
static int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg);
static void mgmtProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) ;
static void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg);
void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) {
int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore;
maxVnodes = maxVnodes > TSDB_MAX_VNODES ? TSDB_MAX_VNODES : maxVnodes;
maxVnodes = maxVnodes < TSDB_MIN_VNODES ? TSDB_MIN_VNODES : maxVnodes;
if (pDnode->numOfTotalVnodes != 0) {
maxVnodes = pDnode->numOfTotalVnodes;
if (pDnode->numOfTotalVnodes == 0) {
pDnode->numOfTotalVnodes = maxVnodes;
}
if (pDnode->alternativeRole == TSDB_DNODE_ROLE_MGMT) {
maxVnodes = 0;
pDnode->numOfTotalVnodes = 0;
}
pDnode->numOfVnodes = maxVnodes;
pDnode->numOfFreeVnodes = maxVnodes;
pDnode->openVnodes = 0;
pDnode->status = TSDB_DN_STATUS_OFFLINE;
}
void mgmtCalcNumOfFreeVnodes(SDnodeObj *pDnode) {
int32_t totalVnodes = 0;
mTrace("dnode:%s, begin calc free vnodes", taosIpStr(pDnode->privateIp));
for (int32_t i = 0; i < pDnode->numOfVnodes; ++i) {
SVnodeLoad *pVload = pDnode->vload + i;
if (pVload->vgId != 0) {
mTrace("%d-dnode:%s, calc free vnodes, exist vnode:%d, vgroup:%d, state:%d %s, dropstate:%d %s, syncstatus:%d %s",
totalVnodes, taosIpStr(pDnode->privateIp), i, pVload->vgId,
pVload->status, taosGetVnodeStatusStr(pVload->status),
pVload->dropStatus, taosGetVnodeDropStatusStr(pVload->dropStatus),
pVload->syncStatus, taosGetVnodeSyncStatusStr(pVload->syncStatus));
totalVnodes++;
}
}
pDnode->numOfFreeVnodes = pDnode->numOfVnodes - totalVnodes;
mTrace("dnode:%s, numOfVnodes:%d, numOfFreeVnodes:%d, totalVnodes:%d",
taosIpStr(pDnode->privateIp), pDnode->numOfVnodes, pDnode->numOfFreeVnodes, totalVnodes);
}
void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId) {
SDnodeObj *pDnode;
for (int32_t i = 0; i < numOfVnodes; ++i) {
pDnode = mgmtGetDnode(vnodeGid[i].ip);
if (pDnode) {
SVnodeLoad *pVload = pDnode->vload + vnodeGid[i].vnode;
memset(pVload, 0, sizeof(SVnodeLoad));
pVload->vnode = vnodeGid[i].vnode;
pVload->vgId = vgId;
mTrace("dnode:%s, vnode:%d add to vgroup:%d", taosIpStr(pDnode->privateIp), vnodeGid[i].vnode, pVload->vgId);
mgmtCalcNumOfFreeVnodes(pDnode);
} else {
mError("dnode:%s, not in dnode DB!!!", taosIpStr(vnodeGid[i].ip));
}
}
}
void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes) {
SDnodeObj *pDnode;
for (int32_t i = 0; i < numOfVnodes; ++i) {
pDnode = mgmtGetDnode(vnodeGid[i].ip);
if (pDnode) {
SVnodeLoad *pVload = pDnode->vload + vnodeGid[i].vnode;
mTrace("dnode:%s, vnode:%d remove from vgroup:%d", taosIpStr(vnodeGid[i].ip), vnodeGid[i].vnode, pVload->vgId);
memset(pVload, 0, sizeof(SVnodeLoad));
mgmtCalcNumOfFreeVnodes(pDnode);
} else {
mError("dnode:%s not in dnode DB!!!", taosIpStr(vnodeGid[i].ip));
}
}
mgmtUpdateDnode(pDnode);
}
bool mgmtCheckModuleInDnode(SDnodeObj *pDnode, int32_t moduleType) {
uint32_t status = pDnode->moduleStatus & (1 << moduleType);
return status > 0;
......@@ -318,12 +269,6 @@ static int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn)
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "vgid");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 12;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "status");
......@@ -342,11 +287,10 @@ static int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn)
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
// TODO: if other thread drop dnode ????
SDnodeObj *pDnode = NULL;
if (pShow->payloadLen > 0 ) {
uint32_t ip = ip2uint(pShow->payload);
pDnode = mgmtGetDnode(ip);
pDnode = mgmtGetDnodeByIp(ip);
if (NULL == pDnode) {
return TSDB_CODE_NODE_OFFLINE;
}
......@@ -399,10 +343,6 @@ static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, voi
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(uint32_t *)pWrite = pVnode->vnode;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(uint32_t *)pWrite = pVnode->vgId;
cols++;
......@@ -435,19 +375,22 @@ int32_t mgmtInitDnodes() {
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VNODES, mgmtGetVnodeMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VNODES, mgmtRetrieveVnodes);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONFIG_DNODE, mgmtProcessCfgDnodeMsg);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP, mgmtProcessCfgDnodeMsgRsp);
mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_STATUS, mgmtProcessDnodeStatusMsg);
if (mgmtInitDnodesFp) {
return mgmtInitDnodesFp();
} else {
tsDnodeObj.privateIp = inet_addr(tsPrivateIp);;
tsDnodeObj.dnodeId = 1;
tsDnodeObj.privateIp = inet_addr(tsPrivateIp);
tsDnodeObj.publicIp = inet_addr(tsPublicIp);
tsDnodeObj.createdTime = taosGetTimestampMs();
tsDnodeObj.lastReboot = taosGetTimestampSec();
tsDnodeObj.numOfTotalVnodes = tsNumOfTotalVnodes;
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;
tsDnodeObj.status = TSDB_DN_STATUS_OFFLINE;
tsDnodeObj.lastReboot = taosGetTimestampSec();
sprintf(tsDnodeObj.dnodeName, "%d", tsDnodeObj.dnodeId);
mgmtSetDnodeMaxVnodes(&tsDnodeObj);
tsDnodeObj.moduleStatus |= (1 << TSDB_MOD_MGMT);
......@@ -463,21 +406,30 @@ int32_t mgmtInitDnodes() {
void mgmtCleanUpDnodes() {
if (mgmtCleanUpDnodesFp) {
mgmtCleanUpDnodesFp();
(*mgmtCleanUpDnodesFp)();
}
}
SDnodeObj *mgmtGetDnode(uint32_t ip) {
SDnodeObj *mgmtGetDnode(int32_t dnodeId) {
if (mgmtGetDnodeFp) {
return mgmtGetDnodeFp(ip);
} else {
return (*mgmtGetDnodeFp)(dnodeId);
}
if (dnodeId == 1) {
return &tsDnodeObj;
}
return NULL;
}
SDnodeObj *mgmtGetDnodeByIp(uint32_t ip) {
if (mgmtGetDnodeByIpFp) {
return (*mgmtGetDnodeByIpFp)(ip);
}
return &tsDnodeObj;
}
int32_t mgmtGetDnodesNum() {
if (mgmtGetDnodesNumFp) {
return mgmtGetDnodesNumFp();
return (*mgmtGetDnodesNumFp)();
} else {
return 1;
}
......@@ -485,7 +437,7 @@ int32_t mgmtGetDnodesNum() {
int32_t mgmtUpdateDnode(SDnodeObj *pDnode) {
if (mgmtUpdateDnodeFp) {
return mgmtUpdateDnodeFp(pDnode);
return (*mgmtUpdateDnodeFp)(pDnode);
} else {
return 0;
}
......@@ -493,7 +445,7 @@ int32_t mgmtUpdateDnode(SDnodeObj *pDnode) {
void *mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode) {
if (mgmtGetNextDnodeFp) {
return mgmtGetNextDnodeFp(pShow, pDnode);
return (*mgmtGetNextDnodeFp)(pShow, pDnode);
} else {
if (*pDnode == NULL) {
*pDnode = &tsDnodeObj;
......@@ -507,14 +459,12 @@ void *mgmtGetNextDnode(SShowObj *pShow, SDnodeObj **pDnode) {
void mgmtSetDnodeUnRemove(SDnodeObj *pDnode) {
if (mgmtSetDnodeUnRemoveFp) {
mgmtSetDnodeUnRemoveFp(pDnode);
(*mgmtSetDnodeUnRemoveFp)(pDnode);
}
}
bool mgmtCheckConfigShow(SGlobalConfig *cfg) {
if (cfg->cfgType & TSDB_CFG_CTYPE_B_CLUSTER)
return false;
if (cfg->cfgType & TSDB_CFG_CTYPE_B_NOT_PRINT)
if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_SHOW))
return false;
return true;
}
......@@ -532,6 +482,11 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return;
SCMCfgDnodeMsg *pCmCfgDnode = pMsg->pCont;
if (pCmCfgDnode->ip[0] == 0) {
strcpy(pCmCfgDnode->ip, tsPrivateIp);
} else {
strcpy(pCmCfgDnode->ip, pCmCfgDnode->ip);
}
uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip);
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
......@@ -558,3 +513,107 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
rpcSendResponse(&rpcRsp);
}
static void mgmtProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) {
mTrace("cfg vnode rsp is received");
}
void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SDMStatusMsg *pStatus = rpcMsg->pCont;
pStatus->dnodeId = htonl(pStatus->dnodeId);
SDnodeObj *pDnode = NULL;
if (pStatus->dnodeId == 0) {
pDnode = mgmtGetDnodeByIp(htonl(pStatus->privateIp));
if (pDnode == NULL) {
mTrace("dnode not created, privateIp:%s", taosIpStr(htonl(pStatus->privateIp)));
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_DNODE_NOT_EXIST);
return;
}
} else {
pDnode = mgmtGetDnode(pStatus->dnodeId);
if (pDnode == NULL) {
mError("dnode:%d, not exist, privateIp:%s", taosIpStr(pStatus->dnodeId), pStatus->dnodeName);
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_DNODE_NOT_EXIST);
return;
}
}
uint32_t version = htonl(pStatus->version);
if (version != tsVersion) {
mError("dnode:%d, status msg version:%d not equal with mnode:%d", pDnode->dnodeId, version, tsVersion);
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_INVALID_MSG_VERSION);
return ;
}
uint32_t lastPrivateIp = pDnode->privateIp;
uint32_t lastPublicIp = pDnode->publicIp;
pDnode->privateIp = htonl(pStatus->privateIp);
pDnode->publicIp = htonl(pStatus->publicIp);
pDnode->lastReboot = htonl(pStatus->lastReboot);
pDnode->numOfTotalVnodes = htons(pStatus->numOfTotalVnodes);
pDnode->numOfCores = htons(pStatus->numOfCores);
pDnode->diskAvailable = pStatus->diskAvailable;
pDnode->alternativeRole = pStatus->alternativeRole;
if (pStatus->dnodeId == 0) {
mTrace("dnode:%d, first access, privateIp:%s, name:%s, ", pDnode->dnodeId, taosIpStr(pDnode->privateIp), pDnode->dnodeName);
mgmtSetDnodeMaxVnodes(pDnode);
}
if (lastPrivateIp != pDnode->privateIp || lastPublicIp != pDnode->publicIp) {
mgmtUpdateVgroupIp(pDnode);
//mgmtUpdateMnodeIp();
}
int32_t openVnodes = htons(pStatus->openVnodes);
for (int32_t j = 0; j < openVnodes; ++j) {
pDnode->vload[j].vgId = htonl(pStatus->load[j].vgId);
pDnode->vload[j].totalStorage = htobe64(pStatus->load[j].totalStorage);
pDnode->vload[j].compStorage = htobe64(pStatus->load[j].compStorage);
pDnode->vload[j].pointsWritten = htobe64(pStatus->load[j].pointsWritten);
SVgObj *pVgroup = mgmtGetVgroup(pDnode->vload[j].vgId);
if (pVgroup == NULL) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pDnode->privateIp);
mPrint("dnode:%d, vnode:%d not exist in mnode, drop it", pDnode->dnodeId, pDnode->vload[j].vgId);
mgmtSendDropVnodeMsg(pDnode->vload[j].vgId, &ipSet, NULL);
}
}
if (pDnode->status != TSDB_DN_STATUS_READY) {
mTrace("dnode:%d, from offline to online", pDnode->dnodeId);
pDnode->status = TSDB_DN_STATUS_READY;
mgmtStartBalanceTimer(200);
}
int32_t contLen = sizeof(SDMStatusRsp) + TSDB_MAX_VNODES * sizeof(SVnodeAccess);
SDMStatusRsp *pRsp = rpcMallocCont(contLen);
if (pRsp == NULL) {
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_SERV_OUT_OF_MEMORY);
return;
}
mgmtGetMnodeIpList(&pRsp->ipList);
pRsp->dnodeState.dnodeId = htonl(pDnode->dnodeId);
pRsp->dnodeState.moduleStatus = htonl(pDnode->moduleStatus);
pRsp->dnodeState.createdTime = htonl(pDnode->createdTime / 1000);
pRsp->dnodeState.numOfVnodes = 0;
contLen = sizeof(SDMStatusRsp);
//TODO: set vnode access
SRpcMsg rpcRsp = {
.handle = rpcMsg->handle,
.code = TSDB_CODE_SUCCESS,
.pCont = pRsp,
.contLen = contLen
};
rpcSendResponse(&rpcRsp);
}
src/mnode/src/mgmtMnode.c
浏览文件 @ fcf29f14
......@@ -15,8 +15,162 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "trpc.h"
#include "tschemautil.h"
#include "mgmtMnode.h"
#include "mgmtUser.h"
int32_t (*mgmtAddMnodeFp)(uint32_t privateIp, uint32_t publicIp) = NULL;
int32_t (*mgmtRemoveMnodeFp)(uint32_t privateIp) = NULL;
int32_t (*mgmtGetMnodesNumFp)() = NULL;
void * (*mgmtGetNextMnodeFp)(SShowObj *pShow, SMnodeObj **pMnode) = NULL;
static SMnodeObj tsMnodeObj = {0};
static int32_t mgmtGetMnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
bool mgmtCheckRedirect(void *handle) {
return false;
}
int32_t mgmtAddMnode(uint32_t privateIp, uint32_t publicIp) {
if (mgmtAddMnodeFp) {
return (*mgmtAddMnodeFp)(privateIp, publicIp);
} else {
return 0;
}
}
int32_t mgmtRemoveMnode(uint32_t privateIp) {
if (mgmtRemoveMnodeFp) {
return (*mgmtRemoveMnodeFp)(privateIp);
} else {
return 0;
}
}
static int32_t mgmtGetMnodesNum() {
if (mgmtGetMnodesNumFp) {
return (*mgmtGetMnodesNumFp)();
} else {
return 1;
}
}
static void *mgmtGetNextMnode(SShowObj *pShow, SMnodeObj **pMnode) {
if (mgmtGetNextMnodeFp) {
return (*mgmtGetNextMnodeFp)(pShow, pMnode);
} else {
if (*pMnode == NULL) {
*pMnode = &tsMnodeObj;
} else {
*pMnode = NULL;
}
}
return *pMnode;
}
static 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, "private ip");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create 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;
return 0;
}
static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
int32_t numOfRows = 0;
int32_t cols = 0;
SMnodeObj *pMnode = NULL;
char *pWrite;
char ipstr[32];
while (numOfRows < rows) {
pShow->pNode = mgmtGetNextMnode(pShow, (SMnodeObj **)&pMnode);
if (pMnode == NULL) break;
cols = 0;
tinet_ntoa(ipstr, pMnode->privateIp);
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, 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;
}
void mgmtGetMnodeIpList(SRpcIpSet *ipSet) {
ipSet->inUse = 0;
ipSet->port = htons(tsMnodeDnodePort);
ipSet->numOfIps = 1;
ipSet->ip[0] = htonl(inet_addr(tsMasterIp));
}
\ No newline at end of file
src/mnode/src/mgmtNormalTable.c
浏览文件 @ fcf29f14
......@@ -540,7 +540,7 @@ int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta
if (usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
} else {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].privateIp;
}
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
pMeta->vpeerDesc[i].vgId = htonl(pVgroup->vgId);
......
src/mnode/src/mgmtShell.c
浏览文件 @ fcf29f14
......@@ -392,7 +392,7 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
}
if (pConnectMsg->db[0]) {
char dbName[TSDB_TABLE_ID_LEN] = {0};
char dbName[TSDB_TABLE_ID_LEN * 3] = {0};
sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db);
SDbObj *pDb = mgmtGetDb(dbName);
if (pDb == NULL) {
......
src/mnode/src/mgmtSuperTable.c
浏览文件 @ fcf29f14
......@@ -65,8 +65,6 @@ static void mgmtSuperTableActionInit() {
mgmtSuperTableActionFp[SDB_TYPE_DECODE] = mgmtSuperTableActionDecode;
mgmtSuperTableActionFp[SDB_TYPE_RESET] = mgmtSuperTableActionReset;
mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy;
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mgmtGetShowSuperTableMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mgmtRetrieveShowSuperTables);
}
void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
......
src/mnode/src/mgmtTable.c
浏览文件 @ fcf29f14
......@@ -52,6 +52,8 @@ static void mgmtProcessMultiTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessSuperTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg);
static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg);
static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg);
static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg);
static int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
......@@ -84,8 +86,8 @@ int32_t mgmtInitTables() {
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP, mgmtProcessCreateTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_TABLE_RSP, mgmtProcessDropTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, NULL);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, NULL);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, mgmtProcessAlterTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, mgmtProcessDropStableRsp);
return TSDB_CODE_SUCCESS;
}
......@@ -109,17 +111,15 @@ STableInfo* mgmtGetTable(char *tableId) {
return NULL;
}
STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid) {
SDnodeObj *pObj = mgmtGetDnode(dnodeIp);
if (pObj != NULL && vnode >= 0 && vnode < pObj->numOfVnodes) {
int32_t vgId = pObj->vload[vnode].vgId;
SVgObj *pVgroup = mgmtGetVgroup(vgId);
if (pVgroup) {
return pVgroup->tableList[sid];
}
STableInfo* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_t sid) {
SDnodeObj *pObj = mgmtGetDnode(dnodeId);
SVgObj *pVgroup = mgmtGetVgroup(vnode);
if (pObj == NULL || pVgroup == NULL) {
return NULL;
}
return NULL;
return pVgroup->tableList[sid];
}
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp) {
......@@ -587,7 +587,7 @@ void mgmtProcessAlterTableMsg(SQueuedMsg *pMsg) {
void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) {
SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SDbObj* pDb = mgmtGetDbByTableId(pTable->tableId);
if (pDb == NULL || pDb->dropStatus != TSDB_DB_STATUS_READY) {
if (pDb == NULL || pDb->dirty) {
mError("table:%s, failed to get table meta, db not selected", pTable->tableId);
rpcRsp.code = TSDB_CODE_DB_NOT_SELECTED;
rpcSendResponse(&rpcRsp);
......@@ -783,6 +783,10 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
free(queueMsg);
}
static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg) {
mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
}
static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->handle == NULL) return;
......@@ -831,3 +835,37 @@ static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SUCCESS);
free(queueMsg);
}
static void mgmtProcessDropStableRsp(SRpcMsg *rpcMsg) {
mTrace("drop stable rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
}
//
//
//static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
// SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
// pCfg->dnode = htonl(pCfg->dnode);
// pCfg->vnode = htonl(pCfg->vnode);
// pCfg->sid = htonl(pCfg->sid);
// mTrace("dnode:%s, vnode:%d, sid:%d, receive table config msg", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
//
// if (!sdbMaster) {
// mError("dnode:%s, vnode:%d, sid:%d, not master, redirect it", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
// return;
// }
//
// STableInfo *pTable = mgmtGetTableByPos(pCfg->dnode, pCfg->vnode, pCfg->sid);
// if (pTable == NULL) {
// mError("dnode:%s, vnode:%d, sid:%d, table not found", taosIpStr(pCfg->dnode), pCfg->vnode, pCfg->sid);
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_INVALID_TABLE, NULL, 0);
// return;
// }
//
// mgmtSendRspToDnode(thandle, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
//
// //TODO
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
// mgmtSendCreateTableMsg(NULL, &ipSet, NULL);
//}
//
\ No newline at end of file
src/mnode/src/mgmtVgroup.c
浏览文件 @ fcf29f14
......@@ -111,11 +111,11 @@ int32_t mgmtInitVgroups() {
if (tsIsCluster && pVgroup->vnodeGid[0].publicIp == 0) {
pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp);
pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp);
pVgroup->vnodeGid[0].privateIp = inet_addr(tsPrivateIp);
sdbUpdateRow(tsVgroupSdb, pVgroup, tsVgUpdateSize, 1);
}
mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
// mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
}
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta);
......@@ -131,9 +131,6 @@ SVgObj *mgmtGetVgroup(int32_t vgId) {
return (SVgObj *)sdbGetRow(tsVgroupSdb, &vgId);
}
/*
* TODO: check if there is enough sids
*/
SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb) {
return pDb->pHead;
}
......@@ -162,13 +159,13 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg) {
pVgroup->idPool = taosInitIdPool(pDb->cfg.maxSessions);
mgmtAddVgroupIntoDb(pDb, pVgroup);
mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
// mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
sdbInsertRow(tsVgroupSdb, pVgroup, 0);
mPrint("vgroup:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
mPrint("vgroup:%d, dnode:%s vnode:%d", pVgroup->vgId, taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode);
mPrint("vgroup:%d, dnode:%d vnode:%d", pVgroup->vgId, pVgroup->vnodeGid[i].dnodeId, pVgroup->vnodeGid[i].vnode);
}
pMsg->ahandle = pVgroup;
......@@ -305,9 +302,9 @@ int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
}
char *mgmtGetVnodeStatus(SVgObj *pVgroup, SVnodeGid *pVnode) {
SDnodeObj *pDnode = mgmtGetDnode(pVnode->ip);
SDnodeObj *pDnode = mgmtGetDnode(pVnode->dnodeId);
if (pDnode == NULL) {
mError("dnode:%s, vgroup:%d, vnode:%d dnode not exist", taosIpStr(pVnode->ip), pVgroup->vgId, pVnode->vnode);
mError("vgroup:%d, not exist in dnode:%d", pVgroup->vgId, pDnode->dnodeId);
return "null";
}
......@@ -315,14 +312,13 @@ char *mgmtGetVnodeStatus(SVgObj *pVgroup, SVnodeGid *pVnode) {
return "offline";
}
SVnodeLoad *vload = pDnode->vload + pVnode->vnode;
if (vload->vgId != pVgroup->vgId || vload->vnode != pVnode->vnode) {
mError("dnode:%s, vgroup:%d, vnode:%d not same with dnode vgroup:%d vnode:%d",
taosIpStr(pVnode->ip), pVgroup->vgId, pVnode->vnode, vload->vgId, vload->vnode);
return "null";
for (int i = 0; i < pDnode->openVnodes; ++i) {
if (pDnode->vload[i].vgId == pVgroup->vgId) {
return (char*)taosGetVnodeStatusStr(pDnode->vload[i].status);
}
}
return (char*)taosGetVnodeStatusStr(vload->status);
return "null";
}
int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
......@@ -362,7 +358,7 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
cols++;
for (int32_t i = 0; i < maxReplica; ++i) {
tinet_ntoa(ipstr, pVgroup->vnodeGid[i].ip);
tinet_ntoa(ipstr, pVgroup->vnodeGid[i].privateIp);
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, ipstr);
cols++;
......@@ -372,7 +368,7 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
if (pVgroup->vnodeGid[i].ip != 0) {
if (pVgroup->vnodeGid[i].dnodeId != 0) {
char *vnodeStatus = mgmtGetVnodeStatus(pVgroup, pVgroup->vnodeGid + i);
strcpy(pWrite, vnodeStatus);
} else {
......@@ -394,6 +390,11 @@ int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pCo
}
static void *mgmtVgroupActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
SVgObj *pVgroup = row;
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
pVgroup->vnodeGid[i].vnode = pVgroup->vgId;
}
return NULL;
}
......@@ -405,7 +406,7 @@ static void *mgmtVgroupActionDelete(void *row, char *str, int32_t size, int32_t
mgmtRemoveVgroupFromDb(pDb, pVgroup);
}
mgmtUnSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes);
// mgmtUnSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes);
tfree(pVgroup->tableList);
return NULL;
......@@ -489,15 +490,14 @@ void mgmtRemoveTableFromVgroup(SVgObj *pVgroup, STableInfo *pTable) {
taosFreeId(pVgroup->idPool, pTable->sid);
}
SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup) {
SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) return NULL;
SMDCreateVnodeMsg *pVnode = rpcMallocCont(sizeof(SMDCreateVnodeMsg));
if (pVnode == NULL) return NULL;
pVnode->vnode = htonl(vnode);
pVnode->cfg = pDb->cfg;
pVnode->cfg = pDb->cfg;
SVnodeCfg *pCfg = &pVnode->cfg;
pCfg->vgId = htonl(pVgroup->vgId);
......@@ -516,8 +516,7 @@ SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
SVnodeDesc *vpeerDesc = pVnode->vpeerDesc;
for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {
vpeerDesc[j].vgId = htonl(pVgroup->vgId);
vpeerDesc[j].ip = htonl(pVgroup->vnodeGid[j].ip);
vpeerDesc[j].vnode = htonl(pVgroup->vnodeGid[j].vnode);
vpeerDesc[j].ip = htonl(pVgroup->vnodeGid[j].privateIp);
}
return pVnode;
......@@ -544,7 +543,7 @@ SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
.port = tsDnodeMnodePort
};
for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
ipSet.ip[i] = pVgroup->vnodeGid[i].ip;
ipSet.ip[i] = pVgroup->vnodeGid[i].privateIp;
}
return ipSet;
}
......@@ -559,9 +558,9 @@ SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip) {
return ipSet;
}
void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, vnode, ahandle);
SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup, vnode);
void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, pVgroup->vgId, ahandle);
SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup);
SRpcMsg rpcMsg = {
.handle = ahandle,
.pCont = pCreate,
......@@ -575,8 +574,8 @@ void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, vo
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
mTrace("vgroup:%d, send create all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].privateIp);
mgmtSendCreateVnodeMsg(pVgroup, &ipSet, ahandle);
}
}
......@@ -615,17 +614,17 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
free(queueMsg);
}
static SMDDropVnodeMsg *mgmtBuildDropVnodeMsg(SVgObj *pVgroup) {
static SMDDropVnodeMsg *mgmtBuildDropVnodeMsg(int32_t vgId) {
SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
if (pDrop == NULL) return NULL;
pDrop->vgId = htonl(pVgroup->vgId);
pDrop->vgId = htonl(vgId);
return pDrop;
}
static void mgmtSendDropVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
mTrace("vgroup:%d, send drop vnode msg, ahandle:%p", pVgroup->vgId, ahandle);
SMDDropVnodeMsg *pDrop = mgmtBuildDropVnodeMsg(pVgroup);
void mgmtSendDropVnodeMsg(int32_t vgId, SRpcIpSet *ipSet, void *ahandle) {
mTrace("vgroup:%d, send drop vnode msg, ahandle:%p", vgId, ahandle);
SMDDropVnodeMsg *pDrop = mgmtBuildDropVnodeMsg(vgId);
SRpcMsg rpcMsg = {
.handle = ahandle,
.pCont = pDrop,
......@@ -639,8 +638,8 @@ static void mgmtSendDropVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandl
static void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle) {
mTrace("vgroup:%d, send drop all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
mgmtSendDropVnodeMsg(pVgroup, &ipSet, ahandle);
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].privateIp);
mgmtSendDropVnodeMsg(pVgroup->vgId, &ipSet, ahandle);
}
}
......@@ -675,4 +674,49 @@ static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
mgmtAddToShellQueue(newMsg);
free(queueMsg);
}
\ No newline at end of file
}
void mgmtUpdateVgroupIp(SDnodeObj *pDnode) {
void * pNode = NULL;
SVgObj *pVgroup = NULL;
while (1) {
pNode = sdbFetchRow(tsVgroupSdb, pNode, (void **)&pVgroup);
if (pVgroup == NULL) break;
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SVnodeGid *vnodeGid = pVgroup->vnodeGid + i;
if (vnodeGid->dnodeId == pDnode->dnodeId) {
mPrint("vgroup:%d, dnode:%d, privateIp:%s change to %s, publicIp:%s change to %s",
pVgroup->vgId, vnodeGid->dnodeId, pDnode->privateIp, taosIpStr(vnodeGid->privateIp),
pDnode->publicIp, taosIpStr(vnodeGid->publicIp));
vnodeGid->publicIp = pDnode->publicIp;
vnodeGid->privateIp = pDnode->privateIp;
sdbUpdateRow(tsVgroupSdb, pVgroup, tsVgUpdateSize, 1);
}
}
}
}
//static void mgmtProcessVnodeCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *pConn) {
// if (!sdbMaster) {
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_REDIRECT, NULL, 0);
// return;
// }
//
// SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) pCont;
// pCfg->dnode = htonl(pCfg->dnode);
// pCfg->vnode = htonl(pCfg->vnode);
//
// SVgObj *pVgroup = mgmtGetVgroupByVnode(pCfg->dnode, pCfg->vnode);
// if (pVgroup == NULL) {
// mTrace("dnode:%s, vnode:%d, no vgroup info", taosIpStr(pCfg->dnode), pCfg->vnode);
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_NOT_ACTIVE_VNODE, NULL, 0);
// return;
// }
//
// mgmtSendRspToDnode(pConn, msgType + 1, TSDB_CODE_SUCCESS, NULL, 0);
//
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pCfg->dnode);
// mgmtSendCreateVnodeMsg(pVgroup, pCfg->vnode, &ipSet, NULL);
//}
//
\ No newline at end of file
src/util/inc/ihash.h
浏览文件 @ fcf29f14
......@@ -36,7 +36,7 @@ int32_t taosHashInt(void *handle, uint64_t key);
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *));
char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *));
void taosVisitIntHashWithFp(void *handle, void (*fp)(char *, void *), void *param);
int32_t taosGetIntHashSize(void *handle);
......
src/util/src/ihash.c
浏览文件 @ fcf29f14
......@@ -187,7 +187,6 @@ void taosCleanUpIntHash(void *handle) {
free(pObj);
}
void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
......@@ -202,7 +201,7 @@ void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
if (fp != NULL) fp(pNode->data);
if (fp != NULL) (*fp)(pNode->data);
free(pNode);
pNode = pNext;
}
......@@ -219,7 +218,7 @@ void taosCleanUpIntHashWithFp(void *handle, void (*fp)(char *)) {
free(pObj);
}
char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *)) {
void taosVisitIntHashWithFp(void *handle, int (*fp)(char *, void *), void *param) {
IHashObj * pObj;
IHashNode *pNode, *pNext;
char * pData = NULL;
......@@ -234,21 +233,13 @@ char *taosVisitIntHashWithFp(void *handle, int (*fp)(char *)) {
pNode = pObj->hashList[i];
while (pNode) {
pNext = pNode->next;
int flag = fp(pNode->data);
if (flag) {
pData = pNode->data;
goto VisitEnd;
}
(*fp)(pNode->data, param);
pNode = pNext;
}
}
}
VisitEnd:
pthread_mutex_unlock(&pObj->mutex);
return pData;
}
int32_t taosGetIntHashSize(void *handle) {
......
src/util/src/tglobalcfg.c
浏览文件 @ fcf29f14
......@@ -135,7 +135,7 @@ int tsOfflineThreshold = 864000; // seconds 10days
int tsMgmtEqualVnodeNum = 4;
int tsEnableHttpModule = 1;
int tsEnableMonitorModule = 1;
int tsEnableMonitorModule = 0;
int tsRestRowLimit = 10240;
int tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
......
src/util/src/tstatus.c
浏览文件 @ fcf29f14
......@@ -18,13 +18,13 @@
char* taosGetVgroupStatusStr(int32_t vgroupStatus) {
switch (vgroupStatus) {
case TSDB_VG_STATUS_READY: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_IN_PROGRESS: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_NO_DISK_PERMISSIONS: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_SERVER_NO_PACE: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_SERV_OUT_OF_MEMORY: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_INIT_FAILED: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_FULL: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_READY: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_IN_PROGRESS: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_NO_DISK_PERMISSIONS: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_SERVER_NO_PACE: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_SERV_OUT_OF_MEMORY: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_INIT_FAILED: return (char*)tstrerror(vgroupStatus);
case TSDB_VG_STATUS_FULL: return (char*)tstrerror(vgroupStatus);
default: return "undefined";
}
}
......
src/vnode/tsdb/inc/tsdb.h
浏览文件 @ fcf29f14
......@@ -35,7 +35,6 @@ extern "C" {
// --------- TSDB REPOSITORY CONFIGURATION DEFINITION
typedef struct {
int8_t precision;
int32_t vgId;
int32_t tsdbId;
int32_t maxTables; // maximum number of tables this repository can have
int32_t daysPerFile; // day per file sharding policy
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或