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

refact dnode-mnode

This commit is contained in:
Shengliang Guan 2021-11-02 17:46:47 +08:00
parent 84effaaba9
commit 99c7162816
10 changed files with 304 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/common/taosmsg.h
View File

@ -833,9 +833,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;
int32_t mnodeNum; } SCreateMnodeMsg, SDropMnodeMsg;
SDnodeEp mnodeEps[];
} SCreateMnodeMsg;
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;

119
include/server/mnode/mnode.h
View File

@ -20,82 +20,6 @@
extern "C" { extern "C" {
#endif #endif
typedef enum { MN_STATUS_UNINIT = 0, MN_STATUS_INIT = 1, MN_STATUS_READY = 2, MN_STATUS_CLOSING = 3 } EMnStatus;
typedef struct {
/**
* Send messages to other dnodes, such as create vnode message.
*
* @param epSet, the endpoint list of the dnodes.
* @param rpcMsg, message to be sent.
*/
void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
/**
* Send messages to mnode, such as config message.
*
* @param rpcMsg, message to be sent.
*/
void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg);
/**
* Send redirect message to dnode or shell.
*
* @param rpcMsg, message to be sent.
* @param forShell, used to identify whether to send to shell or dnode.
*/
void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell);
/**
* Get the corresponding endpoint information from dnodeId.
*
* @param dnode, the instance of dDnode module.
* @param dnodeId, the id ot dnode.
* @param ep, the endpoint of dnode.
* @param fqdn, the fqdn of dnode.
* @param port, the port of dnode.
*/
void (*GetDnodeEp)(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
} SMnodeFp;
typedef struct {
SMnodeFp fp;
int64_t clusterId;
int32_t dnodeId;
} SMnodePara;
/**
* Initialize and start mnode module.
*
* @param para, initialization parameters.
* @return Error code.
*/
int32_t mnodeInit(SMnodePara para);
/**
* Stop and cleanup mnode module.
*/
void mnodeCleanup();
/**
* Deploy mnode instances in dnode.
*
* @return Error Code.
*/
int32_t mnodeDeploy();
/**
* Delete the mnode instance deployed in dnode.
*/
void mnodeUnDeploy();
/**
* Whether the mnode is in service.
*
* @return Server status.
*/
EMnStatus mnodeGetStatus();
typedef struct { typedef struct {
int64_t numOfDnode; int64_t numOfDnode;
int64_t numOfMnode; int64_t numOfMnode;
@ -109,32 +33,29 @@ typedef struct {
int64_t compStorage; int64_t compStorage;
} SMnodeStat; } SMnodeStat;
/** typedef struct {
* Get the statistical information of Mnode. void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
* void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg);
* @param stat, statistical information. void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell);
* @return Error Code. void (*GetDnodeEp)(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
*/ } SMnodeFp;
int32_t mnodeGetStatistics(SMnodeStat *stat);
/** typedef struct {
* Get the auth information of Mnode. SMnodeFp fp;
* int64_t clusterId;
* @param user, username. int32_t dnodeId;
* @param spi, security parameter index. } SMnodePara;
* @param encrypt, encrypt algorithm.
* @param secret, key for authentication. int32_t mnodeInit(SMnodePara para);
* @param ckey, ciphering key. void mnodeCleanup();
* @return Error Code. int32_t mnodeDeploy();
*/ void mnodeUnDeploy();
int32_t mnodeStart();
void mnodeStop();
int32_t mnodeGetStatistics(SMnodeStat *stat);
int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey); int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
/**
* Interface for processing messages.
*
* @param rpcMsg, message to be processed.
* @return Error code.
*/
void mnodeProcessMsg(SRpcMsg *rpcMsg); void mnodeProcessMsg(SRpcMsg *rpcMsg);
#ifdef __cplusplus #ifdef __cplusplus

38
include/server/vnode/vnode.h
View File

