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

TD-10431 test for profile

This commit is contained in:
Shengliang Guan 2021-12-21 22:35:45 +08:00
parent 26a3ff9b0f
commit 40122d3ca6
9 changed files with 397 additions and 915 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
include/common/taosmsg.h
View File

@ -79,7 +79,6 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB, "alter-stb" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB, "drop-stb" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB, "drop-stb" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_VGROUP_LIST, "vgroup-list" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_VGROUP_LIST, "vgroup-list" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_STREAM, "kill-stream" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW, "show" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW, "show" )
@ -946,18 +945,6 @@ typedef struct {
char subSqlInfo[TSDB_SHOW_SUBQUERY_LEN]; // include subqueries' index, Obj IDs and states(C-complete/I-imcomplete) char subSqlInfo[TSDB_SHOW_SUBQUERY_LEN]; // include subqueries' index, Obj IDs and states(C-complete/I-imcomplete)
} SQueryDesc; } SQueryDesc;
typedef struct {
char sql[TSDB_SHOW_SQL_LEN];
char dstTable[TSDB_TABLE_NAME_LEN];
int32_t streamId;
int64_t num; // number of computing/cycles
int64_t useconds;
int64_t ctime;
int64_t stime;
int64_t slidingTime;
int64_t interval;
} SStreamDesc;
typedef struct { typedef struct {
int32_t connId; int32_t connId;
int32_t pid; int32_t pid;

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

@ -73,7 +73,6 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TSDB_MSG_TYPE_DROP_STB] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_DROP_STB] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_VGROUP_LIST] = dndProcessMnodeReadMsg; pMgmt->msgFp[TSDB_MSG_TYPE_VGROUP_LIST] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_KILL_STREAM] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_KILL_CONN] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_KILL_CONN] = dndProcessMnodeWriteMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dndProcessMnodeReadMsg; pMgmt->msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dndProcessMnodeReadMsg;
pMgmt->msgFp[TSDB_MSG_TYPE_SHOW] = dndProcessMnodeReadMsg; pMgmt->msgFp[TSDB_MSG_TYPE_SHOW] = dndProcessMnodeReadMsg;

6
source/dnode/mgmt/impl/test/CMakeLists.txt
View File

@ -8,13 +8,11 @@ add_subdirectory(db)
add_subdirectory(dnode) add_subdirectory(dnode)
# add_subdirectory(func) # add_subdirectory(func)
# add_subdirectory(mnode) # add_subdirectory(mnode)
# add_subdirectory(profile) add_subdirectory(profile)
# add_subdirectory(show) add_subdirectory(show)
add_subdirectory(stb) add_subdirectory(stb)
# add_subdirectory(sync) # add_subdirectory(sync)
# add_subdirectory(telem) # add_subdirectory(telem)
# add_subdirectory(trans) # add_subdirectory(trans)
add_subdirectory(user) add_subdirectory(user)
add_subdirectory(vgroup) add_subdirectory(vgroup)
# add_subdirectory(common)

12
source/dnode/mgmt/impl/test/profile/CMakeLists.txt
View File

