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

refactor: node mgmt

This commit is contained in:
Shengliang Guan 2022-05-11 23:43:29 +08:00
parent 65e8316f35
commit add51b499c
14 changed files with 585 additions and 650 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/tmsgcb.h
View File

@ -42,6 +42,7 @@ typedef int32_t (*GetQueueSizeFp)(void *pMgmt, int32_t vgId, EQueueType qtype);
typedef int32_t (*SendReqFp)(SMgmtWrapper* pWrapper, const SEpSet* epSet, SRpcMsg* pReq);
typedef int32_t (*SendMnodeReqFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
typedef void (*SendRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp);
typedef void (*SendMnodeRecvFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq, const SRpcMsg* pRsp);
typedef void (*SendRedirectRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
typedef void (*RegisterBrokenLinkArgFp)(SMgmtWrapper* pWrapper, SRpcMsg* pMsg);
typedef void (*ReleaseHandleFp)(SMgmtWrapper* pWrapper, void* handle, int8_t type);
@ -55,6 +56,7 @@ typedef struct {
GetQueueSizeFp qsizeFp;
SendReqFp sendReqFp;
SendRspFp sendRspFp;
SendMnodeRecvFp sendMnodeRecvFp;
SendRedirectRspFp sendRedirectRspFp;
RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
ReleaseHandleFp releaseHandleFp;
@ -66,6 +68,7 @@ int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq);
int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype);
int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq);
void tmsgSendRsp(const SRpcMsg* pRsp);
void tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp);
void tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg);
void tmsgReleaseHandle(void* handle, int8_t type);

9
source/common/src/tmsgcb.c
View File

@ -69,6 +69,15 @@ void tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
}
}
void tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp) {
SendMnodeRecvFp fp = tsDefaultMsgCb.sendMnodeRecvFp;
if (fp != NULL) {
(*fp)(tsDefaultMsgCb.pWrapper, pReq, pRsp);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}
void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg) {
RegisterBrokenLinkArgFp fp = pMsgCb->registerBrokenLinkArgFp;
if (fp != NULL) {

1
source/dnode/mgmt/CMakeLists.txt
View File

@ -5,6 +5,7 @@ add_subdirectory(mgmt_mnode)
add_subdirectory(mgmt_qnode)
add_subdirectory(mgmt_snode)
add_subdirectory(mgmt_vnode)
add_subdirectory(mgmt_dnode)
add_subdirectory(test)
aux_source_directory(exe EXEC_SRC)

9
source/dnode/mgmt/mgmt_dnode/CMakeLists.txt Normal file
View File

@ -0,0 +1,9 @@
aux_source_directory(src MGMT_DNODE)
add_library(mgmt_dnode STATIC ${MGMT_DNODE})
target_include_directories(
mgmt_dnode
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
mgmt_dnode node_util
)

69
source/dnode/mgmt/mgmt_dnode/inc/dmInt.h Normal file
View File

@ -0,0 +1,69 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_QNODE_INT_H_
#define _TD_DND_QNODE_INT_H_
#include "dmUtil.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SDnodeMgmt {
struct SDnode *pDnode;
SMsgCb msgCb;
const char *path;
const char *name;
TdThread *statusThreadId;
TdThread *monitorThreadId;
SSingleWorker mgmtWorker;
ProcessCreateNodeFp processCreateNodeFp;
ProcessDropNodeFp processDropNodeFp;
IsNodeDeployedFp isNodeDeployedFp;
SDnodeData data;
} SDnodeMgmt;
// dmEps.c
int32_t dmReadEps(SDnodeMgmt *pMgmt);
int32_t dmWriteEps(SDnodeMgmt *pMgmt);
void dmUpdateEps(SDnodeMgmt *pMgmt, SArray *pDnodeEps);
// dmHandle.c
SArray *dmGetMsgHandles();
void dmSendStatusReq(SDnodeMgmt *pMgmt);
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
// dmMonitor.c
void dmGetVnodeLoads(SDnodeMgmt *pMgmt, SMonVloadInfo *pInfo);
void dmGetMnodeLoads(SDnodeMgmt *pMgmt, SMonMloadInfo *pInfo);
void dmSendMonitorReport(SDnodeMgmt *pMgmt);
// dmWorker.c
int32_t dmPutNodeMsgToMgmtQueue(SDnodeMgmt *pMgmt, SNodeMsg *pMsg);
int32_t dmStartStatusThread(SDnodeMgmt *pMgmt);
void dmStopStatusThread(SDnodeMgmt *pMgmt);
int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt);
void dmStopMonitorThread(SDnodeMgmt *pMgmt);
int32_t dmStartWorker(SDnodeMgmt *pMgmt);
void dmStopWorker(SDnodeMgmt *pMgmt);
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_QNODE_INT_H_*/

117
source/dnode/mgmt/node_mgmt/src/dmEps.c → source/dnode/mgmt/mgmt_dnode/src/dmEps.c
View File