@ -66,41 +66,12 @@ typedef struct SVnodeMsg {
char pCont[]; char pCont[];
} SVnodeMsg; } SVnodeMsg;
/**
* Start initialize vnode module.
*
* @return Error code.
*/
int32_t vnodeInit(); int32_t vnodeInit();
/**
* Cleanup vnode module.
*/
void vnodeCleanup(); void vnodeCleanup();
/**
* Get the statistical information of vnode.
*
* @param pVnode,
* @param pStat, statistical information.
* @return Error Code.
*/
int32_t vnodeGetStatistics(SVnode *pVnode, SVnodeStatisic *pStat); int32_t vnodeGetStatistics(SVnode *pVnode, SVnodeStatisic *pStat);
/**
* Get the status of all vnodes.
*
* @param pVnode,
* @param status, status information.
* @return Error Code.
*/
int32_t vnodeGetStatus(SVnode *pVnode, SVnodeStatus *pStatus); int32_t vnodeGetStatus(SVnode *pVnode, SVnodeStatus *pStatus);
/**
* Operation functions of vnode
*
* @return Error Code.
*/
SVnode *vnodeOpen(int32_t vgId, const char *path); SVnode *vnodeOpen(int32_t vgId, const char *path);
void vnodeClose(SVnode *pVnode); void vnodeClose(SVnode *pVnode);
int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg); int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg);
@ -109,14 +80,7 @@ int32_t vnodeDrop(SVnode *pVnode);
int32_t vnodeCompact(SVnode *pVnode); int32_t vnodeCompact(SVnode *pVnode);
int32_t vnodeSync(SVnode *pVnode); int32_t vnodeSync(SVnode *pVnode);
/** void vnodeProcessMsg(SVnode *pVnode, SVnodeMsg *pMsg);
* Interface for processing messages.
*
* @param pVnode,
* @param pMsg, message to be processed.
*
*/
int32_t vnodeProcessMsg(SVnode *pVnode, SVnodeMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

12
include/util/taoserror.h
View File

@ -216,11 +216,13 @@ int32_t* taosGetErrno();
// dnode // dnode
#define TSDB_CODE_DND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0400) //"Message not processed") #define TSDB_CODE_DND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0400) //"Message not processed")
#define TSDB_CODE_DND_OUT_OF_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0401) //"Dnode out of memory") #define TSDB_CODE_DND_OUT_OF_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0401) //"Dnode out of memory")
#define TSDB_CODE_DND_NO_WRITE_ACCESS TAOS_DEF_ERROR_CODE(0, 0x0402) //"No permission for disk files in dnode") #define TSDB_CODE_DND_DNODE_ID_NOT_MATCHED TAOS_DEF_ERROR_CODE(0, 0x0402) //"Dnode Id not matched")
#define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0403) //"Invalid message length") #define TSDB_CODE_DND_MNODE_ALREADY_DROPPED TAOS_DEF_ERROR_CODE(0, 0x0403) //"Mnode already deployed")
#define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0404) //"Action in progress") #define TSDB_CODE_DND_NO_WRITE_ACCESS TAOS_DEF_ERROR_CODE(0, 0x0404) //"No permission for disk files in dnode")
#define TSDB_CODE_DND_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0405) //"Too many vnode directories") #define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0405) //"Invalid message length")
#define TSDB_CODE_DND_EXITING TAOS_DEF_ERROR_CODE(0, 0x0406) //"Dnode is exiting" #define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0406) //"Action in progress")
#define TSDB_CODE_DND_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0407) //"Too many vnode directories")
#define TSDB_CODE_DND_EXITING TAOS_DEF_ERROR_CODE(0, 0x0408) //"Dnode is exiting"
// vnode // vnode
#define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500) //"Action in progress") #define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500) //"Action in progress")

57
source/dnode/mgmt/src/dnodeDnode.c
View File

@ -35,7 +35,6 @@ static struct {
int8_t threadStop; int8_t threadStop;
pthread_t *threadId; pthread_t *threadId;
pthread_mutex_t mutex; pthread_mutex_t mutex;
MsgFp msgFp[TSDB_MSG_TYPE_MAX];
} tsDnode = {0}; } tsDnode = {0};
int32_t dnodeGetDnodeId() { int32_t dnodeGetDnodeId() {
@ -127,7 +126,7 @@ static void dnodeUpdateMnodeEpSet(SEpSet *pEpSet) {
pthread_mutex_unlock(&tsDnode.mutex); pthread_mutex_unlock(&tsDnode.mutex);
} }
static void dnodePrintEps() { static void dnodePrintDnodes() {
dDebug("print dnode endpoint list, num:%d", tsDnode.dnodeEps->dnodeNum); dDebug("print dnode endpoint list, num:%d", tsDnode.dnodeEps->dnodeNum);
for (int32_t i = 0; i < tsDnode.dnodeEps->dnodeNum; i++) { for (int32_t i = 0; i < tsDnode.dnodeEps->dnodeNum; i++) {
SDnodeEp *ep = &tsDnode.dnodeEps->dnodeEps[i]; SDnodeEp *ep = &tsDnode.dnodeEps->dnodeEps[i];
@ -135,7 +134,7 @@ static void dnodePrintEps() {
} }
} }
static void dnodeResetEps(SDnodeEps *pEps) { static void dnodeResetDnodes(SDnodeEps *pEps) {
assert(pEps != NULL); assert(pEps != NULL);
int32_t size = sizeof(SDnodeEps) + pEps->dnodeNum * sizeof(SDnodeEp); int32_t size = sizeof(SDnodeEps) + pEps->dnodeNum * sizeof(SDnodeEp);
@ -171,7 +170,7 @@ static void dnodeResetEps(SDnodeEps *pEps) {
taosHashPut(tsDnode.dnodeHash, &ep->dnodeId, sizeof(int32_t), ep, sizeof(SDnodeEp)); taosHashPut(tsDnode.dnodeHash, &ep->dnodeId, sizeof(int32_t), ep, sizeof(SDnodeEp));
} }
dnodePrintEps(); dnodePrintDnodes();
} }
static bool dnodeIsEpChanged(int32_t dnodeId, char *epStr) { static bool dnodeIsEpChanged(int32_t dnodeId, char *epStr) {
@ -189,7 +188,7 @@ static bool dnodeIsEpChanged(int32_t dnodeId, char *epStr) {
return changed; return changed;
} }
static int32_t dnodeReadEps() { static int32_t dnodeReadDnodes() {
int32_t len = 0; int32_t len = 0;
int32_t maxLen = 30000; int32_t maxLen = 30000;
char *content = calloc(1, maxLen + 1); char *content = calloc(1, maxLen + 1);
@ -199,59 +198,59 @@ static int32_t dnodeReadEps() {
fp = fopen(tsDnode.file, "r"); fp = fopen(tsDnode.file, "r");
if (!fp) { if (!fp) {
dDebug("file %s not exist", tsDnode.file); dDebug("file %s not exist", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
len = (int32_t)fread(content, 1, maxLen, fp); len = (int32_t)fread(content, 1, maxLen, fp);
if (len <= 0) { if (len <= 0) {
dError("failed to read %s since content is null", tsDnode.file); dError("failed to read %s since content is null", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
content[len] = 0; content[len] = 0;
root = cJSON_Parse(content); root = cJSON_Parse(content);
if (root == NULL) { if (root == NULL) {
dError("failed to read %s since invalid json format", tsDnode.file); dError("failed to read %s since invalid json format", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
cJSON *dnodeId = cJSON_GetObjectItem(root, "dnodeId"); cJSON *dnodeId = cJSON_GetObjectItem(root, "dnodeId");
if (!dnodeId || dnodeId->type != cJSON_String) { if (!dnodeId || dnodeId->type != cJSON_String) {
dError("failed to read %s since dnodeId not found", tsDnode.file); dError("failed to read %s since dnodeId not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tsDnode.dnodeId = atoi(dnodeId->valuestring); tsDnode.dnodeId = atoi(dnodeId->valuestring);
cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId"); cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
if (!clusterId || clusterId->type != cJSON_String) { if (!clusterId || clusterId->type != cJSON_String) {
dError("failed to read %s since clusterId not found", tsDnode.file); dError("failed to read %s since clusterId not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tsDnode.clusterId = atoll(clusterId->valuestring); tsDnode.clusterId = atoll(clusterId->valuestring);
cJSON *dropped = cJSON_GetObjectItem(root, "dropped"); cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_String) { if (!dropped || dropped->type != cJSON_String) {
dError("failed to read %s since dropped not found", tsDnode.file); dError("failed to read %s since dropped not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tsDnode.dropped = atoi(dropped->valuestring); tsDnode.dropped = atoi(dropped->valuestring);
cJSON *dnodeInfos = cJSON_GetObjectItem(root, "dnodeInfos"); cJSON *dnodeInfos = cJSON_GetObjectItem(root, "dnodeInfos");
if (!dnodeInfos || dnodeInfos->type != cJSON_Array) { if (!dnodeInfos || dnodeInfos->type != cJSON_Array) {
dError("failed to read %s since dnodeInfos not found", tsDnode.file); dError("failed to read %s since dnodeInfos not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
int32_t dnodeInfosSize = cJSON_GetArraySize(dnodeInfos); int32_t dnodeInfosSize = cJSON_GetArraySize(dnodeInfos);
if (dnodeInfosSize <= 0) { if (dnodeInfosSize <= 0) {
dError("failed to read %s since dnodeInfos size:%d invalid", tsDnode.file, dnodeInfosSize); dError("failed to read %s since dnodeInfos size:%d invalid", tsDnode.file, dnodeInfosSize);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tsDnode.dnodeEps = calloc(1, dnodeInfosSize * sizeof(SDnodeEp) + sizeof(SDnodeEps)); tsDnode.dnodeEps = calloc(1, dnodeInfosSize * sizeof(SDnodeEp) + sizeof(SDnodeEps));
if (tsDnode.dnodeEps == NULL) { if (tsDnode.dnodeEps == NULL) {
dError("failed to calloc dnodeEpList since %s", strerror(errno)); dError("failed to calloc dnodeEpList since %s", strerror(errno));
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tsDnode.dnodeEps->dnodeNum = dnodeInfosSize; tsDnode.dnodeEps->dnodeNum = dnodeInfosSize;
@ -264,36 +263,36 @@ static int32_t dnodeReadEps() {
cJSON *dnodeId = cJSON_GetObjectItem(dnodeInfo, "dnodeId"); cJSON *dnodeId = cJSON_GetObjectItem(dnodeInfo, "dnodeId");
if (!dnodeId || dnodeId->type != cJSON_String) { if (!dnodeId || dnodeId->type != cJSON_String) {
dError("failed to read %s, dnodeId not found", tsDnode.file); dError("failed to read %s, dnodeId not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
pEp->dnodeId = atoi(dnodeId->valuestring); pEp->dnodeId = atoi(dnodeId->valuestring);
cJSON *isMnode = cJSON_GetObjectItem(dnodeInfo, "isMnode"); cJSON *isMnode = cJSON_GetObjectItem(dnodeInfo, "isMnode");
if (!isMnode || isMnode->type != cJSON_String) { if (!isMnode || isMnode->type != cJSON_String) {
dError("failed to read %s, isMnode not found", tsDnode.file); dError("failed to read %s, isMnode not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
pEp->isMnode = atoi(isMnode->valuestring); pEp->isMnode = atoi(isMnode->valuestring);
cJSON *dnodeFqdn = cJSON_GetObjectItem(dnodeInfo, "dnodeFqdn"); cJSON *dnodeFqdn = cJSON_GetObjectItem(dnodeInfo, "dnodeFqdn");
if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) { if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) {
dError("failed to read %s, dnodeFqdn not found", tsDnode.file); dError("failed to read %s, dnodeFqdn not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
tstrncpy(pEp->dnodeFqdn, dnodeFqdn->valuestring, TSDB_FQDN_LEN); tstrncpy(pEp->dnodeFqdn, dnodeFqdn->valuestring, TSDB_FQDN_LEN);
cJSON *dnodePort = cJSON_GetObjectItem(dnodeInfo, "dnodePort"); cJSON *dnodePort = cJSON_GetObjectItem(dnodeInfo, "dnodePort");
if (!dnodePort || dnodePort->type != cJSON_String) { if (!dnodePort || dnodePort->type != cJSON_String) {
dError("failed to read %s, dnodePort not found", tsDnode.file); dError("failed to read %s, dnodePort not found", tsDnode.file);
goto PRASE_EPS_OVER; goto PRASE_DNODE_OVER;
} }
pEp->dnodePort = atoi(dnodePort->valuestring); pEp->dnodePort = atoi(dnodePort->valuestring);
} }
dInfo("succcessed to read file %s", tsDnode.file); dInfo("succcessed to read file %s", tsDnode.file);
dnodePrintEps(); dnodePrintDnodes();
PRASE_EPS_OVER: PRASE_DNODE_OVER:
if (content != NULL) free(content); if (content != NULL) free(content);
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
if (fp != NULL) fclose(fp); if (fp != NULL) fclose(fp);
@ -303,13 +302,13 @@ PRASE_EPS_OVER:
return -1; return -1;
} }
dnodeResetEps(tsDnode.dnodeEps); dnodeResetDnodes(tsDnode.dnodeEps);
terrno = 0; terrno = 0;
return 0; return 0;
} }
static int32_t dnodeWriteEps() { static int32_t dnodeWriteDnodes() {
FILE *fp = fopen(tsDnode.file, "w"); FILE *fp = fopen(tsDnode.file, "w");
if (!fp) { if (!fp) {
dError("failed to write %s since %s", tsDnode.file, strerror(errno)); dError("failed to write %s since %s", tsDnode.file, strerror(errno));
@ -391,7 +390,7 @@ static void dnodeUpdateCfg(SDnodeCfg *pCfg) {
tsDnode.dropped = pCfg->dropped; tsDnode.dropped = pCfg->dropped;
dInfo("dnodeId is set to %d, clusterId is set to %" PRId64, pCfg->dnodeId, pCfg->clusterId); dInfo("dnodeId is set to %d, clusterId is set to %" PRId64, pCfg->dnodeId, pCfg->clusterId);
dnodeWriteEps(); dnodeWriteDnodes();
pthread_mutex_unlock(&tsDnode.mutex); pthread_mutex_unlock(&tsDnode.mutex);
} }
@ -401,13 +400,13 @@ static void dnodeUpdateDnodeEps(SDnodeEps *pEps) {
pthread_mutex_lock(&tsDnode.mutex); pthread_mutex_lock(&tsDnode.mutex);
if (pEps->dnodeNum != tsDnode.dnodeEps->dnodeNum) { if (pEps->dnodeNum != tsDnode.dnodeEps->dnodeNum) {
dnodeResetEps(pEps); dnodeResetDnodes(pEps);
dnodeWriteEps(); dnodeWriteDnodes();
} else { } else {
int32_t size = pEps->dnodeNum * sizeof(SDnodeEp) + sizeof(SDnodeEps); int32_t size = pEps->dnodeNum * sizeof(SDnodeEp) + sizeof(SDnodeEps);
if (memcmp(tsDnode.dnodeEps, pEps, size) != 0) { if (memcmp(tsDnode.dnodeEps, pEps, size) != 0) {
dnodeResetEps(pEps); dnodeResetDnodes(pEps);
dnodeWriteEps(); dnodeWriteDnodes();
} }
} }
@ -493,9 +492,9 @@ int32_t dnodeInitDnode() {
return TSDB_CODE_DND_OUT_OF_MEMORY; return TSDB_CODE_DND_OUT_OF_MEMORY;
} }
int32_t code = dnodeReadEps(); int32_t code = dnodeReadDnodes();
if (code != 0) { if (code != 0) {
dError("failed to read dnode endpoint file since %s", tstrerror(code)); dError("failed to read file:%s since %s", tsDnode.file, tstrerror(code));
return code; return code;
} }

261
source/dnode/mgmt/src/dnodeMnode.c
View File

@ -17,9 +17,211 @@
#include "dnodeMnode.h" #include "dnodeMnode.h"
#include "dnodeDnode.h" #include "dnodeDnode.h"
#include "dnodeTransport.h" #include "dnodeTransport.h"
#include "cJSON.h"
#include "mnode.h" #include "mnode.h"
static struct {
int8_t deployed;
int8_t dropped;
char file[PATH_MAX + 20];
pthread_mutex_t mutex;
} tsMnode = {0};
static int32_t dnodeReadMnode() {
int32_t len = 0;
int32_t maxLen = 300;
char *content = calloc(1, maxLen + 1);
cJSON *root = NULL;
FILE *fp = NULL;
fp = fopen(tsMnode.file, "r");
if (!fp) {
dDebug("file %s not exist", tsMnode.file);
goto PRASE_MNODE_OVER;
}
len = (int32_t)fread(content, 1, maxLen, fp);
if (len <= 0) {
dError("failed to read %s since content is null", tsMnode.file);
goto PRASE_MNODE_OVER;
}
content[len] = 0;
root = cJSON_Parse(content);
if (root == NULL) {
dError("failed to read %s since invalid json format", tsMnode.file);
goto PRASE_MNODE_OVER;
}
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
if (!deployed || deployed->type != cJSON_String) {
dError("failed to read %s since deployed not found", tsMnode.file);
goto PRASE_MNODE_OVER;
}
tsMnode.deployed = atoi(deployed->valuestring);
cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_String) {
dError("failed to read %s since dropped not found", tsMnode.file);
goto PRASE_MNODE_OVER;
}
tsMnode.dropped = atoi(dropped->valuestring);
dInfo("succcessed to read file %s", tsMnode.file);
PRASE_MNODE_OVER:
if (content != NULL) free(content);
if (root != NULL) cJSON_Delete(root);
if (fp != NULL) fclose(fp);
return 0;
}
static int32_t dnodeWriteMnode() {
FILE *fp = fopen(tsMnode.file, "w");
if (!fp) {
dError("failed to write %s since %s", tsMnode.file, strerror(errno));
return -1;
}
int32_t len = 0;
int32_t maxLen = 300;
char *content = calloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"deployed\": \"%d\",\n", tsMnode.dropped);
len += snprintf(content + len, maxLen - len, " \"dropped\": \"%d\",\n", tsMnode.dropped);
len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp);
taosFsyncFile(fileno(fp));
fclose(fp);
free(content);
terrno = 0;
dInfo("successed to write %s", tsMnode.file);
return 0;
}
static int32_t dnodeStartMnode(SCreateMnodeMsg *pCfg) {
int32_t code = 0;
if (pCfg->dnodeId != dnodeGetDnodeId()) {
code = TSDB_CODE_DND_DNODE_ID_NOT_MATCHED;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
if (tsMnode.dropped) {
code = TSDB_CODE_DND_MNODE_ALREADY_DROPPED;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
if (tsMnode.deployed) {
dError("failed to start mnode since its already deployed");
return 0;
}
tsMnode.deployed = 1;
tsMnode.dropped = 0;
code = dnodeWriteMnode();
if (code != 0) {
tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
code = mnodeDeploy();
if (code != 0) {
tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
code = mnodeStart();
if (code != 0) {
tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
tsMnode.deployed = 1;
return 0;
}
static int32_t dnodeDropMnode(SDropMnodeMsg *pCfg) {
int32_t code = 0;
if (pCfg->dnodeId != dnodeGetDnodeId()) {
code = TSDB_CODE_DND_DNODE_ID_NOT_MATCHED;
dError("failed to drop mnode since %s", tstrerror(code));
return code;
}
if (tsMnode.dropped) {
code = TSDB_CODE_DND_MNODE_ALREADY_DROPPED;
dError("failed to drop mnode since %s", tstrerror(code));
return code;
}
if (!tsMnode.deployed) {
dError("failed to drop mnode since not deployed");
return 0;
}
mnodeStop();
tsMnode.deployed = 0;
tsMnode.dropped = 1;
code = dnodeWriteMnode();
if (code != 0) {
tsMnode.deployed = 1;
tsMnode.dropped = 0;
dError("failed to drop mnode since %s", tstrerror(code));
return code;
}
mnodeUnDeploy();
tsMnode.deployed = 0;
return 0;
}
static void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg) {
SCreateMnodeMsg *pCfg = pMsg->pCont;
pCfg->dnodeId = htonl(pCfg->dnodeId);
int32_t code = dnodeStartMnode(pCfg);
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code};
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
}
static void dnodeProcessDropMnodeReq(SRpcMsg *pMsg) {
SDropMnodeMsg *pCfg = pMsg->pCont;
pCfg->dnodeId = htonl(pCfg->dnodeId);
int32_t code = dnodeDropMnode(pCfg);
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code};
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
}
static bool dnodeNeedDeployMnode() {
if (dnodeGetDnodeId() > 0) return false;
if (dnodeGetClusterId() > 0) return false;
if (strcmp(tsFirst, tsLocalEp) != 0) return false;
return true;
}
int32_t dnodeInitMnode() { int32_t dnodeInitMnode() {
tsMnode.dropped = 0;
tsMnode.deployed = 0;
snprintf(tsMnode.file, sizeof(tsMnode.file), "%s/mnode.json", tsDnodeDir);
SMnodePara para; SMnodePara para;
para.fp.GetDnodeEp = dnodeGetDnodeEp; para.fp.GetDnodeEp = dnodeGetDnodeEp;
para.fp.SendMsgToDnode = dnodeSendMsgToDnode; para.fp.SendMsgToDnode = dnodeSendMsgToDnode;
@ -28,46 +230,49 @@ int32_t dnodeInitMnode() {
para.dnodeId = dnodeGetDnodeId(); para.dnodeId = dnodeGetDnodeId();
para.clusterId = dnodeGetClusterId(); para.clusterId = dnodeGetClusterId();
return mnodeInit(para); int32_t code = mnodeInit(para);
if (code != 0) {
dError("failed to init mnode module since %s", tstrerror(code));
return code;
} }
void dnodeCleanupMnode() { mnodeCleanup(); } code = dnodeReadMnode();
if (code != 0) {
static int32_t dnodeStartMnode(SRpcMsg *pMsg) { dError("failed to read file:%s since %s", tsMnode.file, tstrerror(code));
SCreateMnodeMsg *pCfg = pMsg->pCont; return code;
pCfg->dnodeId = htonl(pCfg->dnodeId);
pCfg->mnodeNum = htonl(pCfg->mnodeNum);
for (int32_t i = 0; i < pCfg->mnodeNum; ++i) {
pCfg->mnodeEps[i].dnodeId = htonl(pCfg->mnodeEps[i].dnodeId);
pCfg->mnodeEps[i].dnodePort = htons(pCfg->mnodeEps[i].dnodePort);
} }
if (pCfg->dnodeId != dnodeGetDnodeId()) { if (tsMnode.dropped) {
dDebug("dnode:%d, in create meps msg is not equal with saved dnodeId:%d", pCfg->dnodeId, dnodeGetDnodeId()); dError("mnode already dropped, undeploy it");
return TSDB_CODE_MND_DNODE_ID_NOT_CONFIGURED; mnodeUnDeploy();
return 0;
} }
if (mnodeGetStatus() == MN_STATUS_READY) return 0; if (!tsMnode.deployed) {
bool needDeploy = dnodeNeedDeployMnode();
return mnodeDeploy(); if (needDeploy) {
code = mnodeDeploy();
} else {
return 0;
} }
void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg) { if (code != 0) {
int32_t code = dnodeStartMnode(pMsg); dError("failed to deploy mnode since %s", tstrerror(code));
return code;
SRpcMsg rspMsg = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0, .code = code};
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
} }
void dnodeProcessDropMnodeReq(SRpcMsg *pMsg) { tsMnode.deployed = 1;
int32_t code = dnodeStartMnode(pMsg); }
SRpcMsg rspMsg = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0, .code = code}; return mnodeStart();
}
rpcSendResponse(&rspMsg); void dnodeCleanupMnode() {
rpcFreeCont(pMsg->pCont); if (tsMnode.deployed) {
mnodeStop();
}
mnodeCleanup();
} }
void dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet) { void dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet) {

2
source/dnode/mnode/inc/mnodeInt.h
View File

@ -22,6 +22,8 @@
extern "C" { extern "C" {
#endif #endif
typedef enum { MN_STATUS_UNINIT = 0, MN_STATUS_INIT = 1, MN_STATUS_READY = 2, MN_STATUS_CLOSING = 3 } EMnStatus;
tmr_h mnodeGetTimer(); tmr_h mnodeGetTimer();
int32_t mnodeGetDnodeId(); int32_t mnodeGetDnodeId();
int64_t mnodeGetClusterId(); int64_t mnodeGetClusterId();

3
source/dnode/mnode/src/mondeInt.c
View File

@ -250,3 +250,6 @@ void mnodeCleanup() {
mInfo("mnode is cleaned up"); mInfo("mnode is cleaned up");
} }
} }
int32_t mnodeStart() { return 0; }
void mnodeStop() {}

2
source/dnode/vnode/impl/src/vnodeInt.c
View File

@ -30,4 +30,4 @@ int32_t vnodeDrop(SVnode *pVnode) { return 0; }
int32_t vnodeCompact(SVnode *pVnode) { return 0; } int32_t vnodeCompact(SVnode *pVnode) { return 0; }
int32_t vnodeSync(SVnode *pVnode) { return 0; } int32_t vnodeSync(SVnode *pVnode) { return 0; }
int32_t vnodeProcessMsg(SVnode *pVnode, SVnodeMsg *pMsg) { return 0; } void vnodeProcessMsg(SVnode *pVnode, SVnodeMsg *pMsg) {}

2
source/util/src/terror.c
View File

@ -228,6 +228,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST, "Topic already exists"
// dnode // dnode
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, "Message not processed") TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, "Message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, "Dnode out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, "Dnode out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_DNODE_ID_NOT_MATCHED, "Dnode Id not matched")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DROPPED, "Mnode already deployed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS, "No permission for disk files in dnode") TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS, "No permission for disk files in dnode")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN, "Invalid message length") TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN, "Invalid message length")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS, "Action in progress") TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS, "Action in progress")