@ -1,20 +1,20 @@
add_executable(dndTestProfile "") add_executable(dnode_test_profile "")
target_sources(dndTestProfile target_sources(dnode_test_profile
PRIVATE PRIVATE
"profile.cpp" "profile.cpp"
"../sut/deploy.cpp" "../sut/deploy.cpp"
) )
target_link_libraries( target_link_libraries(
dndTestProfile dnode_test_profile
PUBLIC dnode PUBLIC dnode
PUBLIC util PUBLIC util
PUBLIC os PUBLIC os
PUBLIC gtest_main PUBLIC gtest_main
) )
target_include_directories(dndTestProfile target_include_directories(dnode_test_profile
PUBLIC PUBLIC
"${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt" "${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
"${CMAKE_CURRENT_SOURCE_DIR}/../../inc" "${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
@ -22,6 +22,6 @@ target_include_directories(dndTestProfile
) )
add_test( add_test(
NAME dndTestProfile NAME dnode_test_profile
COMMAND dndTestProfile COMMAND dnode_test_profile
) )

637
source/dnode/mgmt/impl/test/profile/profile.cpp
View File

@ -13,41 +13,174 @@
class DndTestProfile : public ::testing::Test { class DndTestProfile : public ::testing::Test {
protected: protected:
void SetUp() override {} static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
void TearDown() override {} SServer* pServer = createServer(path, fqdn, port, firstEp);
ASSERT(pServer);
return pServer;
}
static void SetUpTestSuite() { static void SetUpTestSuite() {
const char* user = "root"; initLog("/tmp/tdlog");
const char* pass = "taosdata";
const char* path = "/tmp/dndTestProfile";
const char* fqdn = "localhost";
uint16_t port = 9522;
pServer = createServer(path, fqdn, port); const char* fqdn = "localhost";
ASSERT(pServer); const char* firstEp = "localhost:9080";
pClient = createClient(user, pass, fqdn, port); pServer = CreateServer("/tmp/dnode_test_profile", fqdn, 9080, firstEp);
pClient = createClient("root", "taosdata", fqdn, 9080);
taosMsleep(300);
} }
static void TearDownTestSuite() { static void TearDownTestSuite() {
stopServer(pServer); stopServer(pServer);
dropClient(pClient); dropClient(pClient);
pServer = NULL;
pClient = NULL;
} }
static SServer* pServer; static SServer* pServer;
static SClient* pClient; static SClient* pClient;
static int32_t connId; static int32_t connId;
public:
void SetUp() override {}
void TearDown() override {}
void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns) {
SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pShow->type = showType;
strcpy(pShow->db, "");
SRpcMsg showRpcMsg = {0};
showRpcMsg.pCont = pShow;
showRpcMsg.contLen = sizeof(SShowMsg);
showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &showRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
ASSERT_NE(pShowRsp, nullptr);
pShowRsp->showId = htonl(pShowRsp->showId);
pMeta = &pShowRsp->tableMeta;
pMeta->numOfTags = htonl(pMeta->numOfTags);
pMeta->numOfColumns = htonl(pMeta->numOfColumns);
pMeta->sversion = htonl(pMeta->sversion);
pMeta->tversion = htonl(pMeta->tversion);
pMeta->tuid = htobe64(pMeta->tuid);
pMeta->suid = htobe64(pMeta->suid);
showId = pShowRsp->showId;
EXPECT_NE(pShowRsp->showId, 0);
EXPECT_STREQ(pMeta->tbFname, showName);
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->numOfColumns, columns);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->update, 0);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tuid, 0);
EXPECT_EQ(pMeta->suid, 0);
}
void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
SSchema* pSchema = &pMeta->pSchema[index];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, type);
EXPECT_EQ(pSchema->bytes, bytes);
EXPECT_STREQ(pSchema->name, name);
}
void SendThenCheckShowRetrieveMsg(int32_t rows) {
SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pRetrieve->showId = htonl(showId);
pRetrieve->free = 0;
SRpcMsg retrieveRpcMsg = {0};
retrieveRpcMsg.pCont = pRetrieve;
retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &retrieveRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
ASSERT_NE(pRetrieveRsp, nullptr);
pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
EXPECT_EQ(pRetrieveRsp->useconds, 0);
// EXPECT_EQ(pRetrieveRsp->completed, completed);
EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRetrieveRsp->compressed, 0);
EXPECT_EQ(pRetrieveRsp->compLen, 0);
pData = pRetrieveRsp->data;
pos = 0;
}
void CheckInt16(int16_t val) {
int16_t data = *((int16_t*)(pData + pos));
pos += sizeof(int16_t);
EXPECT_EQ(data, val);
}
void CheckInt32(int32_t val) {
int32_t data = *((int32_t*)(pData + pos));
pos += sizeof(int32_t);
EXPECT_EQ(data, val);
}
void CheckInt64(int64_t val) {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_EQ(data, val);
}
void CheckTimestamp() {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_GT(data, 0);
}
void CheckBinary(const char* val, int32_t len) {
pos += sizeof(VarDataLenT);
char* data = (char*)(pData + pos);
pos += len;
EXPECT_STREQ(data, val);
}
void IgnoreBinary(int32_t len) {
pos += sizeof(VarDataLenT);
char* data = (char*)(pData + pos);
pos += len;
}
int32_t showId;
STableMetaMsg* pMeta;
SRetrieveTableRsp* pRetrieveRsp;
char* pData;
int32_t pos;
}; };
SServer* DndTestProfile::pServer; SServer* DndTestProfile::pServer;
SClient* DndTestProfile::pClient; SClient* DndTestProfile::pClient;
int32_t DndTestProfile::connId; int32_t DndTestProfile::connId;
TEST_F(DndTestProfile, SConnectMsg_01) { TEST_F(DndTestProfile, 01_ConnectMsg) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg)); SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
strcpy(pReq->db, ""); strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
@ -69,24 +202,22 @@ TEST_F(DndTestProfile, SConnectMsg_01) {
EXPECT_EQ(pRsp->acctId, 1); EXPECT_EQ(pRsp->acctId, 1);
EXPECT_GT(pRsp->clusterId, 0); EXPECT_GT(pRsp->clusterId, 0);
EXPECT_EQ(pRsp->connId, 1); EXPECT_EQ(pRsp->connId, 1);
EXPECT_EQ(pRsp->superAuth, 1); EXPECT_EQ(pRsp->superUser, 1);
EXPECT_EQ(pRsp->readAuth, 1);
EXPECT_EQ(pRsp->writeAuth, 1);
EXPECT_EQ(pRsp->epSet.inUse, 0); EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1); EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9522); EXPECT_EQ(pRsp->epSet.port[0], 9080);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost"); EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
connId = pRsp->connId; connId = pRsp->connId;
} }
TEST_F(DndTestProfile, SConnectMsg_02) { TEST_F(DndTestProfile, 02_ConnectMsg_InvalidDB) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg)); SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
strcpy(pReq->db, "invalid_db"); strcpy(pReq->db, "invalid_db");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
@ -101,134 +232,28 @@ TEST_F(DndTestProfile, SConnectMsg_02) {
ASSERT_EQ(pMsg->contLen, 0); ASSERT_EQ(pMsg->contLen, 0);
} }
TEST_F(DndTestProfile, SConnectMsg_03) { TEST_F(DndTestProfile, 03_ConnectMsg_Show) {
ASSERT_NE(pClient, nullptr); SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_CONNS, "show connections", 7);
int32_t showId = 0; CheckSchema(0, TSDB_DATA_TYPE_INT, 4, "connId");
CheckSchema(1, TSDB_DATA_TYPE_BINARY, TSDB_USER_LEN + VARSTR_HEADER_SIZE, "user");
CheckSchema(2, TSDB_DATA_TYPE_BINARY, TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE, "program");
CheckSchema(3, TSDB_DATA_TYPE_INT, 4, "pid");
CheckSchema(4, TSDB_DATA_TYPE_BINARY, TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE, "ip:port");
CheckSchema(5, TSDB_DATA_TYPE_TIMESTAMP, 8, "login_time");
CheckSchema(6, TSDB_DATA_TYPE_TIMESTAMP, 8, "last_access");
{ SendThenCheckShowRetrieveMsg(1);
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pReq->type = TSDB_MGMT_TABLE_CONNS;
strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0}; CheckInt32(1);
rpcMsg.pCont = pReq; CheckBinary("root", TSDB_USER_LEN);
rpcMsg.contLen = sizeof(SShowMsg); CheckBinary("dnode_test_profile", TSDB_APP_NAME_LEN);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW; CheckInt32(1234);
IgnoreBinary(TSDB_IPv4ADDR_LEN + 6);
sendMsg(pClient, &rpcMsg); CheckTimestamp();
SRpcMsg* pMsg = pClient->pRsp; CheckTimestamp();
ASSERT_NE(pMsg, nullptr);
SShowRsp* pRsp = (SShowRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->showId = htonl(pRsp->showId);
STableMetaMsg* pMeta = &pRsp->tableMeta;
pMeta->contLen = htonl(pMeta->contLen);
pMeta->numOfColumns = htons(pMeta->numOfColumns);
pMeta->sversion = htons(pMeta->sversion);
pMeta->tversion = htons(pMeta->tversion);
pMeta->tid = htonl(pMeta->tid);
pMeta->uid = htobe64(pMeta->uid);
pMeta->suid = htobe64(pMeta->suid);
showId = pRsp->showId;
EXPECT_NE(pRsp->showId, 0);
EXPECT_EQ(pMeta->contLen, 0);
EXPECT_STREQ(pMeta->tbFname, "");
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->numOfColumns, 7);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tid, 0);
EXPECT_EQ(pMeta->uid, 0);
EXPECT_STREQ(pMeta->sTableName, "");
EXPECT_EQ(pMeta->suid, 0);
SSchema* pSchema = NULL;
pSchema = &pMeta->pSchema[0];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "connId");
pSchema = &pMeta->pSchema[1];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "user");
pSchema = &pMeta->pSchema[2];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "program");
pSchema = &pMeta->pSchema[3];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "pid");
pSchema = &pMeta->pSchema[4];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "ip:port");
pSchema = &pMeta->pSchema[5];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TIMESTAMP);
EXPECT_EQ(pSchema->bytes, 8);
EXPECT_STREQ(pSchema->name, "login_time");
pSchema = &pMeta->pSchema[6];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TIMESTAMP);
EXPECT_EQ(pSchema->bytes, 8);
EXPECT_STREQ(pSchema->name, "last_access");
} }
{ TEST_F(DndTestProfile, 04_HeartBeatMsg) {
SRetrieveTableMsg* pReq = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pReq->showId = htonl(showId);
pReq->free = 0;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SRetrieveTableMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
SRetrieveTableRsp* pRsp = (SRetrieveTableRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->numOfRows = htonl(pRsp->numOfRows);
pRsp->useconds = htobe64(pRsp->useconds);
pRsp->compLen = htonl(pRsp->compLen);
EXPECT_EQ(pRsp->numOfRows, 1);
EXPECT_EQ(pRsp->useconds, 0);
EXPECT_EQ(pRsp->completed, 1);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRsp->compressed, 0);
EXPECT_EQ(pRsp->compLen, 0);
}
}
TEST_F(DndTestProfile, SHeartBeatMsg_01) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SHeartBeatMsg* pReq = (SHeartBeatMsg*)rpcMallocCont(sizeof(SHeartBeatMsg)); SHeartBeatMsg* pReq = (SHeartBeatMsg*)rpcMallocCont(sizeof(SHeartBeatMsg));
@ -236,7 +261,7 @@ TEST_F(DndTestProfile, SHeartBeatMsg_01) {
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
pReq->numOfQueries = htonl(0); pReq->numOfQueries = htonl(0);
pReq->numOfStreams = htonl(0); pReq->numOfStreams = htonl(0);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq; rpcMsg.pCont = pReq;
@ -265,11 +290,11 @@ TEST_F(DndTestProfile, SHeartBeatMsg_01) {
EXPECT_EQ(pRsp->epSet.inUse, 0); EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1); EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9522); EXPECT_EQ(pRsp->epSet.port[0], 9080);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost"); EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
} }
TEST_F(DndTestProfile, SKillConnMsg_01) { TEST_F(DndTestProfile, 05_KillConnMsg) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
{ {
@ -293,7 +318,7 @@ TEST_F(DndTestProfile, SKillConnMsg_01) {
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
pReq->numOfQueries = htonl(0); pReq->numOfQueries = htonl(0);
pReq->numOfStreams = htonl(0); pReq->numOfStreams = htonl(0);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq; rpcMsg.pCont = pReq;
@ -310,7 +335,7 @@ TEST_F(DndTestProfile, SKillConnMsg_01) {
{ {
SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg)); SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
strcpy(pReq->db, ""); strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
@ -332,19 +357,18 @@ TEST_F(DndTestProfile, SKillConnMsg_01) {
EXPECT_EQ(pRsp->acctId, 1); EXPECT_EQ(pRsp->acctId, 1);
EXPECT_GT(pRsp->clusterId, 0); EXPECT_GT(pRsp->clusterId, 0);
EXPECT_GT(pRsp->connId, connId); EXPECT_GT(pRsp->connId, connId);
EXPECT_EQ(pRsp->readAuth, 1); EXPECT_EQ(pRsp->superUser, 1);
EXPECT_EQ(pRsp->writeAuth, 1);
EXPECT_EQ(pRsp->epSet.inUse, 0); EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1); EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9522); EXPECT_EQ(pRsp->epSet.port[0], 9080);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost"); EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
connId = pRsp->connId; connId = pRsp->connId;
} }
} }
TEST_F(DndTestProfile, SKillConnMsg_02) { TEST_F(DndTestProfile, 06_KillConnMsg_InvalidConn) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SKillConnMsg* pReq = (SKillConnMsg*)rpcMallocCont(sizeof(SKillConnMsg)); SKillConnMsg* pReq = (SKillConnMsg*)rpcMallocCont(sizeof(SKillConnMsg));
@ -361,7 +385,7 @@ TEST_F(DndTestProfile, SKillConnMsg_02) {
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_CONN_ID); ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_CONN_ID);
} }
TEST_F(DndTestProfile, SKillQueryMsg_01) { TEST_F(DndTestProfile, 07_KillQueryMsg) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
{ {
@ -387,7 +411,7 @@ TEST_F(DndTestProfile, SKillQueryMsg_01) {
pReq->pid = htonl(1234); pReq->pid = htonl(1234);
pReq->numOfQueries = htonl(0); pReq->numOfQueries = htonl(0);
pReq->numOfStreams = htonl(0); pReq->numOfStreams = htonl(0);
strcpy(pReq->app, "dndTestProfile"); strcpy(pReq->app, "dnode_test_profile");
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq; rpcMsg.pCont = pReq;
@ -416,12 +440,12 @@ TEST_F(DndTestProfile, SKillQueryMsg_01) {
EXPECT_EQ(pRsp->epSet.inUse, 0); EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1); EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9522); EXPECT_EQ(pRsp->epSet.port[0], 9080);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost"); EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
} }
} }
TEST_F(DndTestProfile, SKillQueryMsg_02) { TEST_F(DndTestProfile, 08_KillQueryMsg_InvalidCOnn) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SKillQueryMsg* pReq = (SKillQueryMsg*)rpcMallocCont(sizeof(SKillQueryMsg)); SKillQueryMsg* pReq = (SKillQueryMsg*)rpcMallocCont(sizeof(SKillQueryMsg));
@ -439,291 +463,22 @@ TEST_F(DndTestProfile, SKillQueryMsg_02) {
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_CONN_ID); ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_CONN_ID);
} }
TEST_F(DndTestProfile, SKillQueryMsg_03) { TEST_F(DndTestProfile, 09_KillQueryMsg) {
ASSERT_NE(pClient, nullptr); SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_QUERIES, "show queries", 14);
int32_t showId = 0; CheckSchema(0, TSDB_DATA_TYPE_INT, 4, "queryId");
CheckSchema(1, TSDB_DATA_TYPE_INT, 4, "connId");
CheckSchema(2, TSDB_DATA_TYPE_BINARY, TSDB_USER_LEN + VARSTR_HEADER_SIZE, "user");
CheckSchema(3, TSDB_DATA_TYPE_BINARY, TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE, "ip:port");
CheckSchema(4, TSDB_DATA_TYPE_BINARY, 22 + VARSTR_HEADER_SIZE, "qid");
CheckSchema(5, TSDB_DATA_TYPE_TIMESTAMP, 8, "created_time");
CheckSchema(6, TSDB_DATA_TYPE_BIGINT, 8, "time");
CheckSchema(7, TSDB_DATA_TYPE_BINARY, 18 + VARSTR_HEADER_SIZE, "sql_obj_id");
CheckSchema(8, TSDB_DATA_TYPE_INT, 4, "pid");
CheckSchema(9, TSDB_DATA_TYPE_BINARY, TSDB_EP_LEN + VARSTR_HEADER_SIZE, "ep");
CheckSchema(10, TSDB_DATA_TYPE_BOOL, 1, "stable_query");
CheckSchema(11, TSDB_DATA_TYPE_INT, 4, "sub_queries");
CheckSchema(12, TSDB_DATA_TYPE_BINARY, TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE, "sub_query_info");
CheckSchema(13, TSDB_DATA_TYPE_BINARY, TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, "sql");
{ SendThenCheckShowRetrieveMsg(0);
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pReq->type = TSDB_MGMT_TABLE_QUERIES;
strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SShowMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
SShowRsp* pRsp = (SShowRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->showId = htonl(pRsp->showId);
STableMetaMsg* pMeta = &pRsp->tableMeta;
pMeta->contLen = htonl(pMeta->contLen);
pMeta->numOfColumns = htons(pMeta->numOfColumns);
pMeta->sversion = htons(pMeta->sversion);
pMeta->tversion = htons(pMeta->tversion);
pMeta->tid = htonl(pMeta->tid);
pMeta->uid = htobe64(pMeta->uid);
pMeta->suid = htobe64(pMeta->suid);
showId = pRsp->showId;
EXPECT_NE(pRsp->showId, 0);
EXPECT_EQ(pMeta->contLen, 0);
EXPECT_STREQ(pMeta->tbFname, "");
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->numOfColumns, 14);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tid, 0);
EXPECT_EQ(pMeta->uid, 0);
EXPECT_STREQ(pMeta->sTableName, "");
EXPECT_EQ(pMeta->suid, 0);
SSchema* pSchema = NULL;
pSchema = &pMeta->pSchema[0];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "queryId");
pSchema = &pMeta->pSchema[1];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "connId");
pSchema = &pMeta->pSchema[2];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "user");
pSchema = &pMeta->pSchema[3];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "ip:port");
}
{
SRetrieveTableMsg* pReq = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pReq->showId = htonl(showId);
pReq->free = 0;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SRetrieveTableMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
SRetrieveTableRsp* pRsp = (SRetrieveTableRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->numOfRows = htonl(pRsp->numOfRows);
pRsp->offset = htobe64(pRsp->offset);
pRsp->useconds = htobe64(pRsp->useconds);
pRsp->compLen = htonl(pRsp->compLen);
EXPECT_EQ(pRsp->numOfRows, 0);
EXPECT_EQ(pRsp->offset, 0);
EXPECT_EQ(pRsp->useconds, 0);
EXPECT_EQ(pRsp->completed, 1);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRsp->compressed, 0);
EXPECT_EQ(pRsp->reserved, 0);
EXPECT_EQ(pRsp->compLen, 0);
}
}
TEST_F(DndTestProfile, SKillStreamMsg_01) {
ASSERT_NE(pClient, nullptr);
{
SKillStreamMsg* pReq = (SKillStreamMsg*)rpcMallocCont(sizeof(SKillStreamMsg));
pReq->connId = htonl(connId);
pReq->streamId = htonl(3579);
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SKillStreamMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_KILL_STREAM;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
ASSERT_EQ(pMsg->contLen, 0);
}
{
SHeartBeatMsg* pReq = (SHeartBeatMsg*)rpcMallocCont(sizeof(SHeartBeatMsg));
pReq->connId = htonl(connId);
pReq->pid = htonl(1234);
pReq->numOfQueries = htonl(0);
pReq->numOfStreams = htonl(0);
strcpy(pReq->app, "dndTestProfile");
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SHeartBeatMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_HEARTBEAT;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
SHeartBeatRsp* pRsp = (SHeartBeatRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->connId = htonl(pRsp->connId);
pRsp->queryId = htonl(pRsp->queryId);
pRsp->streamId = htonl(pRsp->streamId);
pRsp->totalDnodes = htonl(pRsp->totalDnodes);
pRsp->onlineDnodes = htonl(pRsp->onlineDnodes);
pRsp->epSet.port[0] = htons(pRsp->epSet.port[0]);
EXPECT_EQ(pRsp->connId, connId);
EXPECT_EQ(pRsp->queryId, 0);
EXPECT_EQ(pRsp->streamId, 3579);
EXPECT_EQ(pRsp->totalDnodes, 1);
EXPECT_EQ(pRsp->onlineDnodes, 1);
EXPECT_EQ(pRsp->killConnection, 0);
EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9522);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
}
}
TEST_F(DndTestProfile, SKillStreamMsg_02) {
ASSERT_NE(pClient, nullptr);
SKillStreamMsg* pReq = (SKillStreamMsg*)rpcMallocCont(sizeof(SKillStreamMsg));
pReq->connId = htonl(2345);
pReq->streamId = htonl(1234);
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SKillStreamMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_KILL_QUERY;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_CONN_ID);
}
TEST_F(DndTestProfile, SKillStreamMsg_03) {
ASSERT_NE(pClient, nullptr);
int32_t showId = 0;
{
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pReq->type = TSDB_MGMT_TABLE_STREAMS;
strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SShowMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
SShowRsp* pRsp = (SShowRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->showId = htonl(pRsp->showId);
STableMetaMsg* pMeta = &pRsp->tableMeta;
pMeta->contLen = htonl(pMeta->contLen);
pMeta->numOfColumns = htons(pMeta->numOfColumns);
pMeta->sversion = htons(pMeta->sversion);
pMeta->tversion = htons(pMeta->tversion);
pMeta->tid = htonl(pMeta->tid);
pMeta->uid = htobe64(pMeta->uid);
pMeta->suid = htobe64(pMeta->suid);
showId = pRsp->showId;
EXPECT_NE(pRsp->showId, 0);
EXPECT_EQ(pMeta->contLen, 0);
EXPECT_STREQ(pMeta->tbFname, "");
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->numOfColumns, 10);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tid, 0);
EXPECT_EQ(pMeta->uid, 0);
EXPECT_STREQ(pMeta->sTableName, "");
EXPECT_EQ(pMeta->suid, 0);
SSchema* pSchema = NULL;
pSchema = &pMeta->pSchema[0];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "streamId");
pSchema = &pMeta->pSchema[1];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "connId");
pSchema = &pMeta->pSchema[2];
pSchema->bytes = htons(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "user");
}
{
SRetrieveTableMsg* pReq = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pReq->showId = htonl(showId);
pReq->free = 0;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SRetrieveTableMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
SRetrieveTableRsp* pRsp = (SRetrieveTableRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->numOfRows = htonl(pRsp->numOfRows);
pRsp->offset = htobe64(pRsp->offset);
pRsp->useconds = htobe64(pRsp->useconds);
pRsp->compLen = htonl(pRsp->compLen);
EXPECT_EQ(pRsp->numOfRows, 0);
EXPECT_EQ(pRsp->offset, 0);
EXPECT_EQ(pRsp->useconds, 0);
EXPECT_EQ(pRsp->completed, 1);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRsp->compressed, 0);
EXPECT_EQ(pRsp->reserved, 0);
EXPECT_EQ(pRsp->compLen, 0);
}
} }

