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

TD-10431 dnode test01

This commit is contained in:
Shengliang Guan 2021-12-03 20:52:44 +08:00
parent 89270a08bf
commit 608b96d384
7 changed files with 330 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mgmt/impl/CMakeLists.txt
View File

@ -14,3 +14,7 @@ target_include_directories(
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mgmt"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

1
source/dnode/mgmt/impl/test/CMakeLists.txt Normal file
View File

@ -0,0 +1 @@
add_subdirectory(test01)

25
source/dnode/mgmt/impl/test/test01/CMakeLists.txt Normal file
View File

@ -0,0 +1,25 @@
add_executable(dndTest01 "")
target_sources(dndTest01
PRIVATE
"dndTest01.cpp"
)
target_link_libraries(
dndTest01
PUBLIC dnode
PUBLIC gtest_main
)
target_include_directories(dndTest01
PUBLIC
"${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
"${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
)
enable_testing()
add_test(
NAME dndTest01
COMMAND dndTest01
)

0
source/dnode/mgmt/impl/test/test01/deploy.cpp Normal file
View File

156
source/dnode/mgmt/impl/test/test01/dndTest01.cpp Normal file
View File

@ -0,0 +1,156 @@
/*
* 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/>.
*/
#include <gtest/gtest.h>
#include <cstring>
#include <iostream>
#include <queue>
#include "tthread.h"
#include "dnode.h"
#include "trpc.h"
typedef struct {
SDnode* pDnode;
pthread_t* threadId;
} SServer;
void* runServer(void* param) {
SServer* pServer = param;
while (1) {
taosMsleep(100);
pthread_testcancel();
}
}
void initOption(SDnodeOpt* pOption) {
pOption->sver = 1;
pOption->numOfCores = 1;
pOption->numOfSupportMnodes = 1;
pOption->numOfSupportVnodes = 1;
pOption->numOfSupportQnodes = 1;
pOption->statusInterval = 1;
pOption->mnodeEqualVnodeNum = 1;
pOption->numOfThreadsPerCore = 1;
pOption->ratioOfQueryCores = 1;
pOption->maxShellConns = 1000;
pOption->shellActivityTimer = 30;
pOption->serverPort = 9527;
strncpy(pOption->dataDir, "./test01");
strncpy(pOption->localEp, "localhost:9527");
strncpy(pOption->localFqdn, "localhost");
tstrncpy(pOption->firstEp, "localhost:9527");
}
Server* createServer() {
SDnodeOpt option = {0};
initOption(&option);
SDnode* pDnode = dndInit(&option);
ASSERT(pDnode);
Server* pServer = calloc(1, sizeof(SServer));
ASSERT(pServer);
pServer->pDnode = pDnode;
pServer->threadId = taosCreateThread(runServer, pServer);
ASSERT(pServer->threadId);
return pServer;
}
void dropServer(SServer* pServer) {
if (pServer->threadId != NULL) {
taosDestoryThread(pServer->threadId);
}
}
typedef struct {
void* clientRpc;
SRpcMsg* pRsp;
tsem_t sem;
} SClient;
static void processClientRsp(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
SClient* pClient = parent;
pClient->pRsp = pMsg;
tsem_post(pMgmt->clientRpc);
}
SClient* createClient() {
SClient* pClient = calloc(1, sizeof(SClient));
ASSERT(pClient);
char secretEncrypt[32] = {0};
char* pass = "taosdata";
taosEncryptPass((uint8_t*)pass, strlen(pass), secretEncrypt);
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.label = "DND-C";
rpcInit.numOfThreads = 1;
rpcInit.cfp = processClientRsp;
rpcInit.sessions = 1024;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = 30 * 1000;
rpcInit.user = "root";
rpcInit.ckey = "key";
rpcInit.parent = pDnode;
rpcInit.secret = (char*)secretEncrypt;
rpcInit.parent = pClient;
// rpcInit.spi = 1;
pClient->clientRpc = rpcOpen(&rpcInit);
ASSERT(pClient->clientRpc);
tsem_init(&pClient->sem, 0, 0);
}
void dropClient(SClient* pClient) {
tsem_destroy(&pClient->sem);
rpcClose(pClient->clientRpc);
}
void sendMsg(SClient* pClient, SRpcMsg* pMsg) {
SEpSet epSet = {0};
epSet.inUse = 0;
epSet.numOfEps = 1;
epSet.port[0] = 9527;
strcpy(epSet.fqdn[0], "localhost");
rpcSendRequest(pMgmt->clientRpc, &epSet, pMsg, NULL);
tsem_wait(pMgmt->clientRpc);
}
class DndTest01 : public ::testing::Test {
protected:
void SetUp() override {
pServer = createServer();
pClient = createClient();
}
void TearDown() override {
dropServer(pServer);
dropClient(pClient);
}
SServer* pServer;
SClient* pClient;
};
TEST_F(DndTest01, connectMsg) {
SConnectMsg *pReq = rpcMallocCont()
}

99
source/dnode/mnode/impl/src/mndAcct.c
View File

@ -14,10 +14,61 @@
*/
#define _DEFAULT_SOURCE
#include "mndInt.h"
#include "mndAcct.h"
#include "mndShow.h"
#define SDB_ACCT_VER 1
static int32_t mnodeCreateDefaultAcct(SMnode *pMnode);
static SSdbRaw *mnodeAcctActionEncode(SAcctObj *pAcct);
static SSdbRow *mnodeAcctActionDecode(SSdbRaw *pRaw);
static int32_t mnodeAcctActionInsert(SSdb *pSdb, SAcctObj *pAcct);
static int32_t mnodeAcctActionDelete(SSdb *pSdb, SAcctObj *pAcct);
static int32_t mnodeAcctActionUpdate(SSdb *pSdb, SAcctObj *pSrcAcct, SAcctObj *pDstAcct);
static int32_t mndProcessCreateAcctMsg(SMnodeMsg *pMnodeMsg);
static int32_t mndProcessAlterAcctMsg(SMnodeMsg *pMnodeMsg);
static int32_t mndProcessDropAcctMsg(SMnodeMsg *pMnodeMsg);
int32_t mndInitAcct(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_ACCT,
.keyType = SDB_KEY_BINARY,
.deployFp = mnodeCreateDefaultAcct,
.encodeFp = (SdbEncodeFp)mnodeAcctActionEncode,
.decodeFp = (SdbDecodeFp)mnodeAcctActionDecode,
.insertFp = (SdbInsertFp)mnodeAcctActionInsert,
.updateFp = (SdbUpdateFp)mnodeAcctActionUpdate,
.deleteFp = (SdbDeleteFp)mnodeAcctActionDelete};
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CREATE_ACCT, mndProcessCreateAcctMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_ALTER_ACCT, mndProcessAlterAcctMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_DROP_ACCT, mndProcessDropAcctMsg);
return sdbSetTable(pMnode->pSdb, table);
}
void mndCleanupAcct(SMnode *pMnode) {}
static int32_t mnodeCreateDefaultAcct(SMnode *pMnode) {
SAcctObj acctObj = {0};
tstrncpy(acctObj.acct, TSDB_DEFAULT_USER, TSDB_USER_LEN);
acctObj.createdTime = taosGetTimestampMs();
acctObj.updateTime = acctObj.createdTime;
acctObj.acctId = 1;
acctObj.cfg = (SAcctCfg){.maxUsers = 1024,
.maxDbs = 1024,
.maxTimeSeries = INT32_MAX,
.maxStreams = 8092,
.maxStorage = INT64_MAX,
.accessState = TSDB_VN_ALL_ACCCESS};
SSdbRaw *pRaw = mnodeAcctActionEncode(&acctObj);
if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mTrace("acct:%s, will be created while deploy sdb", acctObj.acct);
return sdbWrite(pMnode->pSdb, pRaw);
}
static SSdbRaw *mnodeAcctActionEncode(SAcctObj *pAcct) {
SSdbRaw *pRaw = sdbAllocRaw(SDB_ACCT, SDB_ACCT_VER, sizeof(SAcctObj));
if (pRaw == NULL) return NULL;
@ -92,40 +143,20 @@ static int32_t mnodeAcctActionUpdate(SSdb *pSdb, SAcctObj *pSrcAcct, SAcctObj *p
return 0;
}
static int32_t mnodeCreateDefaultAcct(SMnode *pMnode) {
int32_t code = 0;
SAcctObj acctObj = {0};
tstrncpy(acctObj.acct, TSDB_DEFAULT_USER, TSDB_USER_LEN);
acctObj.createdTime = taosGetTimestampMs();
acctObj.updateTime = acctObj.createdTime;
acctObj.acctId = 1;
acctObj.cfg = (SAcctCfg){.maxUsers = 1024,
.maxDbs = 1024,
.maxTimeSeries = INT32_MAX,
.maxStreams = 8092,
.maxStorage = INT64_MAX,
.accessState = TSDB_VN_ALL_ACCCESS};
SSdbRaw *pRaw = mnodeAcctActionEncode(&acctObj);
if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mTrace("acct:%s, will be created while deploy sdb", acctObj.acct);
return sdbWrite(pMnode->pSdb, pRaw);
static int32_t mndProcessCreateAcctMsg(SMnodeMsg *pMnodeMsg) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct msg since %s", terrstr());
return -1;
}
int32_t mndInitAcct(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_ACCT,
.keyType = SDB_KEY_BINARY,
.deployFp = mnodeCreateDefaultAcct,
.encodeFp = (SdbEncodeFp)mnodeAcctActionEncode,
.decodeFp = (SdbDecodeFp)mnodeAcctActionDecode,
.insertFp = (SdbInsertFp)mnodeAcctActionInsert,
.updateFp = (SdbUpdateFp)mnodeAcctActionUpdate,
.deleteFp = (SdbDeleteFp)mnodeAcctActionDelete};
return sdbSetTable(pMnode->pSdb, table);
static int32_t mndProcessAlterAcctMsg(SMnodeMsg *pMnodeMsg) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct msg since %s", terrstr());
return -1;
}
void mndCleanupAcct(SMnode *pMnode) {}
static int32_t mndProcessDropAcctMsg(SMnodeMsg *pMnodeMsg) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process create acct msg since %s", terrstr());
return -1;
}