@ -14,16 +14,16 @@
*/
#define _DEFAULT_SOURCE
#include "dmImp.h"
#include "dmInt.h"
static void dmPrintEps(SDnode *pDnode);
static bool dmIsEpChanged(SDnode *pDnode, int32_t dnodeId, const char *ep);
static void dmResetEps(SDnode *pDnode, SArray *dnodeEps);
static void dmPrintEps(SDnodeMgmt *pMgmt);
static bool dmIsEpChanged(SDnodeMgmt *pMgmt, int32_t dnodeId, const char *ep);
static void dmResetEps(SDnodeMgmt *pMgmt, SArray *dnodeEps);
static void dmGetDnodeEp(SDnode *pDnode, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
taosRLockLatch(&pDnode->data.latch);
static void dmGetDnodeEp(SDnodeMgmt *pMgmt, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
taosRLockLatch(&pMgmt->data.latch);
SDnodeEp *pDnodeEp = taosHashGet(pDnode->data.dnodeHash, &dnodeId, sizeof(int32_t));
SDnodeEp *pDnodeEp = taosHashGet(pMgmt->data.dnodeHash, &dnodeId, sizeof(int32_t));
if (pDnodeEp != NULL) {
if (pPort != NULL) {
*pPort = pDnodeEp->ep.port;
@ -36,10 +36,10 @@ static void dmGetDnodeEp(SDnode *pDnode, int32_t dnodeId, char *pEp, char *pFqdn
}
}
taosRUnLockLatch(&pDnode->data.latch);
taosRUnLockLatch(&pMgmt->data.latch);
}
int32_t dmReadEps(SDnode *pDnode) {
int32_t dmReadEps(SDnodeMgmt *pMgmt) {
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
int32_t len = 0;
int32_t maxLen = 256 * 1024;
@ -48,16 +48,15 @@ int32_t dmReadEps(SDnode *pDnode) {
char file[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
pDnode->data.dnodeEps = taosArrayInit(1, sizeof(SDnodeEp));
if (pDnode->data.dnodeEps == NULL) {
pMgmt->data.dnodeEps = taosArrayInit(1, sizeof(SDnodeEp));
if (pMgmt->data.dnodeEps == NULL) {
dError("failed to calloc dnodeEp array since %s", strerror(errno));
goto _OVER;
}
snprintf(file, sizeof(file), "%s%sdnode.json", pDnode->wrappers[DNODE].path, TD_DIRSEP);
snprintf(file, sizeof(file), "%s%sdnode.json", pMgmt->path, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) {
// dDebug("file %s not exist", file);
code = 0;
goto _OVER;
}
@ -80,21 +79,21 @@ int32_t dmReadEps(SDnode *pDnode) {
dError("failed to read %s since dnodeId not found", file);
goto _OVER;
}
pDnode->data.dnodeId = dnodeId->valueint;
pMgmt->data.dnodeId = dnodeId->valueint;
cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
if (!clusterId || clusterId->type != cJSON_String) {
dError("failed to read %s since clusterId not found", file);
goto _OVER;
}
pDnode->data.clusterId = atoll(clusterId->valuestring);
pMgmt->data.clusterId = atoll(clusterId->valuestring);
cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_Number) {
dError("failed to read %s since dropped not found", file);
goto _OVER;
}
pDnode->data.dropped = dropped->valueint;
pMgmt->data.dropped = dropped->valueint;
cJSON *dnodes = cJSON_GetObjectItem(root, "dnodes");
if (!dnodes || dnodes->type != cJSON_Array) {
@ -144,29 +143,29 @@ int32_t dmReadEps(SDnode *pDnode) {
}
dnodeEp.isMnode = isMnode->valueint;
taosArrayPush(pDnode->data.dnodeEps, &dnodeEp);
taosArrayPush(pMgmt->data.dnodeEps, &dnodeEp);
}
code = 0;
dDebug("succcessed to read file %s", file);
dmPrintEps(pDnode);
dmPrintEps(pMgmt);
_OVER:
if (content != NULL) taosMemoryFree(content);
if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile);
if (taosArrayGetSize(pDnode->data.dnodeEps) == 0) {
if (taosArrayGetSize(pMgmt->data.dnodeEps) == 0) {
SDnodeEp dnodeEp = {0};
dnodeEp.isMnode = 1;
taosGetFqdnPortFromEp(pDnode->data.firstEp, &dnodeEp.ep);
taosArrayPush(pDnode->data.dnodeEps, &dnodeEp);
taosGetFqdnPortFromEp(pMgmt->data.firstEp, &dnodeEp.ep);
taosArrayPush(pMgmt->data.dnodeEps, &dnodeEp);
}
dmResetEps(pDnode, pDnode->data.dnodeEps);
dmResetEps(pMgmt, pMgmt->data.dnodeEps);
if (dmIsEpChanged(pDnode, pDnode->data.dnodeId, pDnode->data.localEp)) {
dError("localEp %s different with %s and need reconfigured", pDnode->data.localEp, file);
if (dmIsEpChanged(pMgmt, pMgmt->data.dnodeId, pMgmt->data.localEp)) {
dError("localEp %s different with %s and need reconfigured", pMgmt->data.localEp, file);
return -1;
}
@ -174,11 +173,11 @@ _OVER:
return code;
}
int32_t dmWriteEps(SDnode *pDnode) {
int32_t dmWriteEps(SDnodeMgmt *pMgmt) {
char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s%sdnode.json.bak", pDnode->wrappers[DNODE].path, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%sdnode.json", pDnode->wrappers[DNODE].path, TD_DIRSEP);
snprintf(file, sizeof(file), "%s%sdnode.json.bak", pMgmt->path, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%sdnode.json", pMgmt->path, TD_DIRSEP);
TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
@ -192,14 +191,14 @@ int32_t dmWriteEps(SDnode *pDnode) {
char *content = taosMemoryCalloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", pDnode->data.dnodeId);
len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%" PRId64 "\",\n", pDnode->data.clusterId);
len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pDnode->data.dropped);
len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", pMgmt->data.dnodeId);
len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%" PRId64 "\",\n", pMgmt->data.clusterId);
len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pMgmt->data.dropped);
len += snprintf(content + len, maxLen - len, " \"dnodes\": [{\n");
int32_t numOfEps = (int32_t)taosArrayGetSize(pDnode->data.dnodeEps);
int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->data.dnodeEps);
for (int32_t i = 0; i < numOfEps; ++i) {
SDnodeEp *pDnodeEp = taosArrayGet(pDnode->data.dnodeEps, i);
SDnodeEp *pDnodeEp = taosArrayGet(pMgmt->data.dnodeEps, i);
len += snprintf(content + len, maxLen - len, " \"id\": %d,\n", pDnodeEp->id);
len += snprintf(content + len, maxLen - len, " \"fqdn\": \"%s\",\n", pDnodeEp->ep.fqdn);
len += snprintf(content + len, maxLen - len, " \"port\": %u,\n", pDnodeEp->ep.port);
@ -223,41 +222,41 @@ int32_t dmWriteEps(SDnode *pDnode) {
return -1;
}
pDnode->data.updateTime = taosGetTimestampMs();
pMgmt->data.updateTime = taosGetTimestampMs();
dDebug("successed to write %s", realfile);
return 0;
}
void dmUpdateEps(SDnode *pDnode, SArray *eps) {
void dmUpdateEps(SDnodeMgmt *pMgmt, SArray *eps) {
int32_t numOfEps = taosArrayGetSize(eps);
if (numOfEps <= 0) return;
taosWLockLatch(&pDnode->data.latch);
taosWLockLatch(&pMgmt->data.latch);
int32_t numOfEpsOld = (int32_t)taosArrayGetSize(pDnode->data.dnodeEps);
int32_t numOfEpsOld = (int32_t)taosArrayGetSize(pMgmt->data.dnodeEps);
if (numOfEps != numOfEpsOld) {
dmResetEps(pDnode, eps);
dmWriteEps(pDnode);
dmResetEps(pMgmt, eps);
dmWriteEps(pMgmt);
} else {
int32_t size = numOfEps * sizeof(SDnodeEp);
if (memcmp(pDnode->data.dnodeEps->pData, eps->pData, size) != 0) {
dmResetEps(pDnode, eps);
dmWriteEps(pDnode);
if (memcmp(pMgmt->data.dnodeEps->pData, eps->pData, size) != 0) {
dmResetEps(pMgmt, eps);
dmWriteEps(pMgmt);
}
}
taosWUnLockLatch(&pDnode->data.latch);
taosWUnLockLatch(&pMgmt->data.latch);
}
static void dmResetEps(SDnode *pDnode, SArray *dnodeEps) {
if (pDnode->data.dnodeEps != dnodeEps) {
SArray *tmp = pDnode->data.dnodeEps;
pDnode->data.dnodeEps = taosArrayDup(dnodeEps);
static void dmResetEps(SDnodeMgmt *pMgmt, SArray *dnodeEps) {
if (pMgmt->data.dnodeEps != dnodeEps) {
SArray *tmp = pMgmt->data.dnodeEps;
pMgmt->data.dnodeEps = taosArrayDup(dnodeEps);
taosArrayDestroy(tmp);
}
pDnode->data.mnodeEps.inUse = 0;
pDnode->data.mnodeEps.numOfEps = 0;
pMgmt->data.mnodeEps.inUse = 0;
pMgmt->data.mnodeEps.numOfEps = 0;
int32_t mIndex = 0;
int32_t numOfEps = (int32_t)taosArrayGetSize(dnodeEps);
@ -266,35 +265,35 @@ static void dmResetEps(SDnode *pDnode, SArray *dnodeEps) {
SDnodeEp *pDnodeEp = taosArrayGet(dnodeEps, i);
if (!pDnodeEp->isMnode) continue;
if (mIndex >= TSDB_MAX_REPLICA) continue;
pDnode->data.mnodeEps.numOfEps++;
pMgmt->data.mnodeEps.numOfEps++;
pDnode->data.mnodeEps.eps[mIndex] = pDnodeEp->ep;
pMgmt->data.mnodeEps.eps[mIndex] = pDnodeEp->ep;
mIndex++;
}
for (int32_t i = 0; i < numOfEps; i++) {
SDnodeEp *pDnodeEp = taosArrayGet(dnodeEps, i);
taosHashPut(pDnode->data.dnodeHash, &pDnodeEp->id, sizeof(int32_t), pDnodeEp, sizeof(SDnodeEp));
taosHashPut(pMgmt->data.dnodeHash, &pDnodeEp->id, sizeof(int32_t), pDnodeEp, sizeof(SDnodeEp));
}
dmPrintEps(pDnode);
dmPrintEps(pMgmt);
}
static void dmPrintEps(SDnode *pDnode) {
int32_t numOfEps = (int32_t)taosArrayGetSize(pDnode->data.dnodeEps);
static void dmPrintEps(SDnodeMgmt *pMgmt) {
int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->data.dnodeEps);
dDebug("print dnode ep list, num:%d", numOfEps);
for (int32_t i = 0; i < numOfEps; i++) {
SDnodeEp *pEp = taosArrayGet(pDnode->data.dnodeEps, i);
SDnodeEp *pEp = taosArrayGet(pMgmt->data.dnodeEps, i);
dDebug("dnode:%d, fqdn:%s port:%u isMnode:%d", pEp->id, pEp->ep.fqdn, pEp->ep.port, pEp->isMnode);
}
}
static bool dmIsEpChanged(SDnode *pDnode, int32_t dnodeId, const char *ep) {
static bool dmIsEpChanged(SDnodeMgmt *pMgmt, int32_t dnodeId, const char *ep) {
bool changed = false;
if (dnodeId == 0) return changed;
taosRLockLatch(&pDnode->data.latch);
taosRLockLatch(&pMgmt->data.latch);
SDnodeEp *pDnodeEp = taosHashGet(pDnode->data.dnodeHash, &dnodeId, sizeof(int32_t));
SDnodeEp *pDnodeEp = taosHashGet(pMgmt->data.dnodeHash, &dnodeId, sizeof(int32_t));
if (pDnodeEp != NULL) {
char epstr[TSDB_EP_LEN + 1] = {0};
snprintf(epstr, TSDB_EP_LEN, "%s:%u", pDnodeEp->ep.fqdn, pDnodeEp->ep.port);
@ -304,6 +303,6 @@ static bool dmIsEpChanged(SDnode *pDnode, int32_t dnodeId, const char *ep) {
}
}
taosRUnLockLatch(&pDnode->data.latch);
taosRUnLockLatch(&pMgmt->data.latch);
return changed;
}

155
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c Normal file
View File

@ -0,0 +1,155 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http:www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dmInt.h"
static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) {
if (pMgmt->data.dnodeId == 0 || pMgmt->data.clusterId == 0) {
dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
taosWLockLatch(&pMgmt->data.latch);
pMgmt->data.dnodeId = pCfg->dnodeId;
pMgmt->data.clusterId = pCfg->clusterId;
dmWriteEps(pMgmt);
taosWUnLockLatch(&pMgmt->data.latch);
}
}
static int32_t dmProcessStatusRsp(SDnodeMgmt *pMgmt, SRpcMsg *pRsp) {
if (pRsp->code != TSDB_CODE_SUCCESS) {
if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pMgmt->data.dropped && pMgmt->data.dnodeId > 0) {
dInfo("dnode:%d, set to dropped since not exist in mnode", pMgmt->data.dnodeId);
pMgmt->data.dropped = 1;
dmWriteEps(pMgmt);
}
} else {
SStatusRsp statusRsp = {0};
if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
pMgmt->data.dnodeVer = statusRsp.dnodeVer;
dmUpdateDnodeCfg(pMgmt, &statusRsp.dnodeCfg);
dmUpdateEps(pMgmt, statusRsp.pDnodeEps);
}
rpcFreeCont(pRsp->pCont);
tFreeSStatusRsp(&statusRsp);
}
return TSDB_CODE_SUCCESS;
}
void dmSendStatusReq(SDnodeMgmt *pMgmt) {
SStatusReq req = {0};
taosRLockLatch(&pMgmt->data.latch);
req.sver = tsVersion;
req.dnodeVer = pMgmt->data.dnodeVer;
req.dnodeId = pMgmt->data.dnodeId;
req.clusterId = pMgmt->data.clusterId;
if (req.clusterId == 0) req.dnodeId = 0;
req.rebootTime = pMgmt->data.rebootTime;
req.updateTime = pMgmt->data.updateTime;
req.numOfCores = tsNumOfCores;
req.numOfSupportVnodes = pMgmt->data.supportVnodes;
tstrncpy(req.dnodeEp, pMgmt->data.localEp, TSDB_EP_LEN);
req.clusterCfg.statusInterval = tsStatusInterval;
req.clusterCfg.checkTime = 0;
char timestr[32] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &req.clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
taosRUnLockLatch(&pMgmt->data.latch);
SMonVloadInfo vinfo = {0};
dmGetVnodeLoads(pMgmt, &vinfo);
req.pVloads = vinfo.pVloads;
pMgmt->data.unsyncedVgId = 0;
pMgmt->data.vndState = TAOS_SYNC_STATE_LEADER;
for (int32_t i = 0; i < taosArrayGetSize(req.pVloads); ++i) {
SVnodeLoad *pLoad = taosArrayGet(req.pVloads, i);
if (pLoad->syncState != TAOS_SYNC_STATE_LEADER && pLoad->syncState != TAOS_SYNC_STATE_FOLLOWER) {
pMgmt->data.unsyncedVgId = pLoad->vgId;
pMgmt->data.vndState = pLoad->syncState;
}
}
SMonMloadInfo minfo = {0};
dmGetMnodeLoads(pMgmt, &minfo);
pMgmt->data.isMnode = minfo.isMnode;
pMgmt->data.mndState = minfo.load.syncState;
int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
void *pHead = rpcMallocCont(contLen);
tSerializeSStatusReq(pHead, contLen, &req);
tFreeSStatusReq(&req);
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
SRpcMsg rpcRsp = {0};
dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
tmsgSendMnodeRecv(&rpcMsg, &rpcRsp);
dmProcessStatusRsp(pMgmt, &rpcRsp);
}
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("auth rsp is received, but not supported yet");
return 0;
}
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("grant rsp is received, but not supported yet");
return 0;
}
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SRpcMsg *pReq = &pMsg->rpcMsg;
SDCfgDnodeReq *pCfg = pReq->pCont;
dError("config req is received, but not supported yet");
return TSDB_CODE_OPS_NOT_SUPPORT;
}
SArray *dmGetMsgHandles() {
int32_t code = -1;
SArray *pArray = taosArrayInit(16, sizeof(SMgmtHandle));
if (pArray == NULL) goto _OVER;
// Requests handled by DNODE
if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_MNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_QNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_QNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_SNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_SNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_BNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_BNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_DND_CONFIG_DNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
// Requests handled by MNODE
if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
code = 0;
_OVER:
if (code != 0) {
taosArrayDestroy(pArray);
return NULL;
} else {
return pArray;
}
}

104
source/dnode/mgmt/mgmt_dnode/src/dmInt.c Normal file
View File

@ -0,0 +1,104 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http:www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dmInt.h"
static int32_t dmStartMgmt(SDnodeMgmt *pMgmt) {
if (dmStartStatusThread(pMgmt) != 0) {
return -1;
}
if (dmStartMonitorThread(pMgmt) != 0) {
return -1;
}
return 0;
}
static void dmStopMgmt(SDnodeMgmt *pMgmt) {
dmStopMonitorThread(pMgmt);
dmStopStatusThread(pMgmt);
}
static int32_t dmOpenMgmt(const SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
dInfo("dnode-mgmt start to init");
SDnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SDnodeMgmt));
if (pMgmt == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMgmt->data.dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
if (pMgmt->data.dnodeHash == NULL) {
dError("failed to init dnode hash");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (dmReadEps(pMgmt) != 0) {
dError("failed to read file since %s", terrstr());
return -1;
}
if (pMgmt->data.dropped) {
dError("dnode will not start since its already dropped");
return -1;
}
if (dmStartWorker(pMgmt) != 0) {
return -1;
}
if (udfStartUdfd(pMgmt->data.dnodeId) != 0) {
dError("failed to start udfd");
}
dInfo("dnode-mgmt is initialized");
return 0;
}
static void dmCloseMgmt(SDnodeMgmt *pMgmt) {
dInfo("dnode-mgmt start to clean up");
dmStopWorker(pMgmt);
taosWLockLatch(&pMgmt->data.latch);
if (pMgmt->data.dnodeEps != NULL) {
taosArrayDestroy(pMgmt->data.dnodeEps);
pMgmt->data.dnodeEps = NULL;
}
if (pMgmt->data.dnodeHash != NULL) {
taosHashCleanup(pMgmt->data.dnodeHash);
pMgmt->data.dnodeHash = NULL;
}
taosWUnLockLatch(&pMgmt->data.latch);
dInfo("dnode-mgmt is cleaned up");
}
static int32_t dmRequireMgmt(const SMgmtInputOpt *pInput, bool *required) {
*required = true;
return 0;
}
SMgmtFunc dmGetMgmtFunc() {
SMgmtFunc mgmtFunc = {0};
mgmtFunc.openFp = dmOpenMgmt;
mgmtFunc.closeFp = (NodeCloseFp)dmCloseMgmt;
mgmtFunc.startFp = (NodeStartFp)dmStartMgmt;
mgmtFunc.stopFp = (NodeStopFp)dmStopMgmt;
mgmtFunc.requiredFp = dmRequireMgmt;
mgmtFunc.getHandlesFp = dmGetMsgHandles;
return mgmtFunc;
}

105
source/dnode/mgmt/mgmt_dnode/src/dmMonitor.c Normal file
View File

@ -0,0 +1,105 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dmInt.h"
#define dmSendLocalRecv(pMgmt, mtype, func, pInfo) \
{ \
SRpcMsg rsp = {0}; \
SRpcMsg req = {.msgType = mtype}; \
SEpSet epset = {.inUse = 0, .numOfEps = 1}; \
tstrncpy(epset.eps[0].fqdn, pMgmt->data.localFqdn, TSDB_FQDN_LEN); \
epset.eps[0].port = pMgmt->data.serverPort; \
\
rpcSendRecv(pMgmt->msgCb.clientRpc, &epset, &req, &rsp); \
if (rsp.code == 0 && rsp.contLen > 0) { \
func(rsp.pCont, rsp.contLen, pInfo); \
} \
rpcFreeCont(rsp.pCont); \
}
static void dmGetMonitorBasicInfo(SDnodeMgmt *pMgmt, SMonBasicInfo *pInfo) {
pInfo->protocol = 1;
pInfo->dnode_id = pMgmt->data.dnodeId;
pInfo->cluster_id = pMgmt->data.clusterId;
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
}
static void dmGetMonitorDnodeInfo(SDnodeMgmt *pMgmt, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pMgmt->data.rebootTime) / (86400000.0f);
pInfo->has_mnode = (*pMgmt->isNodeDeployedFp)(pMgmt->pDnode, MNODE);
pInfo->has_qnode = (*pMgmt->isNodeDeployedFp)(pMgmt->pDnode, QNODE);
pInfo->has_snode = (*pMgmt->isNodeDeployedFp)(pMgmt->pDnode, SNODE);
pInfo->has_bnode = (*pMgmt->isNodeDeployedFp)(pMgmt->pDnode, BNODE);
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size;
}
static void dmGetMonitorInfo(SDnodeMgmt *pMgmt, SMonDmInfo *pInfo) {
dmGetMonitorBasicInfo(pMgmt, &pInfo->basic);
dmGetMonitorDnodeInfo(pMgmt, &pInfo->dnode);
dmGetMonitorSystemInfo(&pInfo->sys);
}
void dmSendMonitorReport(SDnodeMgmt *pMgmt) {
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
dTrace("send monitor report to %s:%u", tsMonitorFqdn, tsMonitorPort);
SMonDmInfo dmInfo = {0};
SMonMmInfo mmInfo = {0};
SMonVmInfo vmInfo = {0};
SMonQmInfo qmInfo = {0};
SMonSmInfo smInfo = {0};
SMonBmInfo bmInfo = {0};
dmGetMonitorInfo(pMgmt, &dmInfo);
dmSendLocalRecv(pMgmt, TDMT_MON_VM_INFO, tDeserializeSMonVmInfo, &vmInfo);
if (dmInfo.dnode.has_mnode) {
dmSendLocalRecv(pMgmt, TDMT_MON_MM_INFO, tDeserializeSMonMmInfo, &mmInfo);
}
if (dmInfo.dnode.has_qnode) {
dmSendLocalRecv(pMgmt, TDMT_MON_QM_INFO, tDeserializeSMonQmInfo, &qmInfo);
}
if (dmInfo.dnode.has_snode) {
dmSendLocalRecv(pMgmt, TDMT_MON_SM_INFO, tDeserializeSMonSmInfo, &smInfo);
}
if (dmInfo.dnode.has_bnode) {
dmSendLocalRecv(pMgmt, TDMT_MON_BM_INFO, tDeserializeSMonBmInfo, &bmInfo);
}
monSetDmInfo(&dmInfo);
monSetMmInfo(&mmInfo);
monSetVmInfo(&vmInfo);
monSetQmInfo(&qmInfo);
monSetSmInfo(&smInfo);
monSetBmInfo(&bmInfo);
tFreeSMonMmInfo(&mmInfo);
tFreeSMonVmInfo(&vmInfo);
tFreeSMonQmInfo(&qmInfo);
tFreeSMonSmInfo(&smInfo);
tFreeSMonBmInfo(&bmInfo);
monSendReport();
}
void dmGetVnodeLoads(SDnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
dmSendLocalRecv(pMgmt, TDMT_MON_VM_LOAD, tDeserializeSMonVloadInfo, pInfo);
}
void dmGetMnodeLoads(SDnodeMgmt *pMgmt, SMonMloadInfo *pInfo) {
dmSendLocalRecv(pMgmt, TDMT_MON_MM_LOAD, tDeserializeSMonMloadInfo, pInfo);
}

84
source/dnode/mgmt/node_mgmt/src/dmWorker.c → source/dnode/mgmt/mgmt_dnode/src/dmWorker.c
View File

@ -14,10 +14,10 @@
*/
#define _DEFAULT_SOURCE
#include "dmImp.h"
#include "dmInt.h"
static void *dmStatusThreadFp(void *param) {
SDnode *pDnode = param;
SDnodeMgmt *pMgmt = param;
int64_t lastTime = taosGetTimestampMs();
setThreadName("dnode-status");
@ -26,14 +26,14 @@ static void *dmStatusThreadFp(void *param) {
taosThreadTestCancel();
taosMsleep(200);
if (pDnode->status != DND_STAT_RUNNING || pDnode->data.dropped) {
if (pMgmt->data.status != DND_STAT_RUNNING || pMgmt->data.dropped) {
continue;
}
int64_t curTime = taosGetTimestampMs();
float interval = (curTime - lastTime) / 1000.0f;
if (interval >= tsStatusInterval) {
dmSendStatusReq(pDnode);
dmSendStatusReq(pMgmt);
lastTime = curTime;
}
}
@ -42,7 +42,7 @@ static void *dmStatusThreadFp(void *param) {
}
static void *dmMonitorThreadFp(void *param) {
SDnode *pDnode = param;
SDnodeMgmt *pMgmt = param;
int64_t lastTime = taosGetTimestampMs();
setThreadName("dnode-monitor");
@ -51,14 +51,14 @@ static void *dmMonitorThreadFp(void *param) {
taosThreadTestCancel();
taosMsleep(200);
if (pDnode->status != DND_STAT_RUNNING || pDnode->data.dropped) {
if (pMgmt->data.status != DND_STAT_RUNNING || pMgmt->data.dropped) {
continue;
}
int64_t curTime = taosGetTimestampMs();
float interval = (curTime - lastTime) / 1000.0f;
if (interval >= tsMonitorInterval) {
dmSendMonitorReport(pDnode);
dmSendMonitorReport(pMgmt);
lastTime = curTime;
}
}
@ -66,46 +66,46 @@ static void *dmMonitorThreadFp(void *param) {
return NULL;
}
int32_t dmStartStatusThread(SDnode *pDnode) {
pDnode->data.statusThreadId = taosCreateThread(dmStatusThreadFp, pDnode);
if (pDnode->data.statusThreadId == NULL) {
int32_t dmStartStatusThread(SDnodeMgmt *pMgmt) {
pMgmt->statusThreadId = taosCreateThread(dmStatusThreadFp, pMgmt);
if (pMgmt->statusThreadId == NULL) {
dError("failed to init dnode status thread");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
dmReportStartup(pDnode, "dnode-status", "initialized");
tmsgReportStartup("dnode-status", "initialized");
return 0;
}
void dmStopStatusThread(SDnode *pDnode) {
if (pDnode->data.statusThreadId != NULL) {
taosDestoryThread(pDnode->data.statusThreadId);
pDnode->data.statusThreadId = NULL;
void dmStopStatusThread(SDnodeMgmt *pMgmt) {
if (pMgmt->statusThreadId != NULL) {
taosDestoryThread(pMgmt->statusThreadId);
pMgmt->statusThreadId = NULL;
}
}
int32_t dmStartMonitorThread(SDnode *pDnode) {
pDnode->data.monitorThreadId = taosCreateThread(dmMonitorThreadFp, pDnode);
if (pDnode->data.monitorThreadId == NULL) {
int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt) {
pMgmt->monitorThreadId = taosCreateThread(dmMonitorThreadFp, pMgmt);
if (pMgmt->monitorThreadId == NULL) {
dError("failed to init dnode monitor thread");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
dmReportStartup(pDnode, "dnode-monitor", "initialized");
tmsgReportStartup("dnode-monitor", "initialized");
return 0;
}
void dmStopMonitorThread(SDnode *pDnode) {
if (pDnode->data.monitorThreadId != NULL) {
taosDestoryThread(pDnode->data.monitorThreadId);
pDnode->data.monitorThreadId = NULL;
void dmStopMonitorThread(SDnodeMgmt *pMgmt) {
if (pMgmt->monitorThreadId != NULL) {
taosDestoryThread(pMgmt->monitorThreadId);
pMgmt->monitorThreadId = NULL;
}
}
static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SDnode *pDnode = pInfo->ahandle;
SDnodeMgmt *pMgmt = pInfo->ahandle;
int32_t code = -1;
tmsg_t msgType = pMsg->rpcMsg.msgType;
@ -113,37 +113,37 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
switch (msgType) {
case TDMT_DND_CONFIG_DNODE:
code = dmProcessConfigReq(pDnode, pMsg);
code = dmProcessConfigReq(pMgmt, pMsg);
break;
case TDMT_MND_AUTH_RSP:
code = dmProcessAuthRsp(pDnode, pMsg);
code = dmProcessAuthRsp(pMgmt, pMsg);
break;
case TDMT_MND_GRANT_RSP:
code = dmProcessGrantRsp(pDnode, pMsg);
code = dmProcessGrantRsp(pMgmt, pMsg);
break;
case TDMT_DND_CREATE_MNODE:
code = dmProcessCreateNodeReq(pDnode, MNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, MNODE, pMsg);
break;
case TDMT_DND_DROP_MNODE:
code = dmProcessDropNodeReq(pDnode, MNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, MNODE, pMsg);
break;
case TDMT_DND_CREATE_QNODE:
code = dmProcessCreateNodeReq(pDnode, QNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, QNODE, pMsg);
break;
case TDMT_DND_DROP_QNODE:
code = dmProcessDropNodeReq(pDnode, QNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, QNODE, pMsg);
break;
case TDMT_DND_CREATE_SNODE:
code = dmProcessCreateNodeReq(pDnode, SNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, SNODE, pMsg);
break;
case TDMT_DND_DROP_SNODE:
code = dmProcessDropNodeReq(pDnode, SNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, SNODE, pMsg);
break;
case TDMT_DND_CREATE_BNODE:
code = dmProcessCreateNodeReq(pDnode, BNODE, pMsg);
code = (*pMgmt->processCreateNodeFp)(pMgmt->pDnode, BNODE, pMsg);
break;
case TDMT_DND_DROP_BNODE:
code = dmProcessDropNodeReq(pDnode, BNODE, pMsg);
code = (*pMgmt->processDropNodeFp)(pMgmt->pDnode, BNODE, pMsg);
break;
default:
break;
@ -165,15 +165,15 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
taosFreeQitem(pMsg);
}
int32_t dmStartWorker(SDnode *pDnode) {
int32_t dmStartWorker(SDnodeMgmt *pMgmt) {
SSingleWorkerCfg cfg = {
.min = 1,
.max = 1,
.name = "dnode-mgmt",
.fp = (FItem)dmProcessMgmtQueue,
.param = pDnode,
.param = pMgmt,
};
if (tSingleWorkerInit(&pDnode->data.mgmtWorker, &cfg) != 0) {
if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) {
dError("failed to start dnode-mgmt worker since %s", terrstr());
return -1;
}
@ -182,13 +182,13 @@ int32_t dmStartWorker(SDnode *pDnode) {
return 0;
}
void dmStopWorker(SDnode *pDnode) {
tSingleWorkerCleanup(&pDnode->data.mgmtWorker);
void dmStopWorker(SDnodeMgmt *pMgmt) {
tSingleWorkerCleanup(&pMgmt->mgmtWorker);
dDebug("dnode workers are closed");
}
int32_t dmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSingleWorker *pWorker = &pWrapper->pDnode->data.mgmtWorker;
int32_t dmPutNodeMsgToMgmtQueue(SDnodeMgmt *pMgmt, SNodeMsg *pMsg) {
SSingleWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, put into worker %s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg);
return 0;

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

@ -1,7 +1,7 @@
aux_source_directory(src IMPLEMENT_SRC)
add_library(dnode STATIC ${IMPLEMENT_SRC})
target_link_libraries(
dnode mgmt_bnode mgmt_mnode mgmt_qnode mgmt_snode mgmt_vnode
dnode mgmt_bnode mgmt_mnode mgmt_qnode mgmt_snode mgmt_vnode mgmt_dnode
)
target_include_directories(
dnode

309
source/dnode/mgmt/node_mgmt/src/dmHandle.c
View File

@ -1,309 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http:www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dmImp.h"
static void dmUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
if (pDnode->data.dnodeId == 0 || pDnode->data.clusterId == 0) {
dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
taosWLockLatch(&pDnode->data.latch);
pDnode->data.dnodeId = pCfg->dnodeId;
pDnode->data.clusterId = pCfg->clusterId;
dmWriteEps(pDnode);
taosWUnLockLatch(&pDnode->data.latch);
}
}
static int32_t dmProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp) {
if (pRsp->code != TSDB_CODE_SUCCESS) {
if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pDnode->data.dropped && pDnode->data.dnodeId > 0) {
dInfo("dnode:%d, set to dropped since not exist in mnode", pDnode->data.dnodeId);
pDnode->data.dropped = 1;
dmWriteEps(pDnode);
}
} else {
SStatusRsp statusRsp = {0};
if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
pDnode->data.dnodeVer = statusRsp.dnodeVer;
dmUpdateDnodeCfg(pDnode, &statusRsp.dnodeCfg);
dmUpdateEps(pDnode, statusRsp.pDnodeEps);
}
rpcFreeCont(pRsp->pCont);
tFreeSStatusRsp(&statusRsp);
}
return TSDB_CODE_SUCCESS;
}
void dmSendStatusReq(SDnode *pDnode) {
SStatusReq req = {0};
taosRLockLatch(&pDnode->data.latch);
req.sver = tsVersion;
req.dnodeVer = pDnode->data.dnodeVer;
req.dnodeId = pDnode->data.dnodeId;
req.clusterId = pDnode->data.clusterId;
if (req.clusterId == 0) req.dnodeId = 0;
req.rebootTime = pDnode->data.rebootTime;
req.updateTime = pDnode->data.updateTime;
req.numOfCores = tsNumOfCores;
req.numOfSupportVnodes = pDnode->data.supportVnodes;
tstrncpy(req.dnodeEp, pDnode->data.localEp, TSDB_EP_LEN);
req.clusterCfg.statusInterval = tsStatusInterval;
req.clusterCfg.checkTime = 0;
char timestr[32] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &req.clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
taosRUnLockLatch(&pDnode->data.latch);
SMonVloadInfo vinfo = {0};
dmGetVnodeLoads(pDnode, &vinfo);
req.pVloads = vinfo.pVloads;
pDnode->data.unsyncedVgId = 0;
pDnode->data.vndState = TAOS_SYNC_STATE_LEADER;
for (int32_t i = 0; i < taosArrayGetSize(req.pVloads); ++i) {
SVnodeLoad *pLoad = taosArrayGet(req.pVloads, i);
if (pLoad->syncState != TAOS_SYNC_STATE_LEADER && pLoad->syncState != TAOS_SYNC_STATE_FOLLOWER) {
pDnode->data.unsyncedVgId = pLoad->vgId;
pDnode->data.vndState = pLoad->syncState;
}
}
SMonMloadInfo minfo = {0};
dmGetMnodeLoads(pDnode, &minfo);
pDnode->data.isMnode = minfo.isMnode;
pDnode->data.mndState = minfo.load.syncState;
int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
void *pHead = rpcMallocCont(contLen);
tSerializeSStatusReq(pHead, contLen, &req);
tFreeSStatusReq(&req);
SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
SRpcMsg rpcRsp = {0};
dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
dmSendToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);
dmProcessStatusRsp(pDnode, &rpcRsp);
}
int32_t dmProcessAuthRsp(SDnode *pDnode, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("auth rsp is received, but not supported yet");
return 0;
}
int32_t dmProcessGrantRsp(SDnode *pDnode, SNodeMsg *pMsg) {
SRpcMsg *pRsp = &pMsg->rpcMsg;
dError("grant rsp is received, but not supported yet");
return 0;
}
int32_t dmProcessConfigReq(SDnode *pDnode, SNodeMsg *pMsg) {
SRpcMsg *pReq = &pMsg->rpcMsg;
SDCfgDnodeReq *pCfg = pReq->pCont;
dError("config req is received, but not supported yet");
return TSDB_CODE_OPS_NOT_SUPPORT;
}
int32_t dmProcessCreateNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper != NULL) {
dmReleaseWrapper(pWrapper);
terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
dError("failed to create node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
pWrapper = &pDnode->wrappers[ntype];
if (taosMkDir(pWrapper->path) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
return -1;
}
int32_t code = (*pWrapper->fp.createFp)(pWrapper, pMsg);
if (code != 0) {
dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been created", pWrapper->name);
(void)dmOpenNode(pWrapper);
pWrapper->required = true;
pWrapper->deployed = true;
pWrapper->procType = pDnode->ptype;
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
int32_t dmProcessDropNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
if (pWrapper == NULL) {
terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
dError("failed to drop node since %s", terrstr());
return -1;
}
taosThreadMutexLock(&pDnode->mutex);
int32_t code = (*pWrapper->fp.dropFp)(pWrapper, pMsg);
if (code != 0) {
dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
} else {
dDebug("node:%s, has been dropped", pWrapper->name);
pWrapper->required = false;
pWrapper->deployed = false;
}
dmReleaseWrapper(pWrapper);
if (code == 0) {
dmCloseNode(pWrapper);
taosRemoveDir(pWrapper->path);
}
taosThreadMutexUnlock(&pDnode->mutex);
return code;
}
static void dmSetMgmtMsgHandle(SMgmtWrapper *pWrapper) {
// Requests handled by DNODE
dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE, dmProcessMgmtMsg, 0);
// Requests handled by MNODE
dmSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmProcessMgmtMsg, 0);
dmSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmProcessMgmtMsg, 0);
}
static int32_t dmStartMgmt(SMgmtWrapper *pWrapper) {
if (dmStartStatusThread(pWrapper->pDnode) != 0) {
return -1;
}
if (dmStartMonitorThread(pWrapper->pDnode) != 0) {
return -1;
}
return 0;
}
static void dmStopMgmt(SMgmtWrapper *pWrapper) {
dmStopMonitorThread(pWrapper->pDnode);
dmStopStatusThread(pWrapper->pDnode);
}
static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
dInfo("dnode-mgmt start to init");
SDnode *pDnode = pWrapper->pDnode;
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
monCfg.server = tsMonitorFqdn;
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
return -1;
}
pDnode->data.dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
if (pDnode->data.dnodeHash == NULL) {
dError("failed to init dnode hash");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (dmReadEps(pDnode) != 0) {
dError("failed to read file since %s", terrstr());
return -1;
}
if (pDnode->data.dropped) {
dError("dnode will not start since its already dropped");
return -1;
}
if (dmStartWorker(pDnode) != 0) {
return -1;
}
if (dmInitServer(pDnode) != 0) {
dError("failed to init transport since %s", terrstr());
return -1;
}
dmReportStartup(pDnode, "dnode-transport", "initialized");
if (udfStartUdfd(pDnode->data.dnodeId) != 0) {
dError("failed to start udfd");
}
dInfo("dnode-mgmt is initialized");
return 0;
}
static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
dInfo("dnode-mgmt start to clean up");
SDnode *pDnode = pWrapper->pDnode;
udfStopUdfd();
dmStopWorker(pDnode);
taosWLockLatch(&pDnode->data.latch);
if (pDnode->data.dnodeEps != NULL) {
taosArrayDestroy(pDnode->data.dnodeEps);
pDnode->data.dnodeEps = NULL;
}
if (pDnode->data.dnodeHash != NULL) {
taosHashCleanup(pDnode->data.dnodeHash);
pDnode->data.dnodeHash = NULL;
}
taosWUnLockLatch(&pDnode->data.latch);
dmCleanupClient(pDnode);
dmCleanupServer(pDnode);
dInfo("dnode-mgmt is cleaned up");
}
static int32_t dmRequireMgmt(SMgmtWrapper *pWrapper, bool *required) {
*required = true;
return 0;
}
void dmInitWrapper(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0};
mgmtFp.openFp = dmInitMgmt;
mgmtFp.closeFp = dmCleanupMgmt;
mgmtFp.startFp = dmStartMgmt;
mgmtFp.stopFp = dmStopMgmt;
mgmtFp.requiredFp = dmRequireMgmt;
dmSetMgmtMsgHandle(pWrapper);
pWrapper->name = "dnode";
pWrapper->fp = mgmtFp;
}

211
source/dnode/mgmt/node_mgmt/src/dmMonitor.c
View File

@ -1,211 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dmImp.h"
static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->protocol = 1;
pInfo->dnode_id = pDnode->data.dnodeId;
pInfo->cluster_id = pDnode->data.clusterId;
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
}
static void dmGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pDnode->data.rebootTime) / (86400000.0f);
pInfo->has_mnode = pDnode->wrappers[MNODE].required;
pInfo->has_qnode = pDnode->wrappers[QNODE].required;
pInfo->has_snode = pDnode->wrappers[SNODE].required;
pInfo->has_bnode = pDnode->wrappers[BNODE].required;
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size;
}
static void dmGetMonitorInfo(SDnode *pDnode, SMonDmInfo *pInfo) {
dmGetMonitorBasicInfo(pDnode, &pInfo->basic);
dmGetMonitorSysInfo(&pInfo->sys);
dmGetMonitorDnodeInfo(pDnode, &pInfo->dnode);
}
void dmSendMonitorReport(SDnode *pDnode) {
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
dTrace("send monitor report to %s:%u", tsMonitorFqdn, tsMonitorPort);
SMonDmInfo dmInfo = {0};
SMonMmInfo mmInfo = {0};
SMonVmInfo vmInfo = {0};
SMonQmInfo qmInfo = {0};
SMonSmInfo smInfo = {0};
SMonBmInfo bmInfo = {0};
SRpcMsg req = {0};
SRpcMsg rsp;
SEpSet epset = {.inUse = 0, .numOfEps = 1};
tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
epset.eps[0].port = tsServerPort;
SMgmtWrapper *pWrapper = NULL;
dmGetMonitorInfo(pDnode, &dmInfo);
bool getFromAPI = !tsMultiProcess;
pWrapper = &pDnode->wrappers[MNODE];
if (getFromAPI) {
if (dmMarkWrapper(pWrapper) == 0) {
mmGetMonitorInfo(pWrapper, &mmInfo);
dmReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_MM_INFO;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonMmInfo(rsp.pCont, rsp.contLen, &mmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
pWrapper = &pDnode->wrappers[VNODE];
if (getFromAPI) {
if (dmMarkWrapper(pWrapper) == 0) {
vmGetMonitorInfo(pWrapper, &vmInfo);
dmReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_VM_INFO;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonVmInfo(rsp.pCont, rsp.contLen, &vmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
pWrapper = &pDnode->wrappers[QNODE];
if (getFromAPI) {
if (dmMarkWrapper(pWrapper) == 0) {
qmGetMonitorInfo(pWrapper, &qmInfo);
dmReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_QM_INFO;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonQmInfo(rsp.pCont, rsp.contLen, &qmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
pWrapper = &pDnode->wrappers[SNODE];
if (getFromAPI) {
if (dmMarkWrapper(pWrapper) == 0) {
smGetMonitorInfo(pWrapper, &smInfo);
dmReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_SM_INFO;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonSmInfo(rsp.pCont, rsp.contLen, &smInfo);
}
rpcFreeCont(rsp.pCont);
}
}
pWrapper = &pDnode->wrappers[BNODE];
if (getFromAPI) {
if (dmMarkWrapper(pWrapper) == 0) {
bmGetMonitorInfo(pWrapper, &bmInfo);
dmReleaseWrapper(pWrapper);
}
} else {
if (pWrapper->required) {
req.msgType = TDMT_MON_BM_INFO;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonBmInfo(rsp.pCont, rsp.contLen, &bmInfo);
}
rpcFreeCont(rsp.pCont);
}
}
monSetDmInfo(&dmInfo);
monSetMmInfo(&mmInfo);
monSetVmInfo(&vmInfo);
monSetQmInfo(&qmInfo);
monSetSmInfo(&smInfo);
monSetBmInfo(&bmInfo);
tFreeSMonMmInfo(&mmInfo);
tFreeSMonVmInfo(&vmInfo);
tFreeSMonQmInfo(&qmInfo);
tFreeSMonSmInfo(&smInfo);
tFreeSMonBmInfo(&bmInfo);
monSendReport();
}
void dmGetVnodeLoads(SDnode *pDnode, SMonVloadInfo *pInfo) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, VNODE);
if (pWrapper == NULL) return;
bool getFromAPI = !tsMultiProcess;
if (getFromAPI) {
vmGetVnodeLoads(pWrapper, pInfo);
} else {
SRpcMsg req = {.msgType = TDMT_MON_VM_LOAD};
SRpcMsg rsp = {0};
SEpSet epset = {.inUse = 0, .numOfEps = 1};
tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
epset.eps[0].port = tsServerPort;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonVloadInfo(rsp.pCont, rsp.contLen, pInfo);
}
rpcFreeCont(rsp.pCont);
}
dmReleaseWrapper(pWrapper);
}
void dmGetMnodeLoads(SDnode *pDnode, SMonMloadInfo *pInfo) {
SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, MNODE);
if (pWrapper == NULL) {
pInfo->isMnode = 0;
return;
}
bool getFromAPI = !tsMultiProcess;
if (getFromAPI) {
mmGetMnodeLoads(pWrapper, pInfo);
} else {
SRpcMsg req = {.msgType = TDMT_MON_MM_LOAD};
SRpcMsg rsp = {0};
SEpSet epset = {.inUse = 0, .numOfEps = 1};
tstrncpy(epset.eps[0].fqdn, pDnode->data.localFqdn, TSDB_FQDN_LEN);
epset.eps[0].port = tsServerPort;
dmSendRecv(pDnode, &epset, &req, &rsp);
if (rsp.code == 0 && rsp.contLen > 0) {
tDeserializeSMonMloadInfo(rsp.pCont, rsp.contLen, pInfo);
}
rpcFreeCont(rsp.pCont);
}
dmReleaseWrapper(pWrapper);
}

1
source/dnode/mgmt/node_util/inc/dmUtil.h
View File

@ -181,6 +181,7 @@ typedef struct {
int32_t numOfDisks;
int32_t supportVnodes;
uint16_t serverPort;
EDndRunStatus status;
} SDnodeData;
#ifdef __cplusplus