312
source/dnode/mgmt/impl/test/show/show.cpp
View File

@ -13,61 +13,169 @@
class DndTestShow : public ::testing::Test { class DndTestShow : public ::testing::Test {
protected: protected:
void SetUp() override {} static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
void TearDown() override {} SServer* pServer = createServer(path, fqdn, port, firstEp);
ASSERT(pServer);
return pServer;
}
static void SetUpTestSuite() { static void SetUpTestSuite() {
const char* user = "root"; initLog("/tmp/tdlog");
const char* pass = "taosdata";
const char* path = "/tmp/dndTestShow";
const char* fqdn = "localhost";
uint16_t port = 9523;
pServer = createServer(path, fqdn, port); const char* fqdn = "localhost";
ASSERT(pServer); const char* firstEp = "localhost:9091";
pClient = createClient(user, pass, fqdn, port); pServer = CreateServer("/tmp/dnode_test_show", fqdn, 9091, firstEp);
pClient = createClient("root", "taosdata", fqdn, 9091);
taosMsleep(300);
} }
static void TearDownTestSuite() { static void TearDownTestSuite() {
stopServer(pServer); stopServer(pServer);
dropClient(pClient); dropClient(pClient);
pServer = NULL;
pClient = NULL;
} }
static SServer* pServer; static SServer* pServer;
static SClient* pClient; static SClient* pClient;
static int32_t connId; static int32_t connId;
public:
void SetUp() override {}
void TearDown() override {}
void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns) {
SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pShow->type = showType;
strcpy(pShow->db, "");
SRpcMsg showRpcMsg = {0};
showRpcMsg.pCont = pShow;
showRpcMsg.contLen = sizeof(SShowMsg);
showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &showRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
ASSERT_NE(pShowRsp, nullptr);
pShowRsp->showId = htonl(pShowRsp->showId);
pMeta = &pShowRsp->tableMeta;
pMeta->numOfTags = htonl(pMeta->numOfTags);
pMeta->numOfColumns = htonl(pMeta->numOfColumns);
pMeta->sversion = htonl(pMeta->sversion);
pMeta->tversion = htonl(pMeta->tversion);
pMeta->tuid = htobe64(pMeta->tuid);
pMeta->suid = htobe64(pMeta->suid);
showId = pShowRsp->showId;
EXPECT_NE(pShowRsp->showId, 0);
EXPECT_STREQ(pMeta->tbFname, showName);
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->numOfColumns, columns);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->update, 0);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tuid, 0);
EXPECT_EQ(pMeta->suid, 0);
}
void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
SSchema* pSchema = &pMeta->pSchema[index];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, type);
EXPECT_EQ(pSchema->bytes, bytes);
EXPECT_STREQ(pSchema->name, name);
}
void SendThenCheckShowRetrieveMsg(int32_t rows) {
SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pRetrieve->showId = htonl(showId);
pRetrieve->free = 0;
SRpcMsg retrieveRpcMsg = {0};
retrieveRpcMsg.pCont = pRetrieve;
retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &retrieveRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
ASSERT_NE(pRetrieveRsp, nullptr);
pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
EXPECT_EQ(pRetrieveRsp->useconds, 0);
// EXPECT_EQ(pRetrieveRsp->completed, completed);
EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRetrieveRsp->compressed, 0);
EXPECT_EQ(pRetrieveRsp->compLen, 0);
pData = pRetrieveRsp->data;
pos = 0;
}
void CheckInt16(int16_t val) {
int16_t data = *((int16_t*)(pData + pos));
pos += sizeof(int16_t);
EXPECT_EQ(data, val);
}
void CheckInt32(int32_t val) {
int32_t data = *((int32_t*)(pData + pos));
pos += sizeof(int32_t);
EXPECT_EQ(data, val);
}
void CheckInt64(int64_t val) {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_EQ(data, val);
}
void CheckTimestamp() {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_GT(data, 0);
}
void CheckBinary(const char* val, int32_t len) {
pos += sizeof(VarDataLenT);
char* data = (char*)(pData + pos);
pos += len;
EXPECT_STREQ(data, val);
}
void IgnoreBinary(int32_t len) {
pos += sizeof(VarDataLenT);
char* data = (char*)(pData + pos);
pos += len;
}
int32_t showId;
STableMetaMsg* pMeta;
SRetrieveTableRsp* pRetrieveRsp;
char* pData;
int32_t pos;
}; };
SServer* DndTestShow::pServer; SServer* DndTestShow::pServer;
SClient* DndTestShow::pClient; SClient* DndTestShow::pClient;
int32_t DndTestShow::connId; int32_t DndTestShow::connId;
TEST_F(DndTestShow, SShowMsg_01) { TEST_F(DndTestShow, 01_ShowMsg_InvalidMsgMax) {
ASSERT_NE(pClient, nullptr);
SConnectMsg* pReq = (SConnectMsg*)rpcMallocCont(sizeof(SConnectMsg));
pReq->pid = htonl(1234);
strcpy(pReq->app, "dndTestShow");
strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SConnectMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_CONNECT;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
SConnectRsp* pRsp = (SConnectRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->connId = htonl(pRsp->connId);
EXPECT_EQ(pRsp->connId, 1);
connId = pRsp->connId;
}
TEST_F(DndTestShow, SShowMsg_02) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg)); SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
@ -85,7 +193,7 @@ TEST_F(DndTestShow, SShowMsg_02) {
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_MSG_TYPE); ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_MSG_TYPE);
} }
TEST_F(DndTestShow, SShowMsg_03) { TEST_F(DndTestShow, 02_ShowMsg_InvalidMsgStart) {
ASSERT_NE(pClient, nullptr); ASSERT_NE(pClient, nullptr);
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg)); SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
@ -102,133 +210,3 @@ TEST_F(DndTestShow, SShowMsg_03) {
ASSERT_NE(pMsg, nullptr); ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_MSG_TYPE); ASSERT_EQ(pMsg->code, TSDB_CODE_MND_INVALID_MSG_TYPE);
} }
TEST_F(DndTestShow, SShowMsg_04) {
ASSERT_NE(pClient, nullptr);
int32_t showId = 0;
{
SShowMsg* pReq = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pReq->type = TSDB_MGMT_TABLE_CONNS;
strcpy(pReq->db, "");
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SShowMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
SShowRsp* pRsp = (SShowRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->showId = htonl(pRsp->showId);
STableMetaMsg* pMeta = &pRsp->tableMeta;
pMeta->contLen = htonl(pMeta->contLen);
pMeta->numOfColumns = htons(pMeta->numOfColumns);
pMeta->sversion = htons(pMeta->sversion);
pMeta->tversion = htons(pMeta->tversion);
pMeta->tid = htonl(pMeta->tid);
pMeta->uid = htobe64(pMeta->uid);
pMeta->suid = htobe64(pMeta->suid);
showId = pRsp->showId;
EXPECT_NE(pRsp->showId, 0);
EXPECT_EQ(pMeta->contLen, 0);
EXPECT_STREQ(pMeta->tbFname, "");
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->numOfColumns, 7);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tid, 0);
EXPECT_EQ(pMeta->uid, 0);
EXPECT_STREQ(pMeta->sTableName, "");
EXPECT_EQ(pMeta->suid, 0);
SSchema* pSchema = NULL;
pSchema = &pMeta->pSchema[0];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "connId");
pSchema = &pMeta->pSchema[1];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "user");
pSchema = &pMeta->pSchema[2];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_USER_LEN + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "program");
pSchema = &pMeta->pSchema[3];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_INT);
EXPECT_EQ(pSchema->bytes, 4);
EXPECT_STREQ(pSchema->name, "pid");
pSchema = &pMeta->pSchema[4];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->bytes, TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE);
EXPECT_STREQ(pSchema->name, "ip:port");
pSchema = &pMeta->pSchema[5];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TIMESTAMP);
EXPECT_EQ(pSchema->bytes, 8);
EXPECT_STREQ(pSchema->name, "login_time");
pSchema = &pMeta->pSchema[6];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TIMESTAMP);
EXPECT_EQ(pSchema->bytes, 8);
EXPECT_STREQ(pSchema->name, "last_access");
}
{
SRetrieveTableMsg* pReq = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pReq->showId = htonl(showId);
pReq->free = 0;
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SRetrieveTableMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
SRetrieveTableRsp* pRsp = (SRetrieveTableRsp*)pMsg->pCont;
ASSERT_NE(pRsp, nullptr);
pRsp->numOfRows = htonl(pRsp->numOfRows);
pRsp->offset = htobe64(pRsp->offset);
pRsp->useconds = htobe64(pRsp->useconds);
pRsp->compLen = htonl(pRsp->compLen);
EXPECT_EQ(pRsp->numOfRows, 1);
EXPECT_EQ(pRsp->offset, 0);
EXPECT_EQ(pRsp->useconds, 0);
EXPECT_EQ(pRsp->completed, 1);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRsp->compressed, 0);
EXPECT_EQ(pRsp->reserved, 0);
EXPECT_EQ(pRsp->compLen, 0);
}
}

