TDengine/source/dnode/mgmt/impl/src/dndMnode.c

626 lines
18 KiB
C
Raw Normal View History

2021-11-22 02:03:41 +00:00
/*
* 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 "dndMnode.h"
2022-01-06 11:51:44 +00:00
#include "dndMgmt.h"
2021-11-22 02:03:41 +00:00
#include "dndTransport.h"
2021-12-28 09:38:15 +00:00
#include "dndWorker.h"
2021-11-22 02:03:41 +00:00
2022-01-04 11:44:44 +00:00
static void dndWriteMnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpcMsg *pRpcMsg);
2021-12-28 09:38:15 +00:00
static void dndProcessMnodeQueue(SDnode *pDnode, SMnodeMsg *pMsg);
2021-11-22 02:03:41 +00:00
static SMnode *dndAcquireMnode(SDnode *pDnode) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = NULL;
int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch);
if (pMgmt->deployed && !pMgmt->dropped) {
refCount = atomic_add_fetch_32(&pMgmt->refCount, 1);
pMnode = pMgmt->pMnode;
} else {
terrno = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
}
taosRUnLockLatch(&pMgmt->latch);
2021-11-27 09:23:57 +00:00
if (pMnode != NULL) {
dTrace("acquire mnode, refCount:%d", refCount);
}
2021-11-22 02:03:41 +00:00
return pMnode;
}
static void dndReleaseMnode(SDnode *pDnode, SMnode *pMnode) {
2022-01-08 08:48:09 +00:00
if (pMnode == NULL) return;
2021-11-22 02:03:41 +00:00
2022-01-08 08:48:09 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
taosRLockLatch(&pMgmt->latch);
2022-01-08 08:48:09 +00:00
int32_t refCount = atomic_sub_fetch_32(&pMgmt->refCount, 1);
2021-11-22 02:03:41 +00:00
taosRUnLockLatch(&pMgmt->latch);
2022-01-08 08:48:09 +00:00
dTrace("release mnode, refCount:%d", refCount);
2021-11-22 02:03:41 +00:00
}
static int32_t dndReadMnodeFile(SDnode *pDnode) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
int32_t code = TSDB_CODE_DND_MNODE_READ_FILE_ERROR;
int32_t len = 0;
int32_t maxLen = 4096;
2021-11-22 02:03:41 +00:00
char *content = calloc(1, maxLen + 1);
cJSON *root = NULL;
2021-12-28 09:38:15 +00:00
char file[PATH_MAX + 20];
snprintf(file, PATH_MAX + 20, "%s/mnode.json", pDnode->dir.dnode);
FILE *fp = fopen(file, "r");
2021-11-22 02:03:41 +00:00
if (fp == NULL) {
2021-12-28 09:38:15 +00:00
dDebug("file %s not exist", file);
2021-11-22 02:03:41 +00:00
code = 0;
goto PRASE_MNODE_OVER;
}
len = (int32_t)fread(content, 1, maxLen, fp);
if (len <= 0) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since content is null", file);
2021-11-22 02:03:41 +00:00
goto PRASE_MNODE_OVER;
}
content[len] = 0;
root = cJSON_Parse(content);
if (root == NULL) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since invalid json format", file);
2021-11-22 02:03:41 +00:00
goto PRASE_MNODE_OVER;
}
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
2021-12-02 06:40:03 +00:00
if (!deployed || deployed->type != cJSON_Number) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since deployed not found", file);
2021-11-22 02:03:41 +00:00
goto PRASE_MNODE_OVER;
}
2021-12-02 06:40:03 +00:00
pMgmt->deployed = deployed->valueint;
2021-11-22 02:03:41 +00:00
cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
2021-12-02 06:40:03 +00:00
if (!dropped || dropped->type != cJSON_Number) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since dropped not found", file);
2021-11-22 02:03:41 +00:00
goto PRASE_MNODE_OVER;
}
2021-12-02 06:40:03 +00:00
pMgmt->dropped = dropped->valueint;
2021-11-22 02:03:41 +00:00
2021-12-02 06:40:03 +00:00
cJSON *mnodes = cJSON_GetObjectItem(root, "mnodes");
if (!mnodes || mnodes->type != cJSON_Array) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since nodes not found", file);
goto PRASE_MNODE_OVER;
}
2021-12-02 06:40:03 +00:00
pMgmt->replica = cJSON_GetArraySize(mnodes);
if (pMgmt->replica <= 0 || pMgmt->replica > TSDB_MAX_REPLICA) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since mnodes size %d invalid", file, pMgmt->replica);
goto PRASE_MNODE_OVER;
}
for (int32_t i = 0; i < pMgmt->replica; ++i) {
2021-12-02 06:40:03 +00:00
cJSON *node = cJSON_GetArrayItem(mnodes, i);
if (node == NULL) break;
SReplica *pReplica = &pMgmt->replicas[i];
cJSON *id = cJSON_GetObjectItem(node, "id");
2021-12-02 06:40:03 +00:00
if (!id || id->type != cJSON_Number) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since id not found", file);
goto PRASE_MNODE_OVER;
}
2021-12-02 06:40:03 +00:00
pReplica->id = id->valueint;
cJSON *fqdn = cJSON_GetObjectItem(node, "fqdn");
if (!fqdn || fqdn->type != cJSON_String || fqdn->valuestring == NULL) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since fqdn not found", file);
goto PRASE_MNODE_OVER;
}
tstrncpy(pReplica->fqdn, fqdn->valuestring, TSDB_FQDN_LEN);
cJSON *port = cJSON_GetObjectItem(node, "port");
2021-12-02 06:40:03 +00:00
if (!port || port->type != cJSON_Number) {
2021-12-28 09:38:15 +00:00
dError("failed to read %s since port not found", file);
goto PRASE_MNODE_OVER;
}
2021-12-02 06:40:03 +00:00
pReplica->port = port->valueint;
}
2021-11-22 02:03:41 +00:00
code = 0;
2021-12-28 09:38:15 +00:00
dDebug("succcessed to read file %s, deployed:%d dropped:%d", file, pMgmt->deployed, pMgmt->dropped);
2021-11-22 02:03:41 +00:00
PRASE_MNODE_OVER:
if (content != NULL) free(content);
if (root != NULL) cJSON_Delete(root);
if (fp != NULL) fclose(fp);
terrno = code;
return code;
}
static int32_t dndWriteMnodeFile(SDnode *pDnode) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-12-28 09:38:15 +00:00
char file[PATH_MAX + 20];
snprintf(file, PATH_MAX + 20, "%s/mnode.json.bak", pDnode->dir.dnode);
2021-11-22 02:03:41 +00:00
FILE *fp = fopen(file, "w");
2021-11-27 06:57:16 +00:00
if (fp == NULL) {
2021-11-22 02:03:41 +00:00
terrno = TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR;
dError("failed to write %s since %s", file, terrstr());
return -1;
}
int32_t len = 0;
int32_t maxLen = 4096;
2021-11-22 02:03:41 +00:00
char *content = calloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n");
2021-12-02 06:40:03 +00:00
len += snprintf(content + len, maxLen - len, " \"deployed\": %d,\n", pMgmt->deployed);
2021-12-02 06:40:03 +00:00
len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pMgmt->dropped);
len += snprintf(content + len, maxLen - len, " \"mnodes\": [{\n");
for (int32_t i = 0; i < pMgmt->replica; ++i) {
SReplica *pReplica = &pMgmt->replicas[i];
2021-12-02 06:40:03 +00:00
len += snprintf(content + len, maxLen - len, " \"id\": %d,\n", pReplica->id);
len += snprintf(content + len, maxLen - len, " \"fqdn\": \"%s\",\n", pReplica->fqdn);
2021-12-02 06:40:03 +00:00
len += snprintf(content + len, maxLen - len, " \"port\": %u\n", pReplica->port);
if (i < pMgmt->replica - 1) {
len += snprintf(content + len, maxLen - len, " },{\n");
} else {
len += snprintf(content + len, maxLen - len, " }]\n");
}
}
2021-11-22 02:03:41 +00:00
len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp);
taosFsyncFile(fileno(fp));
fclose(fp);
free(content);
2021-12-28 09:38:15 +00:00
char realfile[PATH_MAX + 20];
snprintf(realfile, PATH_MAX + 20, "%s/mnode.json", pDnode->dir.dnode);
if (taosRenameFile(file, realfile) != 0) {
2021-11-22 02:03:41 +00:00
terrno = TSDB_CODE_DND_MNODE_WRITE_FILE_ERROR;
2021-12-28 09:38:15 +00:00
dError("failed to rename %s since %s", file, terrstr());
2021-11-22 02:03:41 +00:00
return -1;
}
2021-12-28 09:38:15 +00:00
dInfo("successed to write %s, deployed:%d dropped:%d", realfile, pMgmt->deployed, pMgmt->dropped);
2021-11-22 02:03:41 +00:00
return 0;
}
static int32_t dndStartMnodeWorker(SDnode *pDnode) {
2021-12-28 09:38:15 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
if (dndInitWorker(pDnode, &pMgmt->readWorker, DND_WORKER_SINGLE, "mnode-read", 0, 1, dndProcessMnodeQueue) != 0) {
2021-11-27 06:57:16 +00:00
dError("failed to start mnode read worker since %s", terrstr());
return -1;
}
2021-12-28 09:38:15 +00:00
if (dndInitWorker(pDnode, &pMgmt->writeWorker, DND_WORKER_SINGLE, "mnode-write", 0, 1, dndProcessMnodeQueue) != 0) {
2021-11-27 06:57:16 +00:00
dError("failed to start mnode write worker since %s", terrstr());
return -1;
}
2021-12-28 09:38:15 +00:00
if (dndInitWorker(pDnode, &pMgmt->syncWorker, DND_WORKER_SINGLE, "mnode-sync", 0, 1, dndProcessMnodeQueue) != 0) {
2021-11-27 06:57:16 +00:00
dError("failed to start mnode sync worker since %s", terrstr());
return -1;
}
2021-11-22 02:03:41 +00:00
return 0;
}
static void dndStopMnodeWorker(SDnode *pDnode) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
taosWLockLatch(&pMgmt->latch);
pMgmt->deployed = 0;
taosWUnLockLatch(&pMgmt->latch);
2021-12-28 09:38:15 +00:00
while (pMgmt->refCount > 1) {
taosMsleep(10);
}
2021-11-27 07:57:54 +00:00
2021-12-28 09:38:15 +00:00
dndCleanupWorker(&pMgmt->readWorker);
dndCleanupWorker(&pMgmt->writeWorker);
dndCleanupWorker(&pMgmt->syncWorker);
2021-11-22 02:03:41 +00:00
}
static bool dndNeedDeployMnode(SDnode *pDnode) {
if (dndGetDnodeId(pDnode) > 0) {
return false;
}
if (dndGetClusterId(pDnode) > 0) {
return false;
}
2021-11-26 08:28:38 +00:00
2022-01-12 02:46:52 +00:00
if (strcmp(pDnode->cfg.localEp, pDnode->cfg.firstEp) != 0) {
2021-11-22 02:03:41 +00:00
return false;
}
return true;
}
2022-01-04 11:44:44 +00:00
static int32_t dndPutMsgToMWriteQ(SDnode *pDnode, SRpcMsg *pRpcMsg) {
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.writeWorker, pRpcMsg);
}
2021-11-26 08:28:38 +00:00
static void dndInitMnodeOption(SDnode *pDnode, SMnodeOpt *pOption) {
pOption->pDnode = pDnode;
2022-01-04 11:44:44 +00:00
pOption->sendReqToDnodeFp = dndSendReqToDnode;
pOption->sendReqToMnodeFp = dndSendReqToMnode;
pOption->sendRedirectRspFp = dndSendRedirectRsp;
pOption->putReqToMWriteQFp = dndPutMsgToMWriteQ;
2021-11-26 08:28:38 +00:00
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
2022-01-12 02:46:52 +00:00
pOption->cfg.sver = pDnode->env.sver;
pOption->cfg.enableTelem = pDnode->env.enableTelem;
pOption->cfg.statusInterval = pDnode->cfg.statusInterval;
pOption->cfg.shellActivityTimer = pDnode->cfg.shellActivityTimer;
pOption->cfg.timezone = pDnode->env.timezone;
pOption->cfg.charset = pDnode->env.charset;
pOption->cfg.locale = pDnode->env.locale;
pOption->cfg.gitinfo = pDnode->env.gitinfo;
pOption->cfg.buildinfo = pDnode->env.buildinfo;
2021-11-26 08:28:38 +00:00
}
static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
dndInitMnodeOption(pDnode, pOption);
pOption->replica = 1;
pOption->selfIndex = 0;
SReplica *pReplica = &pOption->replicas[0];
pReplica->id = 1;
2022-01-12 02:46:52 +00:00
pReplica->port = pDnode->cfg.serverPort;
memcpy(pReplica->fqdn, pDnode->cfg.localFqdn, TSDB_FQDN_LEN);
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->selfIndex = pOption->selfIndex;
pMgmt->replica = pOption->replica;
memcpy(&pMgmt->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
2021-11-26 08:28:38 +00:00
}
static void dndBuildMnodeOpenOption(SDnode *pDnode, SMnodeOpt *pOption) {
dndInitMnodeOption(pDnode, pOption);
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pOption->selfIndex = pMgmt->selfIndex;
pOption->replica = pMgmt->replica;
memcpy(&pOption->replicas, pMgmt->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
2021-11-26 08:28:38 +00:00
}
2022-01-07 10:03:28 +00:00
static int32_t dndBuildMnodeOptionFromReq(SDnode *pDnode, SMnodeOpt *pOption, SDCreateMnodeReq *pCreate) {
2021-11-26 08:28:38 +00:00
dndInitMnodeOption(pDnode, pOption);
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
2022-01-07 10:03:28 +00:00
pOption->replica = pCreate->replica;
2021-11-26 08:28:38 +00:00
pOption->selfIndex = -1;
2022-01-07 10:03:28 +00:00
for (int32_t i = 0; i < pCreate->replica; ++i) {
SReplica *pReplica = &pOption->replicas[i];
2022-01-07 10:03:28 +00:00
pReplica->id = pCreate->replicas[i].id;
pReplica->port = pCreate->replicas[i].port;
memcpy(pReplica->fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
2021-11-26 08:28:38 +00:00
if (pReplica->id == pOption->dnodeId) {
pOption->selfIndex = i;
2021-11-22 02:03:41 +00:00
}
}
2021-11-26 08:28:38 +00:00
if (pOption->selfIndex == -1) {
2021-11-22 02:03:41 +00:00
dError("failed to build mnode options since %s", terrstr());
return -1;
}
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->selfIndex = pOption->selfIndex;
pMgmt->replica = pOption->replica;
memcpy(&pMgmt->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
2021-11-22 02:03:41 +00:00
return 0;
}
2021-11-26 08:28:38 +00:00
static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOpt *pOption) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
2021-11-29 04:09:18 +00:00
SMnode *pMnode = mndOpen(pDnode->dir.mnode, pOption);
2021-11-22 02:03:41 +00:00
if (pMnode == NULL) {
dError("failed to open mnode since %s", terrstr());
2021-12-01 09:27:31 +00:00
return -1;
}
2021-12-28 09:38:15 +00:00
if (dndStartMnodeWorker(pDnode) != 0) {
dError("failed to start mnode worker since %s", terrstr());
2021-12-01 09:27:31 +00:00
mndClose(pMnode);
mndDestroy(pDnode->dir.mnode);
return -1;
2021-11-22 02:03:41 +00:00
}
2021-12-28 09:38:15 +00:00
pMgmt->deployed = 1;
if (dndWriteMnodeFile(pDnode) != 0) {
dError("failed to write mnode file since %s", terrstr());
2021-12-01 09:27:31 +00:00
pMgmt->deployed = 0;
2021-11-22 02:03:41 +00:00
dndStopMnodeWorker(pDnode);
2021-11-29 04:09:18 +00:00
mndClose(pMnode);
mndDestroy(pDnode->dir.mnode);
2021-12-01 09:27:31 +00:00
return -1;
2021-11-22 02:03:41 +00:00
}
taosWLockLatch(&pMgmt->latch);
pMgmt->pMnode = pMnode;
pMgmt->deployed = 1;
taosWUnLockLatch(&pMgmt->latch);
2021-12-01 09:27:31 +00:00
dInfo("mnode open successfully");
2021-11-22 02:03:41 +00:00
return 0;
}
2021-11-26 08:28:38 +00:00
static int32_t dndAlterMnode(SDnode *pDnode, SMnodeOpt *pOption) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
dError("failed to alter mnode since %s", terrstr());
return -1;
}
2021-11-29 04:09:18 +00:00
if (mndAlter(pMnode, pOption) != 0) {
2021-11-22 02:03:41 +00:00
dError("failed to alter mnode since %s", terrstr());
dndReleaseMnode(pDnode, pMnode);
return -1;
}
dndReleaseMnode(pDnode, pMnode);
return 0;
}
static int32_t dndDropMnode(SDnode *pDnode) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
dError("failed to drop mnode since %s", terrstr());
return -1;
}
taosRLockLatch(&pMgmt->latch);
pMgmt->dropped = 1;
taosRUnLockLatch(&pMgmt->latch);
if (dndWriteMnodeFile(pDnode) != 0) {
taosRLockLatch(&pMgmt->latch);
pMgmt->dropped = 0;
taosRUnLockLatch(&pMgmt->latch);
dndReleaseMnode(pDnode, pMnode);
dError("failed to drop mnode since %s", terrstr());
return -1;
}
2021-12-23 11:59:42 +00:00
dndReleaseMnode(pDnode, pMnode);
2021-11-22 02:03:41 +00:00
dndStopMnodeWorker(pDnode);
2021-12-28 09:38:15 +00:00
pMgmt->deployed = 0;
2021-11-22 02:03:41 +00:00
dndWriteMnodeFile(pDnode);
2021-11-29 04:09:18 +00:00
mndClose(pMnode);
2021-12-23 11:59:42 +00:00
pMgmt->pMnode = NULL;
2021-11-29 04:09:18 +00:00
mndDestroy(pDnode->dir.mnode);
2021-11-22 02:03:41 +00:00
return 0;
}
2022-01-07 10:03:28 +00:00
static SDCreateMnodeReq *dndParseCreateMnodeReq(SRpcMsg *pReq) {
SDCreateMnodeReq *pCreate = pReq->pCont;
2022-01-06 12:24:46 +00:00
pCreate->dnodeId = htonl(pCreate->dnodeId);
for (int32_t i = 0; i < pCreate->replica; ++i) {
pCreate->replicas[i].id = htonl(pCreate->replicas[i].id);
pCreate->replicas[i].port = htons(pCreate->replicas[i].port);
2021-11-22 02:03:41 +00:00
}
2022-01-06 12:24:46 +00:00
return pCreate;
2021-11-22 02:03:41 +00:00
}
2022-01-06 12:24:46 +00:00
int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
2022-01-07 10:03:28 +00:00
SDCreateMnodeReq *pCreate = dndParseCreateMnodeReq(pReq);
2021-11-22 02:03:41 +00:00
2022-01-07 10:03:28 +00:00
if (pCreate->replica <= 1 || pCreate->dnodeId != dndGetDnodeId(pDnode)) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
2022-01-07 10:30:20 +00:00
dError("failed to create mnode since %s", terrstr());
2021-11-22 02:03:41 +00:00
return -1;
2022-01-07 10:03:28 +00:00
}
2021-11-29 04:09:18 +00:00
2022-01-07 10:03:28 +00:00
SMnodeOpt option = {0};
if (dndBuildMnodeOptionFromReq(pDnode, &option, pCreate) != 0) {
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
2022-01-07 10:30:20 +00:00
dError("failed to create mnode since %s", terrstr());
2022-01-07 10:03:28 +00:00
return -1;
2021-11-22 02:03:41 +00:00
}
2022-01-07 10:03:28 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
dndReleaseMnode(pDnode, pMnode);
terrno = TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED;
2022-01-07 10:30:20 +00:00
dError("failed to create mnode since %s", terrstr());
2022-01-07 10:03:28 +00:00
return -1;
}
2022-01-07 10:30:20 +00:00
dDebug("start to create mnode");
2022-01-07 10:03:28 +00:00
return dndOpenMnode(pDnode, &option);
2021-11-22 02:03:41 +00:00
}
2022-01-06 12:24:46 +00:00
int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
2022-01-07 10:03:28 +00:00
SDAlterMnodeReq *pAlter = dndParseCreateMnodeReq(pReq);
2021-11-22 02:03:41 +00:00
2022-01-06 12:24:46 +00:00
if (pAlter->dnodeId != dndGetDnodeId(pDnode)) {
2022-01-07 10:03:28 +00:00
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
2022-01-07 10:30:20 +00:00
dError("failed to alter mnode since %s", terrstr());
2021-11-22 02:03:41 +00:00
return -1;
}
2021-12-23 09:47:21 +00:00
SMnodeOpt option = {0};
2022-01-06 12:24:46 +00:00
if (dndBuildMnodeOptionFromReq(pDnode, &option, pAlter) != 0) {
2022-01-07 10:03:28 +00:00
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
2022-01-07 10:30:20 +00:00
dError("failed to alter mnode since %s", terrstr());
2021-12-23 09:47:21 +00:00
return -1;
}
2022-01-07 10:03:28 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
terrno = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
2022-01-07 10:30:20 +00:00
dError("failed to alter mnode since %s", terrstr());
2021-12-23 09:47:21 +00:00
return -1;
}
2022-01-07 10:30:20 +00:00
dDebug("start to alter mnode");
2022-01-07 10:03:28 +00:00
int32_t code = dndAlterMnode(pDnode, &option);
dndReleaseMnode(pDnode, pMnode);
return code;
2021-11-22 02:03:41 +00:00
}
2022-01-06 12:24:46 +00:00
int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
2022-01-07 10:03:28 +00:00
SDDropMnodeReq *pDrop = pReq->pCont;
2022-01-06 12:24:46 +00:00
pDrop->dnodeId = htonl(pDrop->dnodeId);
2021-11-22 02:03:41 +00:00
2022-01-06 12:24:46 +00:00
if (pDrop->dnodeId != dndGetDnodeId(pDnode)) {
2022-01-07 10:03:28 +00:00
terrno = TSDB_CODE_DND_MNODE_INVALID_OPTION;
2022-01-07 10:30:20 +00:00
dError("failed to drop mnode since %s", terrstr());
2021-11-22 02:03:41 +00:00
return -1;
}
2022-01-07 10:03:28 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
2022-01-07 11:12:50 +00:00
if (pMnode == NULL) {
2022-01-07 10:03:28 +00:00
terrno = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
2022-01-07 10:30:20 +00:00
dError("failed to drop mnode since %s", terrstr());
2022-01-07 10:03:28 +00:00
return -1;
}
2022-01-07 10:30:20 +00:00
dDebug("start to drop mnode");
2022-01-07 10:03:28 +00:00
int32_t code = dndDropMnode(pDnode);
dndReleaseMnode(pDnode, pMnode);
return code;
2021-11-22 02:03:41 +00:00
}
2021-12-28 09:38:15 +00:00
static void dndProcessMnodeQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
2021-12-28 09:38:15 +00:00
mndProcessMsg(pMsg);
2021-11-22 02:03:41 +00:00
dndReleaseMnode(pDnode, pMnode);
} else {
2021-11-29 04:09:18 +00:00
mndSendRsp(pMsg, terrno);
2021-11-22 02:03:41 +00:00
}
2021-11-29 04:09:18 +00:00
mndCleanupMsg(pMsg);
2021-11-22 02:03:41 +00:00
}
2021-12-28 09:38:15 +00:00
static void dndWriteMnodeMsgToWorker(SDnode *pDnode, SDnodeWorker *pWorker, SRpcMsg *pRpcMsg) {
int32_t code = TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) {
2021-12-28 09:38:15 +00:00
SMnodeMsg *pMsg = mndInitMsg(pMnode, pRpcMsg);
if (pMsg == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
} else {
code = dndWriteMsgToWorker(pWorker, pMsg, 0);
2022-01-07 10:03:28 +00:00
if (code != 0) code = terrno;
2021-12-13 07:35:33 +00:00
}
2021-11-22 02:03:41 +00:00
2021-12-28 09:38:15 +00:00
if (code != 0) {
mndCleanupMsg(pMsg);
2021-12-13 07:35:33 +00:00
}
2021-11-22 02:03:41 +00:00
}
dndReleaseMnode(pDnode, pMnode);
2021-12-28 09:38:15 +00:00
if (code != 0) {
if (pRpcMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pRpcMsg->handle, .ahandle = pRpcMsg->ahandle, .code = code};
2021-12-13 07:35:33 +00:00
rpcSendResponse(&rsp);
}
2021-12-28 09:38:15 +00:00
rpcFreeCont(pRpcMsg->pCont);
2021-11-22 02:03:41 +00:00
}
}
2021-12-28 09:38:15 +00:00
void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.writeWorker, pMsg);
2021-11-22 02:03:41 +00:00
}
2021-12-28 09:38:15 +00:00
void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.syncWorker, pMsg);
2021-11-22 02:03:41 +00:00
}
2021-12-28 09:38:15 +00:00
void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndWriteMnodeMsgToWorker(pDnode, &pDnode->mmgmt.readWorker, pMsg);
2021-11-22 02:03:41 +00:00
}
int32_t dndInitMnode(SDnode *pDnode) {
dInfo("dnode-mnode start to init");
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
taosInitRWLatch(&pMgmt->latch);
if (dndReadMnodeFile(pDnode) != 0) {
return -1;
}
if (pMgmt->dropped) {
dInfo("mnode has been deployed and needs to be deleted");
2021-11-29 04:09:18 +00:00
mndDestroy(pDnode->dir.mnode);
2021-11-22 02:03:41 +00:00
return 0;
}
if (!pMgmt->deployed) {
bool needDeploy = dndNeedDeployMnode(pDnode);
if (!needDeploy) {
dDebug("mnode does not need to be deployed");
return 0;
}
dInfo("start to deploy mnode");
2021-11-26 08:28:38 +00:00
SMnodeOpt option = {0};
dndBuildMnodeDeployOption(pDnode, &option);
return dndOpenMnode(pDnode, &option);
2021-11-22 02:03:41 +00:00
} else {
dInfo("start to open mnode");
2021-11-26 08:28:38 +00:00
SMnodeOpt option = {0};
dndBuildMnodeOpenOption(pDnode, &option);
return dndOpenMnode(pDnode, &option);
2021-11-22 02:03:41 +00:00
}
}
void dndCleanupMnode(SDnode *pDnode) {
dInfo("dnode-mnode start to clean up");
2021-12-28 09:38:15 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
if (pMgmt->pMnode) {
dndStopMnodeWorker(pDnode);
mndClose(pMgmt->pMnode);
pMgmt->pMnode = NULL;
}
2021-11-22 02:03:41 +00:00
dInfo("dnode-mnode is cleaned up");
}
int32_t dndGetUserAuthFromMnode(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
2021-11-26 03:42:56 +00:00
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
2021-11-22 02:03:41 +00:00
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) {
terrno = TSDB_CODE_APP_NOT_READY;
dTrace("failed to get user auth since %s", terrstr());
return -1;
}
2021-11-29 04:09:18 +00:00
int32_t code = mndRetriveAuth(pMnode, user, spi, encrypt, secret, ckey);
2021-11-22 02:03:41 +00:00
dndReleaseMnode(pDnode, pMnode);
return code;
}