132
source/dnode/mnode/impl/src/mndUser.c
View File

@ -14,12 +14,76 @@
*/
#define _DEFAULT_SOURCE
#include "mndUser.h"
#include "mndShow.h"
#include "mndSync.h"
#include "mndTrans.h"
#include "tkey.h"
#define SDB_USER_VER 1
static int32_t mndCreateDefaultUsers(SMnode *pMnode);
static SSdbRaw *mndUserActionEncode(SUserObj *pUser);
static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw);
static int32_t mndUserActionInsert(SSdb *pSdb, SUserObj *pUser);
static int32_t mndUserActionDelete(SSdb *pSdb, SUserObj *pUser);
static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pSrcUser, SUserObj *pDstUser);
static int32_t mndCreateUser(SMnode *pMnode, char *acct, char *user, char *pass, SMnodeMsg *pMsg);
static int32_t mndProcessCreateUserMsg(SMnodeMsg *pMsg);
static int32_t mndProcessAlterUserMsg(SMnodeMsg *pMsg);
static int32_t mndProcessDropUserMsg(SMnodeMsg *pMsg);
int32_t mndInitUser(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_USER,
.keyType = SDB_KEY_BINARY,
.deployFp = (SdbDeployFp)mndCreateDefaultUsers,
.encodeFp = (SdbEncodeFp)mndUserActionEncode,
.decodeFp = (SdbDecodeFp)mndUserActionDecode,
.insertFp = (SdbInsertFp)mndUserActionInsert,
.updateFp = (SdbUpdateFp)mndUserActionUpdate,
.deleteFp = (SdbDeleteFp)mndUserActionDelete};
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CREATE_USER, mndProcessCreateUserMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_ALTER_USER, mndProcessAlterUserMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_DROP_USER, mndProcessDropUserMsg);
return sdbSetTable(pMnode->pSdb, table);
}
void mndCleanupUser(SMnode *pMnode) {}
static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char *pass) {
SUserObj userObj = {0};
tstrncpy(userObj.user, user, TSDB_USER_LEN);
tstrncpy(userObj.acct, acct, TSDB_USER_LEN);
taosEncryptPass((uint8_t *)pass, strlen(pass), userObj.pass);
userObj.createdTime = taosGetTimestampMs();
userObj.updateTime = userObj.createdTime;
if (strcmp(user, TSDB_DEFAULT_USER) == 0) {
userObj.superAuth = 1;
}
SSdbRaw *pRaw = mndUserActionEncode(&userObj);
if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mTrace("user:%s, will be created while deploy sdb", userObj.user);
return sdbWrite(pMnode->pSdb, pRaw);
}
static int32_t mndCreateDefaultUsers(SMnode *pMnode) {
if (mndCreateDefaultUser(pMnode, TSDB_DEFAULT_USER, TSDB_DEFAULT_USER, TSDB_DEFAULT_PASS) != 0) {
return -1;
}
if (mndCreateDefaultUser(pMnode, TSDB_DEFAULT_USER, "_" TSDB_DEFAULT_USER, TSDB_DEFAULT_PASS) != 0) {
return -1;
}
return 0;
}
static SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
SSdbRaw *pRaw = sdbAllocRaw(SDB_USER, SDB_USER_VER, sizeof(SUserObj));
if (pRaw == NULL) return NULL;
@ -103,36 +167,14 @@ static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pSrcUser, SUserObj *pDs
return 0;
}
static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char *pass) {
SUserObj userObj = {0};
tstrncpy(userObj.user, user, TSDB_USER_LEN);
tstrncpy(userObj.acct, acct, TSDB_USER_LEN);
taosEncryptPass((uint8_t *)pass, strlen(pass), userObj.pass);
userObj.createdTime = taosGetTimestampMs();
userObj.updateTime = userObj.createdTime;
if (strcmp(user, TSDB_DEFAULT_USER) == 0) {
userObj.superAuth = 1;
}
SSdbRaw *pRaw = mndUserActionEncode(&userObj);
if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mTrace("user:%s, will be created while deploy sdb", userObj.user);
return sdbWrite(pMnode->pSdb, pRaw);
SUserObj *mndAcquireUser(SMnode *pMnode, const char *userName) {
SSdb *pSdb = pMnode->pSdb;
return sdbAcquire(pSdb, SDB_USER, &userName);
}
static int32_t mndCreateDefaultUsers(SMnode *pMnode) {
if (mndCreateDefaultUser(pMnode, TSDB_DEFAULT_USER, TSDB_DEFAULT_USER, TSDB_DEFAULT_PASS) != 0) {
return -1;
}
if (mndCreateDefaultUser(pMnode, TSDB_DEFAULT_USER, "_" TSDB_DEFAULT_USER, TSDB_DEFAULT_PASS) != 0) {
return -1;
}
return 0;
void mndReleaseUser(SMnode *pMnode, SUserObj *pUser) {
SSdb *pSdb = pMnode->pSdb;
sdbRelease(pSdb, pUser);
}
static int32_t mndCreateUser(SMnode *pMnode, char *acct, char *user, char *pass, SMnodeMsg *pMsg) {
@ -222,30 +264,14 @@ static int32_t mndProcessCreateUserMsg(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
int32_t mndInitUser(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_USER,
.keyType = SDB_KEY_BINARY,
.deployFp = (SdbDeployFp)mndCreateDefaultUsers,
.encodeFp = (SdbEncodeFp)mndUserActionEncode,
.decodeFp = (SdbDecodeFp)mndUserActionDecode,
.insertFp = (SdbInsertFp)mndUserActionInsert,
.updateFp = (SdbUpdateFp)mndUserActionUpdate,
.deleteFp = (SdbDeleteFp)mndUserActionDelete};
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CREATE_USER, mndProcessCreateUserMsg);
return sdbSetTable(pMnode->pSdb, table);
}
void mndCleanupUser(SMnode *pMnode) {}
SUserObj *mndAcquireUser(SMnode *pMnode, const char *userName) {
SSdb *pSdb = pMnode->pSdb;
return sdbAcquire(pSdb, SDB_USER, &userName);
}
void mndReleaseUser(SMnode *pMnode, SUserObj *pUser) {
SSdb *pSdb = pMnode->pSdb;
sdbRelease(pSdb, pUser);
static int32_t mndProcessAlterUserMsg(SMnodeMsg *pMsg) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process alter user msg since %s", terrstr());
return -1;
}
static int32_t mndProcessDropUserMsg(SMnodeMsg *pMsg) {
terrno = TSDB_CODE_MND_MSG_NOT_PROCESSED;
mError("failed to process drop user msg since %s", terrstr());
return -1;
}