299
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -23,29 +23,25 @@
#define QUERY_ID_SIZE 20 #define QUERY_ID_SIZE 20
#define QUERY_OBJ_ID_SIZE 18 #define QUERY_OBJ_ID_SIZE 18
#define SUBQUERY_INFO_SIZE 6 #define SUBQUERY_INFO_SIZE 6
#define QUERY_STREAM_SAVE_SIZE 20 #define QUERY_SAVE_SIZE 20
typedef struct { typedef struct {
int32_t id;
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
char app[TSDB_APP_NAME_LEN]; // app name that invokes taosc char app[TSDB_APP_NAME_LEN]; // app name that invokes taosc
int64_t appStartTimeMs; // app start time
int32_t pid; // pid of app that invokes taosc int32_t pid; // pid of app that invokes taosc
int64_t appStartTime; // app start time
int32_t id;
int8_t killed;
int8_t align;
uint16_t port;
uint32_t ip; uint32_t ip;
int64_t stime; uint16_t port;
int64_t lastAccess; int8_t killed;
int64_t loginTimeMs;
int64_t lastAccessTimeMs;
int32_t queryId; int32_t queryId;
int32_t streamId;
int32_t numOfQueries; int32_t numOfQueries;
int32_t numOfStreams;
SStreamDesc *pStreams;
SQueryDesc *pQueries; SQueryDesc *pQueries;
} SConnObj; } SConnObj;
static SConnObj *mndCreateConn(SMnode *pMnode, char *user, uint32_t ip, uint16_t port, int32_t pid, const char *app, int64_t startTime); static SConnObj *mndCreateConn(SMnode *pMnode, SRpcConnInfo *pInfo, int32_t pid, const char *app, int64_t startTime);
static void mndFreeConn(SConnObj *pConn); static void mndFreeConn(SConnObj *pConn);
static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId); static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId);
static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn); static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn);
@ -54,16 +50,12 @@ static void mndCancelGetNextConn(SMnode *pMnode, void *pIter);
static int32_t mndProcessHeartBeatMsg(SMnodeMsg *pMsg); static int32_t mndProcessHeartBeatMsg(SMnodeMsg *pMsg);
static int32_t mndProcessConnectMsg(SMnodeMsg *pMsg); static int32_t mndProcessConnectMsg(SMnodeMsg *pMsg);
static int32_t mndProcessKillQueryMsg(SMnodeMsg *pMsg); static int32_t mndProcessKillQueryMsg(SMnodeMsg *pMsg);
static int32_t mndProcessKillStreamMsg(SMnodeMsg *pMsg);
static int32_t mndProcessKillConnectionMsg(SMnodeMsg *pMsg); static int32_t mndProcessKillConnectionMsg(SMnodeMsg *pMsg);
static int32_t mndGetConnsMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta); static int32_t mndGetConnsMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
static int32_t mndRetrieveConns(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveConns(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetQueryMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta); static int32_t mndGetQueryMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
static int32_t mndRetrieveQueries(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveQueries(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter); static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter);
static int32_t mndGetStreamMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta);
static int32_t mndRetrieveStreams(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextStream(SMnode *pMnode, void *pIter);
int32_t mndInitProfile(SMnode *pMnode) { int32_t mndInitProfile(SMnode *pMnode) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt; SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@ -79,7 +71,6 @@ int32_t mndInitProfile(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_HEARTBEAT, mndProcessHeartBeatMsg); mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_HEARTBEAT, mndProcessHeartBeatMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CONNECT, mndProcessConnectMsg); mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_CONNECT, mndProcessConnectMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_QUERY, mndProcessKillQueryMsg); mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_QUERY, mndProcessKillQueryMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_STREAM, mndProcessKillStreamMsg);
mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_CONN, mndProcessKillConnectionMsg); mndSetMsgHandle(pMnode, TSDB_MSG_TYPE_KILL_CONN, mndProcessKillConnectionMsg);
mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_CONNS, mndGetConnsMeta); mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_CONNS, mndGetConnsMeta);
@ -88,9 +79,6 @@ int32_t mndInitProfile(SMnode *pMnode) {
mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndGetQueryMeta); mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndGetQueryMeta);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndRetrieveQueries); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndRetrieveQueries);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndCancelGetNextQuery); mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_QUERIES, mndCancelGetNextQuery);
mndAddShowMetaHandle(pMnode, TSDB_MGMT_TABLE_STREAMS, mndGetStreamMeta);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STREAMS, mndRetrieveStreams);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STREAMS, mndCancelGetNextStream);
return 0; return 0;
} }
@ -103,46 +91,43 @@ void mndCleanupProfile(SMnode *pMnode) {
} }
} }
static SConnObj *mndCreateConn(SMnode *pMnode, char *user, uint32_t ip, uint16_t port, int32_t pid, const char *app, int64_t startTime) { static SConnObj *mndCreateConn(SMnode *pMnode, SRpcConnInfo *pInfo, int32_t pid, const char *app, int64_t startTime) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt; SProfileMgmt *pMgmt = &pMnode->profileMgmt;
int32_t connId = atomic_add_fetch_32(&pMgmt->connId, 1); int32_t connId = atomic_add_fetch_32(&pMgmt->connId, 1);
if (connId == 0) atomic_add_fetch_32(&pMgmt->connId, 1); if (connId == 0) atomic_add_fetch_32(&pMgmt->connId, 1);
if (startTime == 0) startTime = taosGetTimestampMs();
SConnObj connObj = {.pid = pid, SConnObj connObj = {.id = connId,
.appStartTime = startTime, .appStartTimeMs = startTime,
.id = connId, .pid = pid,
.ip = pInfo->clientIp,
.port = pInfo->clientPort,
.killed = 0, .killed = 0,
.port = port, .loginTimeMs = taosGetTimestampMs(),
.ip = ip, .lastAccessTimeMs = 0,
.stime = taosGetTimestampMs(),
.lastAccess = 0,
.queryId = 0, .queryId = 0,
.streamId = 0,
.numOfQueries = 0, .numOfQueries = 0,
.numOfStreams = 0,
.pStreams = NULL,
.pQueries = NULL}; .pQueries = NULL};
connObj.lastAccess = connObj.stime; connObj.lastAccessTimeMs = connObj.loginTimeMs;
tstrncpy(connObj.user, user, TSDB_USER_LEN); tstrncpy(connObj.user, pInfo->user, TSDB_USER_LEN);
tstrncpy(connObj.app, app, TSDB_APP_NAME_LEN); tstrncpy(connObj.app, app, TSDB_APP_NAME_LEN);
int32_t keepTime = pMnode->cfg.shellActivityTimer * 3; int32_t keepTime = pMnode->cfg.shellActivityTimer * 3;
SConnObj *pConn = taosCachePut(pMgmt->cache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), keepTime * 1000); SConnObj *pConn = taosCachePut(pMgmt->cache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), keepTime * 1000);
if (pConn == NULL) { if (pConn == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("conn:%d, data:%p failed to put into cache since %s, user:%s", connId, pConn, user, terrstr()); mError("conn:%d, data:%p failed to put into cache since %s, user:%s", connId, pConn, pInfo->user, terrstr());
return NULL; return NULL;
} else { } else {
mTrace("conn:%d, data:%p created, user:%s", pConn->id, pConn, user); mTrace("conn:%d, data:%p created, user:%s", pConn->id, pConn, pInfo->user);
return pConn; return pConn;
} }
} }
static void mndFreeConn(SConnObj *pConn) { static void mndFreeConn(SConnObj *pConn) {
tfree(pConn->pQueries); tfree(pConn->pQueries);
tfree(pConn->pStreams);
mTrace("conn:%d, data:%p destroyed", pConn->id, pConn); mTrace("conn:%d, data:%p destroyed", pConn->id, pConn);
} }
@ -156,7 +141,7 @@ static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId) {
} }
int32_t keepTime = pMnode->cfg.shellActivityTimer * 3; int32_t keepTime = pMnode->cfg.shellActivityTimer * 3;
pConn->lastAccess = keepTime * 1000 + (uint64_t)taosGetTimestampMs(); pConn->lastAccessTimeMs = keepTime * 1000 + (uint64_t)taosGetTimestampMs();
mTrace("conn:%d, data:%p acquired from cache", pConn->id, pConn); mTrace("conn:%d, data:%p acquired from cache", pConn->id, pConn);
return pConn; return pConn;
@ -219,7 +204,7 @@ static int32_t mndProcessConnectMsg(SMnodeMsg *pMsg) {
mndReleaseDb(pMnode, pDb); mndReleaseDb(pMnode, pDb);
} }
SConnObj *pConn = mndCreateConn(pMnode, info.user, info.clientIp, info.clientPort, pReq->pid, pReq->app, pReq->startTime); SConnObj *pConn = mndCreateConn(pMnode, &info, pReq->pid, pReq->app, pReq->startTime);
if (pConn == NULL) { if (pConn == NULL) {
mError("user:%s, failed to login from %s while create connection since %s", pMsg->user, ip, terrstr()); mError("user:%s, failed to login from %s while create connection since %s", pMsg->user, ip, terrstr());
return -1; return -1;
@ -254,16 +239,14 @@ static int32_t mndProcessConnectMsg(SMnodeMsg *pMsg) {
static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pMsg) { static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pMsg) {
pConn->numOfQueries = 0; pConn->numOfQueries = 0;
pConn->numOfStreams = 0;
int32_t numOfQueries = htonl(pMsg->numOfQueries); int32_t numOfQueries = htonl(pMsg->numOfQueries);
int32_t numOfStreams = htonl(pMsg->numOfStreams);
if (numOfQueries > 0) { if (numOfQueries > 0) {
if (pConn->pQueries == NULL) { if (pConn->pQueries == NULL) {
pConn->pQueries = calloc(sizeof(SQueryDesc), QUERY_STREAM_SAVE_SIZE); pConn->pQueries = calloc(sizeof(SQueryDesc), QUERY_SAVE_SIZE);
} }
pConn->numOfQueries = MIN(QUERY_STREAM_SAVE_SIZE, numOfQueries); pConn->numOfQueries = MIN(QUERY_SAVE_SIZE, numOfQueries);
int32_t saveSize = pConn->numOfQueries * sizeof(SQueryDesc); int32_t saveSize = pConn->numOfQueries * sizeof(SQueryDesc);
if (saveSize > 0 && pConn->pQueries != NULL) { if (saveSize > 0 && pConn->pQueries != NULL) {
@ -271,19 +254,6 @@ static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pMsg) {
} }
} }
if (numOfStreams > 0) {
if (pConn->pStreams == NULL) {
pConn->pStreams = calloc(sizeof(SStreamDesc), QUERY_STREAM_SAVE_SIZE);
}
pConn->numOfStreams = MIN(QUERY_STREAM_SAVE_SIZE, numOfStreams);
int32_t saveSize = pConn->numOfStreams * sizeof(SStreamDesc);
if (saveSize > 0 && pConn->pStreams != NULL) {
memcpy(pConn->pStreams, pMsg->pData + numOfQueries * sizeof(SQueryDesc), saveSize);
}
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -303,7 +273,7 @@ static int32_t mndProcessHeartBeatMsg(SMnodeMsg *pMsg) {
SConnObj *pConn = mndAcquireConn(pMnode, pReq->connId); SConnObj *pConn = mndAcquireConn(pMnode, pReq->connId);
if (pConn == NULL) { if (pConn == NULL) {
pConn = mndCreateConn(pMnode, info.user, info.clientIp, info.clientPort, pReq->pid, pReq->app, 0); pConn = mndCreateConn(pMnode, &info, pReq->pid, pReq->app, 0);
if (pConn == NULL) { if (pConn == NULL) {
mError("user:%s, conn:%d is freed and failed to create new conn since %s", pMsg->user, pReq->connId, terrstr()); mError("user:%s, conn:%d is freed and failed to create new conn since %s", pMsg->user, pReq->connId, terrstr());
return -1; return -1;
@ -343,11 +313,6 @@ static int32_t mndProcessHeartBeatMsg(SMnodeMsg *pMsg) {
pRsp->killConnection = 1; pRsp->killConnection = 1;
} }
if (pConn->streamId != 0) {
pRsp->streamId = htonl(pConn->streamId);
pConn->streamId = 0;
}
if (pConn->queryId != 0) { if (pConn->queryId != 0) {
pRsp->queryId = htonl(pConn->queryId); pRsp->queryId = htonl(pConn->queryId);
pConn->queryId = 0; pConn->queryId = 0;
@ -395,37 +360,6 @@ static int32_t mndProcessKillQueryMsg(SMnodeMsg *pMsg) {
} }
} }
static int32_t mndProcessKillStreamMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pMsg->user);
if (pUser == NULL) return 0;
if (!pUser->superUser) {
mndReleaseUser(pMnode, pUser);
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
mndReleaseUser(pMnode, pUser);
SKillStreamMsg *pKill = pMsg->rpcMsg.pCont;
int32_t connId = htonl(pKill->connId);
int32_t streamId = htonl(pKill->streamId);
mDebug("kill stream msg is received, streamId:%d", streamId);
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &connId, sizeof(int32_t));
if (pConn == NULL) {
mError("connId:%d, failed to kill streamId:%d, conn not exist", connId, streamId);
terrno = TSDB_CODE_MND_INVALID_CONN_ID;
return -1;
} else {
mInfo("connId:%d, streamId:%d is killed by user:%s", connId, streamId, pMsg->user);
pConn->streamId = streamId;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
return TSDB_CODE_SUCCESS;
}
}
static int32_t mndProcessKillConnectionMsg(SMnodeMsg *pMsg) { static int32_t mndProcessKillConnectionMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode; SMnode *pMnode = pMsg->pMnode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt; SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@ -525,6 +459,7 @@ static int32_t mndGetConnsMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *
pShow->numOfRows = taosHashGetSize(pMgmt->cache->pHashTable); pShow->numOfRows = taosHashGetSize(pMgmt->cache->pHashTable);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1]; pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbFname, mndShowStr(pShow->type));
return 0; return 0;
} }
@ -567,12 +502,12 @@ static int32_t mndRetrieveConns(SMnodeMsg *pMsg, SShowObj *pShow, char *data, in
cols++; cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = pConn->stime; *(int64_t *)pWrite = pConn->loginTimeMs;
cols++; cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
if (pConn->lastAccess < pConn->stime) pConn->lastAccess = pConn->stime; if (pConn->lastAccessTimeMs < pConn->loginTimeMs) pConn->lastAccessTimeMs = pConn->loginTimeMs;
*(int64_t *)pWrite = pConn->lastAccess; *(int64_t *)pWrite = pConn->lastAccessTimeMs;
cols++; cols++;
numOfRows++; numOfRows++;
@ -623,7 +558,7 @@ static int32_t mndGetQueryMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *
pSchema[cols].bytes = htonl(pShow->bytes[cols]); pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++; cols++;
pShow->bytes[cols] = 24; pShow->bytes[cols] = 22 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY; pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "qid"); strcpy(pSchema[cols].name, "qid");
pSchema[cols].bytes = htonl(pShow->bytes[cols]); pSchema[cols].bytes = htonl(pShow->bytes[cols]);
@ -693,6 +628,7 @@ static int32_t mndGetQueryMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *
pShow->numOfRows = 1000000; pShow->numOfRows = 1000000;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1]; pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbFname, mndShowStr(pShow->type));
return 0; return 0;
} }
@ -790,6 +726,7 @@ static int32_t mndRetrieveQueries(SMnodeMsg *pMsg, SShowObj *pShow, char *data,
} }
} }
mndVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
return numOfRows; return numOfRows;
} }
@ -798,173 +735,3 @@ static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt; SProfileMgmt *pMgmt = &pMnode->profileMgmt;
taosHashCancelIterate(pMgmt->cache->pHashTable, pIter); taosHashCancelIterate(pMgmt->cache->pHashTable, pIter);
} }
static int32_t mndGetStreamMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta) {
SMnode *pMnode = pMsg->pMnode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pMsg->user);
if (pUser == NULL) return 0;
if (!pUser->superUser) {
mndReleaseUser(pMnode, pUser);
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
mndReleaseUser(pMnode, pUser);
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "streamId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "connId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "user");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "destination");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "exec");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
strcpy(pSchema[cols].name, "time(us)");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "cycles");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
cols++;
pMeta->numOfColumns = htonl(cols);
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = 1000000;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
return 0;
}
static int32_t mndRetrieveStreams(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pMsg->pMnode;
int32_t numOfRows = 0;
SConnObj *pConn = NULL;
int32_t cols = 0;
char *pWrite;
void *pIter;
char ipStr[TSDB_IPv4ADDR_LEN + 6];
while (numOfRows < rows) {
pIter = mndGetNextConn(pMnode, pShow->pIter, &pConn);
if (pConn == NULL) {
pShow->pIter = pIter;
break;
}
if (numOfRows + pConn->numOfStreams >= rows) {
mndCancelGetNextConn(pMnode, pIter);
break;
}
pShow->pIter = pIter;
for (int32_t i = 0; i < pConn->numOfStreams; ++i) {
SStreamDesc *pDesc = pConn->pStreams + i;
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = htobe64(pDesc->streamId);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = htobe64(pConn->id);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pConn->user, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pDesc->dstTable, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
snprintf(ipStr, sizeof(ipStr), "%s:%u", taosIpStr(pConn->ip), pConn->port);
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, ipStr, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = htobe64(pDesc->ctime);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = htobe64(pDesc->stime);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = htobe64(pDesc->useconds);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pDesc->sql, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t *)pWrite = (int32_t)htobe64(pDesc->num);
cols++;
numOfRows++;
}
}
pShow->numOfReads += numOfRows;
return numOfRows;
}
static void mndCancelGetNextStream(SMnode *pMnode, void *pIter) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
taosHashCancelIterate(pMgmt->cache->pHashTable, pIter);
}

