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

update authentications

This commit is contained in:
Shengliang Guan 2021-11-26 11:42:56 +08:00
parent 04ea23584f
commit b6c28a4914
5 changed files with 101 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mgmt/impl/inc/dndInt.h
View File

@ -109,9 +109,9 @@ typedef struct SDnode {
SDnodeOpt opt; SDnodeOpt opt;
SDnodeDir dir; SDnodeDir dir;
SDnodeMgmt dmgmt; SDnodeMgmt dmgmt;
SMnodeMgmt m; SMnodeMgmt mmgmt;
SVnodesMgmt vmgmt; SVnodesMgmt vmgmt;
STransMgmt t; STransMgmt tmgmt;
SStartupMsg startup; SStartupMsg startup;
} SDnode; } SDnode;

8
source/dnode/mgmt/impl/src/dndDnode.c
View File

@ -84,17 +84,17 @@ void dndSendRedirectMsg(SDnode *pDnode, SRpcMsg *pMsg) {
rpcSendRedirectRsp(pMsg->handle, &epSet); rpcSendRedirectRsp(pMsg->handle, &epSet);
} }
static void dndUpdateMnodeEpSet(SDnode *pDnd, SEpSet *pEpSet) { static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
dInfo("mnode is changed, num:%d inUse:%d", pEpSet->numOfEps, pEpSet->inUse); dInfo("mnode is changed, num:%d inUse:%d", pEpSet->numOfEps, pEpSet->inUse);
dndWLockDnode(pDnd); dndWLockDnode(pDnode);
pDnd->dmgmt.mnodeEpSet = *pEpSet; pDnode->dmgmt.mnodeEpSet = *pEpSet;
for (int32_t i = 0; i < pEpSet->numOfEps; ++i) { for (int32_t i = 0; i < pEpSet->numOfEps; ++i) {
dInfo("mnode index:%d %s:%u", i, pEpSet->fqdn[i], pEpSet->port[i]); dInfo("mnode index:%d %s:%u", i, pEpSet->fqdn[i], pEpSet->port[i]);
} }
dndWUnLockDnode(pDnd); dndWUnLockDnode(pDnode);
} }
static void dndPrintDnodes(SDnode *pDnode) { static void dndPrintDnodes(SDnode *pDnode) {

76
source/dnode/mgmt/impl/src/dndMnode.c
View File

@ -67,7 +67,7 @@ static int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
static int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg); static int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
static SMnode *dndAcquireMnode(SDnode *pDnode) { static SMnode *dndAcquireMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = NULL; SMnode *pMnode = NULL;
int32_t refCount = 0; int32_t refCount = 0;
@ -85,7 +85,7 @@ static SMnode *dndAcquireMnode(SDnode *pDnode) {
} }
static void dndReleaseMnode(SDnode *pDnode, SMnode *pMnode) { static void dndReleaseMnode(SDnode *pDnode, SMnode *pMnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
int32_t refCount = 0; int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
@ -98,7 +98,7 @@ static void dndReleaseMnode(SDnode *pDnode, SMnode *pMnode) {
} }
static int32_t dndReadMnodeFile(SDnode *pDnode) { static int32_t dndReadMnodeFile(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
int32_t code = TSDB_CODE_DND_MNODE_READ_FILE_ERROR; int32_t code = TSDB_CODE_DND_MNODE_READ_FILE_ERROR;
int32_t len = 0; int32_t len = 0;
int32_t maxLen = 300; int32_t maxLen = 300;
@ -152,7 +152,7 @@ PRASE_MNODE_OVER:
} }
static int32_t dndWriteMnodeFile(SDnode *pDnode) { static int32_t dndWriteMnodeFile(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
char file[PATH_MAX + 20] = {0}; char file[PATH_MAX + 20] = {0};
snprintf(file, sizeof(file), "%s.bak", pMgmt->file); snprintf(file, sizeof(file), "%s.bak", pMgmt->file);
@ -212,7 +212,7 @@ static int32_t dndStartMnodeWorker(SDnode *pDnode) {
} }
static void dndStopMnodeWorker(SDnode *pDnode) { static void dndStopMnodeWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
taosWLockLatch(&pMgmt->latch); taosWLockLatch(&pMgmt->latch);
pMgmt->deployed = 0; pMgmt->deployed = 0;
@ -296,7 +296,7 @@ static int32_t dndBuildMnodeOptions(SDnode *pDnode, SMnodeOptions *pOptions, SCr
} }
static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOptions *pOptions) { static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOptions *pOptions) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
int32_t code = dndStartMnodeWorker(pDnode); int32_t code = dndStartMnodeWorker(pDnode);
if (code != 0) { if (code != 0) {
@ -332,7 +332,7 @@ static int32_t dndOpenMnode(SDnode *pDnode, SMnodeOptions *pOptions) {
} }
static int32_t dndAlterMnode(SDnode *pDnode, SMnodeOptions *pOptions) { static int32_t dndAlterMnode(SDnode *pDnode, SMnodeOptions *pOptions) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) { if (pMnode == NULL) {
@ -351,7 +351,7 @@ static int32_t dndAlterMnode(SDnode *pDnode, SMnodeOptions *pOptions) {
} }
static int32_t dndDropMnode(SDnode *pDnode) { static int32_t dndDropMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) { if (pMnode == NULL) {
@ -458,7 +458,7 @@ static void dndProcessMnodeMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg) {
} }
static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg) { static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) { if (pMnode != NULL) {
@ -472,7 +472,7 @@ static void dndProcessMnodeReadQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
} }
static void dndProcessMnodeWriteQueue(SDnode *pDnode, SMnodeMsg *pMsg) { static void dndProcessMnodeWriteQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) { if (pMnode != NULL) {
@ -486,7 +486,7 @@ static void dndProcessMnodeWriteQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
} }
static void dndProcessMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) { static void dndProcessMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) { if (pMnode != NULL) {
@ -500,7 +500,7 @@ static void dndProcessMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
} }
static void dndProcessMnodeSyncQueue(SDnode *pDnode, SMnodeMsg *pMsg) { static void dndProcessMnodeSyncQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode != NULL) { if (pMnode != NULL) {
@ -532,7 +532,7 @@ static int32_t dndWriteMnodeMsgToQueue(SMnode *pMnode, taos_queue pQueue, SRpcMs
} }
void dndProcessMnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pRpcMsg, SEpSet *pEpSet) { void dndProcessMnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pRpcMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg)); SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg));
@ -545,7 +545,7 @@ void dndProcessMnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pRpcMsg, SEpSet *pEpSet) {
} }
void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) { void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pWriteQ, pMsg) != 0) { if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pWriteQ, pMsg) != 0) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno}; SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
@ -557,7 +557,7 @@ void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) { void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) { if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno}; SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
@ -569,7 +569,7 @@ void dndProcessMnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) { void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) { if (pMnode == NULL || dndWriteMnodeMsgToQueue(pMnode, pMgmt->pSyncQ, pMsg) != 0) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno}; SRpcMsg rsp = {.handle = pMsg->handle, .code = terrno};
@ -581,7 +581,7 @@ void dndProcessMnodeReadMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
static int32_t dndPutMsgIntoMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) { static int32_t dndPutMsgIntoMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) { if (pMnode == NULL) {
@ -594,7 +594,7 @@ static int32_t dndPutMsgIntoMnodeApplyQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
} }
static int32_t dndAllocMnodeMgmtQueue(SDnode *pDnode) { static int32_t dndAllocMnodeMgmtQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pMgmtQ = tWorkerAllocQueue(&pMgmt->mgmtPool, NULL, (FProcessItem)dndProcessMnodeMgmtQueue); pMgmt->pMgmtQ = tWorkerAllocQueue(&pMgmt->mgmtPool, NULL, (FProcessItem)dndProcessMnodeMgmtQueue);
if (pMgmt->pMgmtQ == NULL) { if (pMgmt->pMgmtQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -604,13 +604,13 @@ static int32_t dndAllocMnodeMgmtQueue(SDnode *pDnode) {
} }
static void dndFreeMnodeMgmtQueue(SDnode *pDnode) { static void dndFreeMnodeMgmtQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->mgmtPool, pMgmt->pMgmtQ); tWorkerFreeQueue(&pMgmt->mgmtPool, pMgmt->pMgmtQ);
pMgmt->pMgmtQ = NULL; pMgmt->pMgmtQ = NULL;
} }
static int32_t dndInitMnodeMgmtWorker(SDnode *pDnode) { static int32_t dndInitMnodeMgmtWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->mgmtPool; SWorkerPool *pPool = &pMgmt->mgmtPool;
pPool->name = "mnode-mgmt"; pPool->name = "mnode-mgmt";
pPool->min = 1; pPool->min = 1;
@ -624,13 +624,13 @@ static int32_t dndInitMnodeMgmtWorker(SDnode *pDnode) {
} }
static void dndCleanupMnodeMgmtWorker(SDnode *pDnode) { static void dndCleanupMnodeMgmtWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
; ;
tWorkerCleanup(&pMgmt->mgmtPool); tWorkerCleanup(&pMgmt->mgmtPool);
} }
static int32_t dndAllocMnodeReadQueue(SDnode *pDnode) { static int32_t dndAllocMnodeReadQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pReadQ = tWorkerAllocQueue(&pMgmt->readPool, NULL, (FProcessItem)dndProcessMnodeReadQueue); pMgmt->pReadQ = tWorkerAllocQueue(&pMgmt->readPool, NULL, (FProcessItem)dndProcessMnodeReadQueue);
if (pMgmt->pReadQ == NULL) { if (pMgmt->pReadQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -640,13 +640,13 @@ static int32_t dndAllocMnodeReadQueue(SDnode *pDnode) {
} }
static void dndFreeMnodeReadQueue(SDnode *pDnode) { static void dndFreeMnodeReadQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->readPool, pMgmt->pReadQ); tWorkerFreeQueue(&pMgmt->readPool, pMgmt->pReadQ);
pMgmt->pReadQ = NULL; pMgmt->pReadQ = NULL;
} }
static int32_t dndInitMnodeReadWorker(SDnode *pDnode) { static int32_t dndInitMnodeReadWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->readPool; SWorkerPool *pPool = &pMgmt->readPool;
pPool->name = "mnode-read"; pPool->name = "mnode-read";
pPool->min = 0; pPool->min = 0;
@ -660,12 +660,12 @@ static int32_t dndInitMnodeReadWorker(SDnode *pDnode) {
} }
static void dndCleanupMnodeReadWorker(SDnode *pDnode) { static void dndCleanupMnodeReadWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->readPool); tWorkerCleanup(&pMgmt->readPool);
} }
static int32_t dndAllocMnodeWriteQueue(SDnode *pDnode) { static int32_t dndAllocMnodeWriteQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pWriteQ = tWorkerAllocQueue(&pMgmt->writePool, NULL, (FProcessItem)dndProcessMnodeWriteQueue); pMgmt->pWriteQ = tWorkerAllocQueue(&pMgmt->writePool, NULL, (FProcessItem)dndProcessMnodeWriteQueue);
if (pMgmt->pWriteQ == NULL) { if (pMgmt->pWriteQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -675,13 +675,13 @@ static int32_t dndAllocMnodeWriteQueue(SDnode *pDnode) {
} }
static void dndFreeMnodeWriteQueue(SDnode *pDnode) { static void dndFreeMnodeWriteQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->writePool, pMgmt->pWriteQ); tWorkerFreeQueue(&pMgmt->writePool, pMgmt->pWriteQ);
pMgmt->pWriteQ = NULL; pMgmt->pWriteQ = NULL;
} }
static int32_t dndAllocMnodeApplyQueue(SDnode *pDnode) { static int32_t dndAllocMnodeApplyQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pApplyQ = tWorkerAllocQueue(&pMgmt->writePool, NULL, (FProcessItem)dndProcessMnodeApplyQueue); pMgmt->pApplyQ = tWorkerAllocQueue(&pMgmt->writePool, NULL, (FProcessItem)dndProcessMnodeApplyQueue);
if (pMgmt->pApplyQ == NULL) { if (pMgmt->pApplyQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -691,13 +691,13 @@ static int32_t dndAllocMnodeApplyQueue(SDnode *pDnode) {
} }
static void dndFreeMnodeApplyQueue(SDnode *pDnode) { static void dndFreeMnodeApplyQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->writePool, pMgmt->pApplyQ); tWorkerFreeQueue(&pMgmt->writePool, pMgmt->pApplyQ);
pMgmt->pApplyQ = NULL; pMgmt->pApplyQ = NULL;
} }
static int32_t dndInitMnodeWriteWorker(SDnode *pDnode) { static int32_t dndInitMnodeWriteWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->writePool; SWorkerPool *pPool = &pMgmt->writePool;
pPool->name = "mnode-write"; pPool->name = "mnode-write";
pPool->min = 0; pPool->min = 0;
@ -711,12 +711,12 @@ static int32_t dndInitMnodeWriteWorker(SDnode *pDnode) {
} }
static void dndCleanupMnodeWriteWorker(SDnode *pDnode) { static void dndCleanupMnodeWriteWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->writePool); tWorkerCleanup(&pMgmt->writePool);
} }
static int32_t dndAllocMnodeSyncQueue(SDnode *pDnode) { static int32_t dndAllocMnodeSyncQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->pSyncQ = tWorkerAllocQueue(&pMgmt->syncPool, NULL, (FProcessItem)dndProcessMnodeSyncQueue); pMgmt->pSyncQ = tWorkerAllocQueue(&pMgmt->syncPool, NULL, (FProcessItem)dndProcessMnodeSyncQueue);
if (pMgmt->pSyncQ == NULL) { if (pMgmt->pSyncQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -726,13 +726,13 @@ static int32_t dndAllocMnodeSyncQueue(SDnode *pDnode) {
} }
static void dndFreeMnodeSyncQueue(SDnode *pDnode) { static void dndFreeMnodeSyncQueue(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerFreeQueue(&pMgmt->syncPool, pMgmt->pSyncQ); tWorkerFreeQueue(&pMgmt->syncPool, pMgmt->pSyncQ);
pMgmt->pSyncQ = NULL; pMgmt->pSyncQ = NULL;
} }
static int32_t dndInitMnodeSyncWorker(SDnode *pDnode) { static int32_t dndInitMnodeSyncWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SWorkerPool *pPool = &pMgmt->syncPool; SWorkerPool *pPool = &pMgmt->syncPool;
pPool->name = "mnode-sync"; pPool->name = "mnode-sync";
pPool->min = 0; pPool->min = 0;
@ -741,13 +741,13 @@ static int32_t dndInitMnodeSyncWorker(SDnode *pDnode) {
} }
static void dndCleanupMnodeSyncWorker(SDnode *pDnode) { static void dndCleanupMnodeSyncWorker(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
tWorkerCleanup(&pMgmt->syncPool); tWorkerCleanup(&pMgmt->syncPool);
} }
int32_t dndInitMnode(SDnode *pDnode) { int32_t dndInitMnode(SDnode *pDnode) {
dInfo("dnode-mnode start to init"); dInfo("dnode-mnode start to init");
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
taosInitRWLatch(&pMgmt->latch); taosInitRWLatch(&pMgmt->latch);
if (dndInitMnodeMgmtWorker(pDnode) != 0) { if (dndInitMnodeMgmtWorker(pDnode) != 0) {
@ -791,7 +791,7 @@ int32_t dndInitMnode(SDnode *pDnode) {
} }
void dndCleanupMnode(SDnode *pDnode) { void dndCleanupMnode(SDnode *pDnode) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
dInfo("dnode-mnode start to clean up"); dInfo("dnode-mnode start to clean up");
dndStopMnodeWorker(pDnode); dndStopMnodeWorker(pDnode);
@ -801,7 +801,7 @@ void dndCleanupMnode(SDnode *pDnode) {
} }
int32_t dndGetUserAuthFromMnode(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) { int32_t dndGetUserAuthFromMnode(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
SMnodeMgmt *pMgmt = &pDnode->m; SMnodeMgmt *pMgmt = &pDnode->mmgmt;
SMnode *pMnode = dndAcquireMnode(pDnode); SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) { if (pMnode == NULL) {

78
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -25,6 +25,10 @@
#include "dndMnode.h" #include "dndMnode.h"
#include "dndVnodes.h" #include "dndVnodes.h"
#define INTERNAL_USER "_internal"
#define INTERNAL_CKEY "_key"
#define INTERNAL_SECRET "_secret"
static void dndInitMsgFp(STransMgmt *pMgmt) { static void dndInitMsgFp(STransMgmt *pMgmt) {
// msg from client to dnode // msg from client to dnode
pMgmt->msgFp[TSDB_MSG_TYPE_SUBMIT] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_SUBMIT] = dndProcessVnodeWriteMsg;
@ -121,7 +125,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
SDnode *pDnode = parent; SDnode *pDnode = parent;
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
int32_t msgType = pMsg->msgType; int32_t msgType = pMsg->msgType;
@ -143,19 +147,19 @@ static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
static int32_t dndInitClient(SDnode *pDnode) { static int32_t dndInitClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
SRpcInit rpcInit; SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.label = "DND-C"; rpcInit.label = "DND-C";
rpcInit.numOfThreads = 1; rpcInit.numOfThreads = 1;
rpcInit.cfp = dndProcessResponse; rpcInit.cfp = dndProcessResponse;
rpcInit.sessions = TSDB_MAX_VNODES << 4; rpcInit.sessions = 8;
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000; rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000;
rpcInit.user = "-internal"; rpcInit.user = INTERNAL_USER;
rpcInit.ckey = "-key"; rpcInit.ckey = INTERNAL_CKEY;
rpcInit.secret = "-secret"; rpcInit.secret = INTERNAL_SECRET;
pMgmt->clientRpc = rpcOpen(&rpcInit); pMgmt->clientRpc = rpcOpen(&rpcInit);
if (pMgmt->clientRpc == NULL) { if (pMgmt->clientRpc == NULL) {
@ -167,7 +171,7 @@ static int32_t dndInitClient(SDnode *pDnode) {
} }
static void dndCleanupClient(SDnode *pDnode) { static void dndCleanupClient(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
if (pMgmt->clientRpc) { if (pMgmt->clientRpc) {
rpcClose(pMgmt->clientRpc); rpcClose(pMgmt->clientRpc);
pMgmt->clientRpc = NULL; pMgmt->clientRpc = NULL;
@ -176,8 +180,8 @@ static void dndCleanupClient(SDnode *pDnode) {
} }
static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) { static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) {
SDnode *pDnode = param; SDnode *pDnode = param;
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
int32_t msgType = pMsg->msgType; int32_t msgType = pMsg->msgType;
if (msgType == TSDB_MSG_TYPE_NETWORK_TEST) { if (msgType == TSDB_MSG_TYPE_NETWORK_TEST) {
@ -218,24 +222,56 @@ static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) { static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
SEpSet epSet = {0}; SEpSet epSet = {0};
dndGetMnodeEpSet(pDnode, &epSet); dndGetMnodeEpSet(pDnode, &epSet);
rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp); rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp);
} }
static int32_t dndAuthInternalMsg(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
if (strcmp(user, INTERNAL_USER) == 0) {
// A simple temporary implementation
char pass[32] = {0};
taosEncryptPass((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
memcpy(secret, pass, TSDB_KEY_LEN);
*spi = 0;
*encrypt = 0;
*ckey = 0;
return 0;
} else if (strcmp(user, TSDB_NETTEST_USER) == 0) {
// A simple temporary implementation
char pass[32] = {0};
taosEncryptPass((uint8_t *)(TSDB_NETTEST_USER), strlen(TSDB_NETTEST_USER), pass);
memcpy(secret, pass, TSDB_KEY_LEN);
*spi = 0;
*encrypt = 0;
*ckey = 0;
return 0;
} else {
return -1;
}
}
static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey) { static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
SDnode *pDnode = parent; SDnode *pDnode = parent;
if (dndGetUserAuthFromMnode(pDnode, user, spi, encrypt, secret, ckey) != 0) { if (dndAuthInternalMsg(parent, user, spi, encrypt, secret, ckey) == 0) {
if (terrno != TSDB_CODE_APP_NOT_READY) { dTrace("get internal auth success");
dTrace("failed to get user auth from mnode since %s", terrstr()); return 0;
return -1;
}
} }
dDebug("user:%s, send auth msg to mnodes", user); if (dndGetUserAuthFromMnode(pDnode, user, spi, encrypt, secret, ckey) == 0) {
dTrace("get auth from internal mnode");
return 0;
}
if (terrno != TSDB_CODE_APP_NOT_READY) {
dTrace("failed to get user auth from internal mnode since %s", terrstr());
return -1;
}
dDebug("user:%s, send auth msg to other mnodes", user);
SAuthMsg *pMsg = rpcMallocCont(sizeof(SAuthMsg)); SAuthMsg *pMsg = rpcMallocCont(sizeof(SAuthMsg));
tstrncpy(pMsg->user, user, TSDB_USER_LEN); tstrncpy(pMsg->user, user, TSDB_USER_LEN);
@ -246,14 +282,14 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
if (rpcRsp.code != 0) { if (rpcRsp.code != 0) {
terrno = rpcRsp.code; terrno = rpcRsp.code;
dError("user:%s, failed to get user auth from mnodes since %s", user, terrstr()); dError("user:%s, failed to get user auth from other mnodes since %s", user, terrstr());
} else { } else {
SAuthRsp *pRsp = rpcRsp.pCont; SAuthRsp *pRsp = rpcRsp.pCont;
memcpy(secret, pRsp->secret, TSDB_KEY_LEN); memcpy(secret, pRsp->secret, TSDB_KEY_LEN);
memcpy(ckey, pRsp->ckey, TSDB_KEY_LEN); memcpy(ckey, pRsp->ckey, TSDB_KEY_LEN);
*spi = pRsp->spi; *spi = pRsp->spi;
*encrypt = pRsp->encrypt; *encrypt = pRsp->encrypt;
dDebug("user:%s, success to get user auth from mnodes", user); dDebug("user:%s, success to get user auth from other mnodes", user);
} }
rpcFreeCont(rpcRsp.pCont); rpcFreeCont(rpcRsp.pCont);
@ -261,7 +297,7 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
} }
static int32_t dndInitServer(SDnode *pDnode) { static int32_t dndInitServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
dndInitMsgFp(pMgmt); dndInitMsgFp(pMgmt);
int32_t numOfThreads = (int32_t)((pDnode->opt.numOfCores * pDnode->opt.numOfThreadsPerCore) / 2.0); int32_t numOfThreads = (int32_t)((pDnode->opt.numOfCores * pDnode->opt.numOfThreadsPerCore) / 2.0);
@ -290,7 +326,7 @@ static int32_t dndInitServer(SDnode *pDnode) {
} }
static void dndCleanupServer(SDnode *pDnode) { static void dndCleanupServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
if (pMgmt->serverRpc) { if (pMgmt->serverRpc) {
rpcClose(pMgmt->serverRpc); rpcClose(pMgmt->serverRpc);
pMgmt->serverRpc = NULL; pMgmt->serverRpc = NULL;
@ -317,7 +353,7 @@ void dndCleanupTrans(SDnode *pDnode) {
} }
void dndSendMsgToDnode(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pMsg) { void dndSendMsgToDnode(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pMsg) {
STransMgmt *pMgmt = &pDnode->t; STransMgmt *pMgmt = &pDnode->tmgmt;
rpcSendRequest(pMgmt->clientRpc, pEpSet, pMsg, NULL); rpcSendRequest(pMgmt->clientRpc, pEpSet, pMsg, NULL);
} }

11
source/dnode/mnode/impl/src/mnodeAuth.c
View File

@ -21,16 +21,5 @@ int32_t mnodeInitAuth() { return 0; }
void mnodeCleanupAuth() {} void mnodeCleanupAuth() {}
int32_t mnodeRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) { int32_t mnodeRetriveAuth(SMnode *pMnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
if (strcmp(user, TSDB_NETTEST_USER) == 0) {
char pass[32] = {0};
taosEncryptPass((uint8_t *)user, strlen(user), pass);
*spi = 0;
*encrypt = 0;
*ckey = 0;
memcpy(secret, pass, TSDB_KEY_LEN);
mDebug("nettest user is authorized");
return 0;
}
return 0; return 0;
} }