4
tests/script/general/user/basic1.sim
View File

@ -26,7 +26,7 @@ if $rows != 2 then
endi endi
print $data00 $data01 $data02 print $data00 $data01 $data02
print $data10 $data11 $data22 print $data10 $data11 $data12
print $data20 $data11 $data22 print $data20 $data11 $data22
print $data30 $data31 $data32 print $data30 $data31 $data32
@ -38,7 +38,7 @@ if $rows != 3 then
endi endi
print $data00 $data01 $data02 print $data00 $data01 $data02
print $data10 $data11 $data22 print $data10 $data11 $data12
print $data20 $data11 $data22 print $data20 $data11 $data22
print $data30 $data31 $data32 print $data30 $data31 $data32
print $data40 $data41 $data42 print $data40 $data41 $data42

12
tests/script/general/user/pass_len.sim
View File

@ -3,8 +3,6 @@ system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1 system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c wallevel -v 0 system sh/cfg.sh -n dnode1 -c wallevel -v 0
system sh/exec.sh -n dnode1 -s start system sh/exec.sh -n dnode1 -s start
sleep 2000
sql connect sql connect
$i = 0 $i = 0
@ -29,7 +27,7 @@ step12:
sql create user $user PASS 'taosdata' sql create user $user PASS 'taosdata'
sql show users sql show users
if $rows != 4 then if $rows != 2 then
return -1 return -1
endi endi
@ -40,7 +38,7 @@ sql drop user $user -x step2
step2: step2:
sql create user $user PASS '1' sql create user $user PASS '1'
sql show users sql show users
if $rows != 5 then if $rows != 3 then
return -1 return -1
endi endi
@ -52,7 +50,7 @@ step3:
sql create user $user PASS 'abc0123456789' sql create user $user PASS 'abc0123456789'
sql show users sql show users
if $rows != 6 then if $rows != 3 then
return -1 return -1
endi endi
@ -63,7 +61,7 @@ sql create user $user PASS 'abcd012345678901234567891234567890' -x step4
return -1 return -1
step4: step4:
sql show users sql show users
if $rows != 6 then if $rows != 4 then
return -1 return -1
endi endi
@ -75,7 +73,7 @@ while $i < 3
endw endw
sql show users sql show users
if $rows != 3 then if $rows != 1 then
return -1 return -1
endi endi