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

modify the code for synchronously creating tables

This commit is contained in:
slguan 2020-03-12 18:13:57 +08:00
parent c1d33817be
commit 66e5cd341d
18 changed files with 562 additions and 662 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/dnode/src/dnodeMClient.c
View File

@ -57,7 +57,6 @@ void dnodeCleanupMClient() {
} }
static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) { static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) {
if (dnodeProcessMgmtRspFp[pMsg->msgType]) { if (dnodeProcessMgmtRspFp[pMsg->msgType]) {
(*dnodeProcessMgmtRspFp[pMsg->msgType])(pMsg); (*dnodeProcessMgmtRspFp[pMsg->msgType])(pMsg);
} else { } else {

27
src/dnode/src/dnodeMgmt.c
View File

@ -98,14 +98,13 @@ void dnodeMgmt(SRpcMsg *pMsg) {
if (dnodeProcessMgmtMsgFp[pMsg->msgType]) { if (dnodeProcessMgmtMsgFp[pMsg->msgType]) {
(*dnodeProcessMgmtMsgFp[pMsg->msgType])(pMsg); (*dnodeProcessMgmtMsgFp[pMsg->msgType])(pMsg);
} else { } else {
terrno = TSDB_CODE_MSG_NOT_PROCESSED; SRpcMsg rsp;
rsp.handle = pMsg->handle;
rsp.code = TSDB_CODE_MSG_NOT_PROCESSED;
rsp.pCont = NULL;
rpcSendResponse(&rsp);
} }
SRpcMsg rsp;
rsp.handle = pMsg->handle;
rsp.code = terrno;
rsp.pCont = NULL;
rpcSendResponse(&rsp);
rpcFreeCont(pMsg->pCont); // free the received message rpcFreeCont(pMsg->pCont); // free the received message
} }
@ -275,15 +274,15 @@ static void dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions); pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile); pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId); // SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
if (pVnodeObj != NULL) { // if (pVnodeObj != NULL) {
rpcRsp.code = TSDB_CODE_SUCCESS; // rpcRsp.code = TSDB_CODE_SUCCESS;
} else { // } else {
rpcRsp.code = dnodeCreateVnode(pCreate); // rpcRsp.code = dnodeCreateVnode(pCreate);
} // }
rpcRsp.code = TSDB_CODE_SUCCESS;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
rpcFreeCont(rpcMsg->pCont);
} }
static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) { static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
@ -301,7 +300,6 @@ static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
} }
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
rpcFreeCont(rpcMsg->pCont);
} }
static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) { static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
@ -321,7 +319,6 @@ static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
} }
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
rpcFreeCont(rpcMsg->pCont);
} }
static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) { static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {

34
src/inc/mnode.h
View File

@ -39,12 +39,7 @@ extern "C" {
#include "ttimer.h" #include "ttimer.h"
#include "tutil.h" #include "tutil.h"
// internal globals typedef struct {
extern char version[];
extern void *tsMgmtTmr;
extern char tsMgmtDirectory[];
typedef struct {
uint32_t privateIp; uint32_t privateIp;
int32_t sid; int32_t sid;
uint32_t moduleStatus; uint32_t moduleStatus;
@ -87,11 +82,6 @@ typedef struct {
int32_t vnode; int32_t vnode;
} SVnodeGid; } SVnodeGid;
typedef struct {
int32_t sid;
int32_t vgId; // vnode group ID
} STableGid;
typedef struct { typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
int8_t type; int8_t type;
@ -248,16 +238,32 @@ typedef struct {
int16_t offset[TSDB_MAX_COLUMNS]; int16_t offset[TSDB_MAX_COLUMNS];
int16_t bytes[TSDB_MAX_COLUMNS]; int16_t bytes[TSDB_MAX_COLUMNS];
void * signature; void * signature;
uint16_t payloadLen; /* length of payload*/ uint16_t payloadLen;
char payload[]; /* payload for wildcard match in show tables */ char payload[];
} SShowObj; } SShowObj;
//mgmtSystem typedef struct {
uint8_t msgType;
int8_t expected;
int8_t received;
int8_t successed;
int32_t contLen;
int32_t code;
void *ahandle;
void *thandle;
void *pCont;
SDbObj *pDb;
SUserObj *pUser;
} SQueuedMsg;
int32_t mgmtInitSystem(); int32_t mgmtInitSystem();
int32_t mgmtStartSystem(); int32_t mgmtStartSystem();
void mgmtCleanUpSystem(); void mgmtCleanUpSystem();
void mgmtStopSystem(); void mgmtStopSystem();
extern char version[];
extern void *tsMgmtTmr;
extern char tsMgmtDirectory[];
#ifdef __cplusplus #ifdef __cplusplus
} }

1
src/inc/taosmsg.h
View File

@ -238,6 +238,7 @@ typedef struct SSchema {
} SSchema; } SSchema;
typedef struct { typedef struct {
int32_t vgId;
int32_t vnode; //the index of vnode int32_t vnode; //the index of vnode
uint32_t ip; uint32_t ip;
} SVnodeDesc; } SVnodeDesc;

16
src/mnode/inc/mgmtProfile.h
View File

@ -28,22 +28,6 @@ bool mgmtCheckQhandle(uint64_t qhandle);
void mgmtSaveQhandle(void *qhandle); void mgmtSaveQhandle(void *qhandle);
void mgmtFreeQhandle(void *qhandle); void mgmtFreeQhandle(void *qhandle);
enum {
TSDB_PROCESS_CREATE_VGROUP,
TSDB_PROCESS_CREATE_VGROUP_GET_META,
TSDB_PROCESS_CREATE_TABLE,
TSDB_PROCESS_CREATE_TABLE_GET_META,
};
typedef struct {
void *thandle; // come from uplayer
void *ahandle; // object to process
void *cont; // additional information of object to process
int32_t type; // the type of sync process
int32_t received; // num of received, such as numOfVnodes
int32_t contLen; // the length of additional information
} SProcessInfo;
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

5
src/mnode/inc/mgmtShell.h
View File

@ -23,13 +23,16 @@ extern "C" {
int32_t mgmtInitShell(); int32_t mgmtInitShell();
void mgmtCleanUpShell(); void mgmtCleanUpShell();
void mgmtAddShellMsgHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)); void mgmtAddShellMsgHandle(uint8_t msgType, void (*fp)(SQueuedMsg *queuedMsg));
typedef int32_t (*SShowMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn); typedef int32_t (*SShowMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn); typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtAddShellShowMetaHandle(uint8_t showType, SShowMetaFp fp); void mgmtAddShellShowMetaHandle(uint8_t showType, SShowMetaFp fp);
void mgmtAddShellShowRetrieveHandle(uint8_t showType, SShowRetrieveFp fp); void mgmtAddShellShowRetrieveHandle(uint8_t showType, SShowRetrieveFp fp);
void mgmtAddToShellQueue(SQueuedMsg *queuedMsg);
void mgmtSendSimpleResp(void *thandle, int32_t code);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

5
src/mnode/inc/mgmtTable.h
View File

@ -33,7 +33,6 @@ STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid);
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp); int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp);
int32_t mgmtRetrieveMetricMeta(void *pConn, char **pStart, SSuperTableMetaMsg *pInfo); int32_t mgmtRetrieveMetricMeta(void *pConn, char **pStart, SSuperTableMetaMsg *pInfo);
int32_t mgmtCreateTable(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore); int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore);
int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter); int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter);
@ -44,10 +43,6 @@ void mgmtSetTableDirty(STableInfo *pTable, bool isDirty);
SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable); SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable);
SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable); SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable);
void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
void mgmtProcessCreateTable(SVgObj *pVgroup, SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
void mgmtProcessCreateVgroup(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
src/mnode/inc/mgmtVgroup.h
View File

@ -29,7 +29,7 @@ void mgmtCleanUpVgroups();
SVgObj *mgmtGetVgroup(int32_t vgId); SVgObj *mgmtGetVgroup(int32_t vgId);
SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode); SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
SVgObj *mgmtCreateVgroup(SDbObj *pDb); void mgmtCreateVgroup(SQueuedMsg *pMsg);
int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup); int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup);
void mgmtUpdateVgroup(SVgObj *pVgroup); void mgmtUpdateVgroup(SVgObj *pVgroup);

4
src/mnode/src/mgmtBalance.c
View File

@ -56,10 +56,10 @@ int32_t mgmtAllocVnodes(SVgObj *pVgroup) {
} }
if (selectedVnode == -1) { if (selectedVnode == -1) {
mError("vgroup:%d alloc vnode failed, free vnodes:%d", pVgroup->vgId, pDnode->numOfFreeVnodes); mError("alloc vnode failed, free vnodes:%d", pDnode->numOfFreeVnodes);
return -1; return -1;
} else { } else {
mTrace("vgroup:%d allocate vnode:%d, last allocated vnode:%d", pVgroup->vgId, selectedVnode, lastAllocVode); mTrace("allocate vnode:%d, last allocated vnode:%d", selectedVnode, lastAllocVode);
pVgroup->vnodeGid[0].vnode = selectedVnode; pVgroup->vnodeGid[0].vnode = selectedVnode;
pDnode->lastAllocVnode = selectedVnode + 1; pDnode->lastAllocVnode = selectedVnode + 1;
if (pDnode->lastAllocVnode >= pDnode->numOfVnodes) pDnode->lastAllocVnode = 0; if (pDnode->lastAllocVnode >= pDnode->numOfVnodes) pDnode->lastAllocVnode = 0;

54
src/mnode/src/mgmtDClient.c
View File

@ -84,39 +84,6 @@ static void mgmtProcessRspFromDnode(SRpcMsg *rpcMsg) {
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
} }
//static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
// mTrace("create table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
// if (rpcMsg->handle == NULL) return;
//
// SProcessInfo *info = rpcMsg->handle;
// assert(info->type == TSDB_PROCESS_CREATE_TABLE || info->type == TSDB_PROCESS_CREATE_TABLE_GET_META);
//
// STableInfo *pTable = info->ahandle;
// if (rpcMsg->code != TSDB_CODE_SUCCESS) {
// mError("table:%s, failed to create in dnode, code:%d, set it dirty", pTable->tableId, rpcMsg->code);
// mgmtSetTableDirty(pTable, true);
// } else {
// mTrace("table:%s, created in dnode", pTable->tableId);
// mgmtSetTableDirty(pTable, false);
// }
//
// if (rpcMsg->code != TSDB_CODE_SUCCESS) {
// SRpcMsg rpcRsp = {.handle = info->thandle, .pCont = NULL, .contLen = 0, .code = rpcMsg->code, .msgType = 0};
// rpcSendResponse(&rpcMsg);
// } else {
// if (info->type == TSDB_PROCESS_CREATE_TABLE_GET_META) {
// mTrace("table:%s, start to process get meta", pTable->tableId);
// mgmtProcessGetTableMeta(pTable, rpcMsg->handle);
// } else {
// SRpcMsg rpcRsp = {.handle = info->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
// rpcSendResponse(&rpcMsg);
// }
// }
//
// free(info);
//}
//
//static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) { //static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
// mTrace("drop table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code); // mTrace("drop table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//} //}
@ -125,27 +92,6 @@ static void mgmtProcessRspFromDnode(SRpcMsg *rpcMsg) {
// mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code); // mTrace("alter table rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
//} //}
// //
//static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
// mTrace("create vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);
// if (rpcMsg->handle == NULL) return;
//
// SProcessInfo *info = rpcMsg->handle;
// assert(info->type == TSDB_PROCESS_CREATE_VGROUP || info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META);
//
// info->received++;
// SVgObj *pVgroup = info->ahandle;
//
// bool isGetMeta = false;
// if (info->type == TSDB_PROCESS_CREATE_VGROUP_GET_META) {
// isGetMeta = true;
// }
//
// mTrace("vgroup:%d, received:%d numOfVnodes:%d", pVgroup->vgId, info->received, pVgroup->numOfVnodes);
// if (info->received == pVgroup->numOfVnodes) {
// mgmtProcessCreateTable(pVgroup, info->cont, info->contLen, info->thandle, isGetMeta);
// free(info);
// }
//}
// //
//static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) { //static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
// mTrace("drop vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code); // mTrace("drop vnode rsp received, handle:%p code:%d", rpcMsg->handle, rpcMsg->code);

16
src/mnode/src/mgmtDServer.c
View File

@ -210,22 +210,6 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
// } // }
//} //}
// //
//void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
// mTrace("vgroup:%d, send create all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
// for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
// mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
// }
//}
//
//void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, vnode, ahandle);
// SMDCreateVnodeMsg *pVpeer = mgmtBuildCreateVnodeMsg(pVgroup, vnode);
// if (pVpeer != NULL) {
// mgmtSendMsgToDnode(ipSet, TSDB_MSG_TYPE_MD_CREATE_VNODE, pVpeer, sizeof(SMDCreateVnodeMsg), ahandle);
// }
//}
//
//void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) { //void mgmtProcessMsgFromDnode(char msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
// if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) { // if (msgType < 0 || msgType >= TSDB_MSG_TYPE_MAX) {
// mError("invalid msg type:%d", msgType); // mError("invalid msg type:%d", msgType);

98
src/mnode/src/mgmtDb.c
View File

@ -41,9 +41,9 @@ static int32_t mgmtDropDb(SDbObj *pDb);
static int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCreateDbMsg(SRpcMsg *rpcMsg); static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg);
static void mgmtProcessAlterDbMsg(SRpcMsg *rpcMsg); static void mgmtProcessAlterDbMsg(SQueuedMsg *pMsg);
static void mgmtProcessDropDbMsg(SRpcMsg *rpcMsg); static void mgmtProcessDropDbMsg(SQueuedMsg *pMsg);
static void *(*mgmtDbActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize); static void *(*mgmtDbActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtDbActionInsert(void *row, char *str, int32_t size, int32_t *ssize); static void *mgmtDbActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
@ -383,6 +383,7 @@ static void mgmtDropDbFromSdb(SDbObj *pDb) {
} }
static int32_t mgmtDropDb(SDbObj *pDb) { static int32_t mgmtDropDb(SDbObj *pDb) {
if (pDb->dropStatus == TSDB_DB_STATUS_DROPPING) { if (pDb->dropStatus == TSDB_DB_STATUS_DROPPING) {
bool finished = mgmtCheckDropDbFinished(pDb); bool finished = mgmtCheckDropDbFinished(pDb);
if (!finished) { if (!finished) {
@ -405,6 +406,7 @@ static int32_t mgmtDropDb(SDbObj *pDb) {
} }
} }
UNUSED_FUNC
static int32_t mgmtDropDbByName(SAcctObj *pAcct, char *name, short ignoreNotExists) { static int32_t mgmtDropDbByName(SAcctObj *pAcct, char *name, short ignoreNotExists) {
SDbObj *pDb = (SDbObj *)sdbGetRow(tsDbSdb, name); SDbObj *pDb = (SDbObj *)sdbGetRow(tsDbSdb, name);
if (pDb == NULL) { if (pDb == NULL) {
@ -904,19 +906,10 @@ void mgmtRemoveTableFromDb(SDbObj *pDb) {
atomic_add_fetch_32(&pDb->numOfTables, -1); atomic_add_fetch_32(&pDb->numOfTables, -1);
} }
static void mgmtProcessCreateDbMsg(SRpcMsg *rpcMsg) { static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) return;
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle);
if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return;
}
SCMCreateDbMsg *pCreate = (SCMCreateDbMsg *) rpcMsg->pCont;
SCMCreateDbMsg *pCreate = pMsg->pCont;
pCreate->maxSessions = htonl(pCreate->maxSessions); pCreate->maxSessions = htonl(pCreate->maxSessions);
pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize); pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize);
pCreate->daysPerFile = htonl(pCreate->daysPerFile); pCreate->daysPerFile = htonl(pCreate->daysPerFile);
@ -928,69 +921,58 @@ static void mgmtProcessCreateDbMsg(SRpcMsg *rpcMsg) {
pCreate->rowsInFileBlock = htonl(pCreate->rowsInFileBlock); pCreate->rowsInFileBlock = htonl(pCreate->rowsInFileBlock);
// pCreate->cacheNumOfBlocks = htonl(pCreate->cacheNumOfBlocks); // pCreate->cacheNumOfBlocks = htonl(pCreate->cacheNumOfBlocks);
int32_t code;
if (mgmtCheckExpired()) { if (mgmtCheckExpired()) {
rpcRsp.code = TSDB_CODE_GRANT_EXPIRED; code = TSDB_CODE_GRANT_EXPIRED;
} else if (!pUser->writeAuth) { } else if (!pMsg->pUser->writeAuth) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} else { } else {
rpcRsp.code = mgmtCreateDb(pUser->pAcct, pCreate); code = mgmtCreateDb(pMsg->pUser->pAcct, pCreate);
if (rpcRsp.code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
mLPrint("DB:%s is created by %s", pCreate->db, pUser->user); mLPrint("DB:%s is created by %s", pCreate->db, pMsg->pUser->user);
} }
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
} }
static void mgmtProcessAlterDbMsg(SRpcMsg *rpcMsg) { static void mgmtProcessAlterDbMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) return;
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SCMAlterDbMsg *pAlter = pMsg->pCont;
if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return;
}
SCMAlterDbMsg *pAlter = (SCMAlterDbMsg *) rpcMsg->pCont;
pAlter->daysPerFile = htonl(pAlter->daysPerFile); pAlter->daysPerFile = htonl(pAlter->daysPerFile);
pAlter->daysToKeep = htonl(pAlter->daysToKeep); pAlter->daysToKeep = htonl(pAlter->daysToKeep);
pAlter->maxSessions = htonl(pAlter->maxSessions) + 1; pAlter->maxSessions = htonl(pAlter->maxSessions) + 1;
if (!pUser->writeAuth) { int32_t code;
rpcRsp.code = TSDB_CODE_NO_RIGHTS; if (!pMsg->pUser->writeAuth) {
code = TSDB_CODE_NO_RIGHTS;
} else { } else {
rpcRsp.code = mgmtAlterDb(pUser->pAcct, pAlter); code = mgmtAlterDb(pMsg->pUser->pAcct, pAlter);
if (rpcRsp.code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
mLPrint("DB:%s is altered by %s", pAlter->db, pUser->user); mLPrint("DB:%s is altered by %s", pAlter->db, pMsg->pUser->user);
} }
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
} }
static void mgmtProcessDropDbMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return;
static void mgmtProcessDropDbMsg(SRpcMsg *rpcMsg) { int32_t code;
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (pMsg->pUser->superAuth) {
if (mgmtCheckRedirect(rpcMsg->handle)) return; code = TSDB_CODE_OPS_NOT_SUPPORT;
//SCMDropDbMsg *pDrop = rpcMsg->pCont;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); //rpcRsp.code = mgmtDropDbByName(pUser->pAcct, pDrop->db, pDrop->ignoreNotExists);
if (pUser == NULL) { //if (rpcRsp.code == TSDB_CODE_SUCCESS) {
rpcRsp.code = TSDB_CODE_INVALID_USER; // mLPrint("DB:%s is dropped by %s", pDrop->db, pUser->user);
rpcSendResponse(&rpcRsp); //}
return ;
}
if (pUser->superAuth) {
SCMDropDbMsg *pDrop = rpcMsg->pCont;
rpcRsp.code = mgmtDropDbByName(pUser->pAcct, pDrop->db, pDrop->ignoreNotExists);
if (rpcRsp.code == TSDB_CODE_SUCCESS) {
mLPrint("DB:%s is dropped by %s", pDrop->db, pUser->user);
}
} else { } else {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} }
rpcSendResponse(&rpcRsp); if (code != TSDB_CODE_SUCCESS) {
mgmtSendSimpleResp(pMsg->thandle, code);
}
} }

23
src/mnode/src/mgmtDnode.c
View File

@ -43,7 +43,7 @@ static int32_t mgmtGetConfigMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn
static int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCfgDnodeMsg(SRpcMsg *rpcMsg); static void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg);
void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) { void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode) {
int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore; int32_t maxVnodes = pDnode->numOfCores * tsNumOfVnodesPerCore;
@ -93,7 +93,7 @@ void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId) {
memset(pVload, 0, sizeof(SVnodeLoad)); memset(pVload, 0, sizeof(SVnodeLoad));
pVload->vnode = vnodeGid[i].vnode; pVload->vnode = vnodeGid[i].vnode;
pVload->vgId = vgId; pVload->vgId = vgId;
mTrace("dnode:%s, vnode:%d add to vgroup:%d", taosIpStr(vnodeGid[i].ip), vnodeGid[i].vnode, pVload->vgId); mTrace("dnode:%s, vnode:%d add to vgroup:%d", taosIpStr(pDnode->privateIp), vnodeGid[i].vnode, pVload->vgId);
mgmtCalcNumOfFreeVnodes(pDnode); mgmtCalcNumOfFreeVnodes(pDnode);
} else { } else {
mError("dnode:%s, not in dnode DB!!!", taosIpStr(vnodeGid[i].ip)); mError("dnode:%s, not in dnode DB!!!", taosIpStr(vnodeGid[i].ip));
@ -527,21 +527,14 @@ bool mgmtCheckDnodeInOfflineState(SDnodeObj *pDnode) {
return pDnode->status == TSDB_DN_STATUS_OFFLINE; return pDnode->status == TSDB_DN_STATUS_OFFLINE;
} }
void mgmtProcessCfgDnodeMsg(SRpcMsg *rpcMsg) { void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
if (mgmtCheckRedirect(rpcMsg->handle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SCMCfgDnodeMsg *pCmCfgDnode = pMsg->pCont;
if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return;
}
SCMCfgDnodeMsg *pCmCfgDnode = (SCMCfgDnodeMsg *) rpcMsg->pCont;
uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip); uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip);
if (strcmp(pUser->pAcct->user, "root") != 0) { if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else { } else {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(dnodeIp); SRpcIpSet ipSet = mgmtGetIpSetFromIp(dnodeIp);
@ -560,7 +553,7 @@ void mgmtProcessCfgDnodeMsg(SRpcMsg *rpcMsg) {
} }
if (rpcRsp.code == TSDB_CODE_SUCCESS) { if (rpcRsp.code == TSDB_CODE_SUCCESS) {
mTrace("dnode:%s is configured by %s", pCmCfgDnode->ip, pUser->user); mTrace("dnode:%s is configured by %s", pCmCfgDnode->ip, pMsg->pUser->user);
} }
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);

38
src/mnode/src/mgmtProfile.c
View File

@ -558,9 +558,11 @@ bool mgmtCheckQhandle(uint64_t qhandle) {
} }
void mgmtSaveQhandle(void *qhandle) { void mgmtSaveQhandle(void *qhandle) {
mTrace("qhandle:%p is allocated", qhandle);
} }
void mgmtFreeQhandle(void *qhandle) { void mgmtFreeQhandle(void *qhandle) {
mTrace("qhandle:%p is freed", qhandle);
} }
int mgmtGetConns(SShowObj *pShow, void *pConn) { int mgmtGetConns(SShowObj *pShow, void *pConn) {
@ -673,72 +675,72 @@ int32_t mgmtRetrieveConns(SShowObj *pShow, char *data, int32_t rows, void *pConn
return numOfRows; return numOfRows;
} }
void mgmtProcessKillQueryMsg(SRpcMsg *rpcMsg) { void mgmtProcessKillQueryMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
if (mgmtCheckRedirect(rpcMsg->handle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
return; return;
} }
SCMKillQueryMsg *pKill = (SCMKillQueryMsg *) rpcMsg->pCont; SCMKillQueryMsg *pKill = pMsg->pCont;
int32_t code; int32_t code;
if (!pUser->writeAuth) { if (!pUser->writeAuth) {
code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} else { } else {
code = mgmtKillQuery(pKill->queryId, rpcMsg->handle); code = mgmtKillQuery(pKill->queryId, pMsg->thandle);
} }
rpcRsp.code = code; rpcRsp.code = code;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
void mgmtProcessKillStreamMsg(SRpcMsg *rpcMsg) { void mgmtProcessKillStreamMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
if (mgmtCheckRedirect(rpcMsg->handle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
return; return;
} }
SCMKillStreamMsg *pKill = (SCMKillStreamMsg *) rpcMsg->pCont; SCMKillStreamMsg *pKill = pMsg->pCont;
int32_t code; int32_t code;
if (!pUser->writeAuth) { if (!pUser->writeAuth) {
code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} else { } else {
code = mgmtKillStream(pKill->queryId, rpcMsg->handle); code = mgmtKillStream(pKill->queryId, pMsg->thandle);
} }
rpcRsp.code = code; rpcRsp.code = code;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
void mgmtProcessKillConnectionMsg(SRpcMsg *rpcMsg) { void mgmtProcessKillConnectionMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
if (mgmtCheckRedirect(rpcMsg->handle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
return; return;
} }
SCMKillConnMsg *pKill = (SCMKillConnMsg *) rpcMsg->pCont; SCMKillConnMsg *pKill = pMsg->pCont;
int32_t code; int32_t code;
if (!pUser->writeAuth) { if (!pUser->writeAuth) {
code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} else { } else {
code = mgmtKillConnection(pKill->queryId, rpcMsg->handle); code = mgmtKillConnection(pKill->queryId, pMsg->thandle);
} }
rpcRsp.code = code; rpcRsp.code = code;

246
src/mnode/src/mgmtShell.c
View File

@ -41,24 +41,27 @@
typedef int32_t (*SShowMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn); typedef int32_t (*SShowMetaFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn); typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessMsgFromShell(SRpcMsg *pMsg);
static void mgmtProcessShowMsg(SRpcMsg *rpcMsg);
static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg);
static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg);
static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey); static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static bool mgmtCheckMsgReadOnly(int8_t type, void *pCont); static bool mgmtCheckMsgReadOnly(int8_t type, void *pCont);
static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg); static void mgmtProcessMsgFromShell(SRpcMsg *pMsg);
static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg); static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg);
static void mgmtProcessMsgWhileNotReady(SRpcMsg *rpcMsg);
static void mgmtProcessShowMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessRetrieveMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessHeartBeatMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessConnectMsg(SQueuedMsg *queuedMsg);
static void *tsMgmtShellRpc = NULL; static void *tsMgmtShellRpc = NULL;
static void *tsMgmtTranQhandle = NULL; static void *tsMgmtTranQhandle = NULL;
static void (*tsMgmtProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *) = {0}; static void (*tsMgmtProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SQueuedMsg *) = {0};
static SShowMetaFp tsMgmtShowMetaFp[TSDB_MGMT_TABLE_MAX] = {0}; static SShowMetaFp tsMgmtShowMetaFp[TSDB_MGMT_TABLE_MAX] = {0};
static SShowRetrieveFp tsMgmtShowRetrieveFp[TSDB_MGMT_TABLE_MAX] = {0}; static SShowRetrieveFp tsMgmtShowRetrieveFp[TSDB_MGMT_TABLE_MAX] = {0};
int32_t mgmtInitShell() { int32_t mgmtInitShell() {
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_SHOW, mgmtProcessShowMsg); mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_SHOW, mgmtProcessShowMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_RETRIEVE, mgmtProcessRetrieveMsg); mgmtAddShellMsgHandle(TSDB_MSG_TYPE_RETRIEVE, mgmtProcessRetrieveMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_HEARTBEAT, mgmtProcessHeartBeatMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mgmtProcessConnectMsg);
tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT"); tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT");
@ -84,9 +87,6 @@ int32_t mgmtInitShell() {
return -1; return -1;
} }
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_HEARTBEAT, mgmtProcessHeartBeatMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mgmtProcessConnectMsg);
mPrint("server connection to shell is opened"); mPrint("server connection to shell is opened");
return 0; return 0;
} }
@ -104,7 +104,7 @@ void mgmtCleanUpShell() {
} }
} }
void mgmtAddShellMsgHandle(uint8_t showType, void (*fp)(SRpcMsg *rpcMsg)) { void mgmtAddShellMsgHandle(uint8_t showType, void (*fp)(SQueuedMsg *queuedMsg)) {
tsMgmtProcessShellMsgFp[showType] = fp; tsMgmtProcessShellMsgFp[showType] = fp;
} }
@ -117,107 +117,118 @@ void mgmtAddShellShowRetrieveHandle(uint8_t msgType, SShowRetrieveFp fp) {
} }
void mgmtProcessTranRequest(SSchedMsg *sched) { void mgmtProcessTranRequest(SSchedMsg *sched) {
SRpcMsg *rpcMsg = sched->msg; SQueuedMsg *queuedMsg = sched->msg;
(*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(rpcMsg); (*tsMgmtProcessShellMsgFp[queuedMsg->msgType])(queuedMsg);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(queuedMsg->pCont);
free(rpcMsg); free(queuedMsg);
} }
void mgmtAddToTranRequest(SRpcMsg *rpcMsg) { void mgmtAddToShellQueue(SQueuedMsg *queuedMsg) {
SRpcMsg *queuedRpcMsg = malloc(sizeof(SRpcMsg));
memcpy(queuedRpcMsg, rpcMsg, sizeof(SRpcMsg));
SSchedMsg schedMsg; SSchedMsg schedMsg;
schedMsg.msg = queuedRpcMsg; schedMsg.msg = queuedMsg;
schedMsg.fp = mgmtProcessTranRequest; schedMsg.fp = mgmtProcessTranRequest;
taosScheduleTask(tsMgmtTranQhandle, &schedMsg); taosScheduleTask(tsMgmtTranQhandle, &schedMsg);
} }
static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) { static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
if (sdbGetRunStatus() != SDB_STATUS_SERVING) { if (sdbGetRunStatus() != SDB_STATUS_SERVING) {
mTrace("shell msg is ignored since SDB is not ready"); mgmtProcessMsgWhileNotReady(rpcMsg);
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = TSDB_CODE_NOT_READY, .msgType = 0};
rpcSendResponse(&rpcRsp);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
return; return;
} }
mTrace("%s is received", taosMsg[rpcMsg->msgType]); if (tsMgmtProcessShellMsgFp[rpcMsg->msgType] == NULL) {
if (tsMgmtProcessShellMsgFp[rpcMsg->msgType]) {
if (mgmtCheckMsgReadOnly(rpcMsg->msgType, rpcMsg->pCont)) {
(*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(rpcMsg);
rpcFreeCont(rpcMsg->pCont);
} else {
mgmtAddToTranRequest(rpcMsg);
}
} else {
mError("%s is not processed", taosMsg[rpcMsg->msgType]);
mgmtProcessUnSupportMsg(rpcMsg); mgmtProcessUnSupportMsg(rpcMsg);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
return;
}
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle);
if (pUser == NULL) {
mError("thandle:%p, failed to retrieve user info", rpcMsg->handle);
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_INVALID_USER);
rpcFreeCont(rpcMsg->pCont);
return;
}
if (mgmtCheckMsgReadOnly(rpcMsg->msgType, rpcMsg->pCont)) {
SQueuedMsg queuedMsg = {0};
queuedMsg.thandle = rpcMsg->handle;
queuedMsg.msgType = rpcMsg->msgType;
queuedMsg.contLen = rpcMsg->contLen;
queuedMsg.pCont = rpcMsg->pCont;
queuedMsg.pUser = pUser;
(*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(&queuedMsg);
rpcFreeCont(rpcMsg->pCont);
} else {
SQueuedMsg *queuedMsg = calloc(1, sizeof(SQueuedMsg));
queuedMsg->thandle = rpcMsg->handle;
queuedMsg->msgType = rpcMsg->msgType;
queuedMsg->contLen = rpcMsg->contLen;
queuedMsg->pCont = rpcMsg->pCont;
queuedMsg->pUser = pUser;
mgmtAddToShellQueue(queuedMsg);
} }
} }
static void mgmtProcessShowMsg(SRpcMsg *rpcMsg) { static void mgmtProcessShowMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SCMShowMsg *pShowMsg = pMsg->pCont;
SCMShowMsg *pShowMsg = rpcMsg->pCont;
if (pShowMsg->type == TSDB_MGMT_TABLE_DNODE || TSDB_MGMT_TABLE_GRANTS || TSDB_MGMT_TABLE_SCORES) { if (pShowMsg->type == TSDB_MGMT_TABLE_DNODE || TSDB_MGMT_TABLE_GRANTS || TSDB_MGMT_TABLE_SCORES) {
if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) { if (mgmtCheckRedirect(pMsg->thandle)) {
return; return;
} }
} }
int32_t size = sizeof(SCMShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE; if (pShowMsg->type >= TSDB_MGMT_TABLE_MAX) {
SCMShowRsp *pShowRsp = rpcMallocCont(size); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_MSG_TYPE);
if (pShowRsp == NULL) {
rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY;
rpcSendResponse(&rpcRsp);
return; return;
} }
int32_t code; if (!tsMgmtShowMetaFp[pShowMsg->type]) {
if (pShowMsg->type >= TSDB_MGMT_TABLE_MAX) { mError("show type:%d %s is not support", pShowMsg->type, taosMsg[pShowMsg->type]);
code = TSDB_CODE_INVALID_MSG_TYPE; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_OPS_NOT_SUPPORT);
} else { return;
SShowObj *pShow = (SShowObj *) calloc(1, sizeof(SShowObj) + htons(pShowMsg->payloadLen));
pShow->signature = pShow;
pShow->type = pShowMsg->type;
strcpy(pShow->db, pShowMsg->db);
mTrace("pShow:%p is allocated", pShow);
// set the table name query condition
pShow->payloadLen = htons(pShowMsg->payloadLen);
memcpy(pShow->payload, pShowMsg->payload, pShow->payloadLen);
mgmtSaveQhandle(pShow);
pShowRsp->qhandle = htobe64((uint64_t) pShow);
if (tsMgmtShowMetaFp[pShowMsg->type]) {
code = (*tsMgmtShowMetaFp[pShowMsg->type])(&pShowRsp->tableMeta, pShow, rpcMsg->handle);
if (code == 0) {
size = sizeof(SCMShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
} else {
mError("pShow:%p, type:%d %s, failed to get Meta, code:%d", pShow, pShowMsg->type,
taosMsg[(uint8_t) pShowMsg->type], code);
free(pShow);
}
} else {
code = TSDB_CODE_OPS_NOT_SUPPORT;
}
} }
rpcRsp.code = code; int32_t size = sizeof(SCMShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
rpcRsp.pCont = pShowRsp; SCMShowRsp *pShowRsp = rpcMallocCont(size);
rpcRsp.contLen = size; if (pShowRsp == NULL) {
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
return;
}
SShowObj *pShow = (SShowObj *) calloc(1, sizeof(SShowObj) + htons(pShowMsg->payloadLen));
pShow->signature = pShow;
pShow->type = pShowMsg->type;
pShow->payloadLen = htons(pShowMsg->payloadLen);
strcpy(pShow->db, pShowMsg->db);
memcpy(pShow->payload, pShowMsg->payload, pShow->payloadLen);
mgmtSaveQhandle(pShow);
pShowRsp->qhandle = htobe64((uint64_t) pShow);
int32_t code = (*tsMgmtShowMetaFp[pShowMsg->type])(&pShowRsp->tableMeta, pShow, pMsg->thandle);
if (code == 0) {
SRpcMsg rpcRsp = {
.handle = pMsg->thandle,
.pCont = pShowRsp,
.contLen = sizeof(SCMShowRsp) + sizeof(SSchema) * pShow->numOfColumns,
.code = code,
.msgType = 0
};
rpcSendResponse(&rpcRsp);
} else {
mError("pShow:%p, type:%d %s, failed to get Meta, code:%d", pShow, pShowMsg->type, taosMsg[pShowMsg->type], code);
mgmtFreeQhandle(pShow);
rpcFreeCont(pShowRsp);
}
} }
static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) { static void mgmtProcessRetrieveMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
int32_t rowsToRead = 0; int32_t rowsToRead = 0;
int32_t size = 0; int32_t size = 0;
int32_t rowsRead = 0; int32_t rowsRead = 0;
SRetrieveTableMsg *pRetrieve = (SRetrieveTableMsg *) rpcMsg->pCont; SRetrieveTableMsg *pRetrieve = pMsg->pCont;
pRetrieve->qhandle = htobe64(pRetrieve->qhandle); pRetrieve->qhandle = htobe64(pRetrieve->qhandle);
/* /*
@ -226,16 +237,14 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
*/ */
if (!mgmtCheckQhandle(pRetrieve->qhandle)) { if (!mgmtCheckQhandle(pRetrieve->qhandle)) {
mError("retrieve:%p, qhandle:%p is invalid", pRetrieve, pRetrieve->qhandle); mError("retrieve:%p, qhandle:%p is invalid", pRetrieve, pRetrieve->qhandle);
rpcRsp.code = TSDB_CODE_INVALID_QHANDLE; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_QHANDLE);
rpcSendResponse(&rpcRsp);
return; return;
} }
SShowObj *pShow = (SShowObj *)pRetrieve->qhandle; SShowObj *pShow = (SShowObj *)pRetrieve->qhandle;
if (pShow->signature != (void *)pShow) { if (!mgmtCheckQhandle(pShow)) {
mError("pShow:%p, signature:%p, query memory is corrupted", pShow, pShow->signature); mError("pShow:%p, query memory is corrupted", pShow);
rpcRsp.code = TSDB_CODE_MEMORY_CORRUPTED; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_MEMORY_CORRUPTED);
rpcSendResponse(&rpcRsp);
return; return;
} else { } else {
if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) { if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) {
@ -258,10 +267,9 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
// if free flag is set, client wants to clean the resources // if free flag is set, client wants to clean the resources
if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE)
rowsRead = (*tsMgmtShowRetrieveFp[pShow->type])(pShow, pRsp->data, rowsToRead, rpcMsg->handle); rowsRead = (*tsMgmtShowRetrieveFp[pShow->type])(pShow, pRsp->data, rowsToRead, pMsg->thandle);
if (rowsRead < 0) { if (rowsRead < 0) { // TSDB_CODE_ACTION_IN_PROGRESS;
rowsRead = 0; // TSDB_CODE_ACTION_IN_PROGRESS;
rpcFreeCont(pRsp); rpcFreeCont(pRsp);
return; return;
} }
@ -269,8 +277,13 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
pRsp->numOfRows = htonl(rowsRead); pRsp->numOfRows = htonl(rowsRead);
pRsp->precision = htonl(TSDB_TIME_PRECISION_MILLI); // millisecond time precision pRsp->precision = htonl(TSDB_TIME_PRECISION_MILLI); // millisecond time precision
rpcRsp.pCont = pRsp; SRpcMsg rpcRsp = {
rpcRsp.contLen = size; .handle = pMsg->thandle,
.pCont = pRsp,
.contLen = size,
.code = 0,
.msgType = 0
};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
if (rowsToRead == 0) { if (rowsToRead == 0) {
@ -278,21 +291,19 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
} }
} }
static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg) { static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
//SCMHeartBeatMsg *pHBMsg = (SCMHeartBeatMsg *) rpcMsg->pCont; //SCMHeartBeatMsg *pHBMsg = (SCMHeartBeatMsg *) rpcMsg->pCont;
//mgmtSaveQueryStreamList(pHBMsg); //mgmtSaveQueryStreamList(pHBMsg);
SCMHeartBeatRsp *pHBRsp = (SCMHeartBeatRsp *) rpcMallocCont(sizeof(SCMHeartBeatRsp)); SCMHeartBeatRsp *pHBRsp = (SCMHeartBeatRsp *) rpcMallocCont(sizeof(SCMHeartBeatRsp));
if (pHBRsp == NULL) { if (pHBRsp == NULL) {
rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
rpcSendResponse(&rpcRsp);
return; return;
} }
SRpcConnInfo connInfo; SRpcConnInfo connInfo;
if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) { if (rpcGetConnInfo(pMsg->thandle, &connInfo) != 0) {
mError("conn:%p is already released while process heart beat msg", rpcMsg->handle); mError("conn:%p is already released while process heart beat msg", pMsg->thandle);
return; return;
} }
@ -320,8 +331,13 @@ static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg) {
pHBRsp->streamId = 0; pHBRsp->streamId = 0;
pHBRsp->killConnection = 0; pHBRsp->killConnection = 0;
rpcRsp.pCont = pHBRsp; SRpcMsg rpcRsp = {
rpcRsp.contLen = sizeof(SCMHeartBeatRsp); .handle = pMsg->thandle,
.pCont = pHBRsp,
.contLen = sizeof(SCMHeartBeatRsp),
.code = 0,
.msgType = 0
};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
@ -340,13 +356,13 @@ static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secr
} }
} }
static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg) { static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SCMConnectMsg *pConnectMsg = (SCMConnectMsg *) rpcMsg->pCont; SCMConnectMsg *pConnectMsg = pMsg->pCont;
SRpcConnInfo connInfo; SRpcConnInfo connInfo;
if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) { if (rpcGetConnInfo(pMsg->thandle, &connInfo) != 0) {
mError("conn:%p is already released while process connect msg", rpcMsg->handle); mError("thandle:%p is already released while process connect msg", pMsg->thandle);
return; return;
} }
@ -450,6 +466,7 @@ static bool mgmtCheckMsgReadOnly(int8_t type, void *pCont) {
} }
static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg) { static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg) {
mError("%s is not processed", taosMsg[rpcMsg->msgType]);
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = 0, .msgType = 0,
.pCont = 0, .pCont = 0,
@ -459,3 +476,26 @@ static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
static void mgmtProcessMsgWhileNotReady(SRpcMsg *rpcMsg) {
mTrace("%s is ignored since SDB is not ready", taosMsg[rpcMsg->msgType]);
SRpcMsg rpcRsp = {
.msgType = 0,
.pCont = 0,
.contLen = 0,
.code = TSDB_CODE_NOT_READY,
.handle = rpcMsg->handle
};
rpcSendResponse(&rpcRsp);
}
void mgmtSendSimpleResp(void *thandle, int32_t code) {
SRpcMsg rpcRsp = {
.msgType = 0,
.pCont = 0,
.contLen = 0,
.code = code,
.handle = thandle
};
rpcSendResponse(&rpcRsp);
}

426
src/mnode/src/mgmtTable.c
View File

@ -44,14 +44,16 @@
extern void *tsNormalTableSdb; extern void *tsNormalTableSdb;
extern void *tsChildTableSdb; extern void *tsChildTableSdb;
static void mgmtProcessCreateTableMsg(SRpcMsg *rpcMsg); static void mgmtProcessCreateTableMsg(SQueuedMsg *queueMsg);
static void mgmtProcessDropTableMsg(SRpcMsg *rpcMsg); static void mgmtProcessDropTableMsg(SQueuedMsg *queueMsg);
static void mgmtProcessAlterTableMsg(SRpcMsg *rpcMsg); static void mgmtProcessAlterTableMsg(SQueuedMsg *queueMsg);
static void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg); static void mgmtProcessTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg); static void mgmtProcessMultiTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessSuperTableMetaMsg(SRpcMsg *rpcMsg); static void mgmtProcessSuperTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg);
static int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
int32_t mgmtInitTables() { int32_t mgmtInitTables() {
int32_t code = mgmtInitSuperTables(); int32_t code = mgmtInitSuperTables();
@ -79,6 +81,7 @@ int32_t mgmtInitTables() {
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_STABLE_META, mgmtProcessSuperTableMetaMsg); mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_STABLE_META, mgmtProcessSuperTableMetaMsg);
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mgmtGetShowTableMeta); mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mgmtGetShowTableMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables); mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP, mgmtProcessCreateTableRsp);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -131,170 +134,56 @@ int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, boo
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void mgmtCreateTable(SVgObj *pVgroup, SQueuedMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->pCont;
pCreate->numOfColumns = htons(pCreate->numOfColumns);
pCreate->numOfTags = htons(pCreate->numOfTags);
pCreate->sqlLen = htons(pCreate->sqlLen);
void mgmtProcessCreateVgroup(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) { SSchema *pSchema = (SSchema*) pCreate->schema;
SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; for (int32_t i = 0; i < pCreate->numOfColumns + pCreate->numOfTags; ++i) {
SDbObj *pDb = mgmtGetDb(pCreate->db); pSchema->bytes = htons(pSchema->bytes);
if (pDb == NULL) { pSchema->colId = i;
mError("table:%s, failed to create vgroup, db not found", pCreate->tableId); pSchema++;
rpcRsp.code = TSDB_CODE_INVALID_DB;
rpcSendResponse(&rpcRsp);
return;
} }
SVgObj *pVgroup = mgmtCreateVgroup(pDb);
if (pVgroup == NULL) {
mError("table:%s, failed to alloc vnode to vgroup", pCreate->tableId);
rpcRsp.code = TSDB_CODE_NO_ENOUGH_DNODES;
rpcSendResponse(&rpcRsp);
return;
}
void *cont = rpcMallocCont(contLen);
if (cont == NULL) {
mError("table:%s, failed to create table, can not alloc memory", pCreate->tableId);
rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY;
rpcSendResponse(&rpcRsp);
return;
}
memcpy(cont, pCreate, contLen);
SProcessInfo *info = calloc(1, sizeof(SProcessInfo));
info->type = TSDB_PROCESS_CREATE_VGROUP;
info->thandle = thandle;
info->ahandle = pVgroup;
info->cont = cont;
info->contLen = contLen;
if (isGetMeta) {
info->type = TSDB_PROCESS_CREATE_VGROUP_GET_META;
}
mgmtSendCreateVgroupMsg(pVgroup, info);
}
//void mgmtSendCreateTableMsg(SMDCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle) {
// mTrace("table:%s, send create table msg, ahandle:%p", pCreate->tableId, ahandle);
// SRpcMsg rpcMsg = {
// .handle = ahandle,
// .pCont = pCreate,
// .contLen = htonl(pCreate->contLen),
// .code = 0,
// .msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
// };
// rpcSendRequest(tsMgmtDClientRpc, ipSet, &rpcMsg);
//}
//
void mgmtProcessCreateTable(SVgObj *pVgroup, SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
assert(pVgroup != NULL);
SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
int32_t sid = taosAllocateId(pVgroup->idPool); int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid < 0) { if (sid < 0) {
mTrace("table:%s, no enough sid in vgroup:%d, start to create a new vgroup", pCreate->tableId, pVgroup->vgId); mTrace("thandle:%p, no enough sid in vgroup:%d, start to create a new one", pMsg->thandle, pVgroup->vgId);
mgmtProcessCreateVgroup(pCreate, contLen, thandle, isGetMeta); mgmtCreateVgroup(pMsg);
return; return;
} }
int32_t code;
STableInfo *pTable; STableInfo *pTable;
SMDCreateTableMsg *pDCreate = NULL; SMDCreateTableMsg *pMDCreate = NULL;
if (pCreate->numOfColumns == 0) { if (pCreate->numOfColumns == 0) {
mTrace("table:%s, start to create child table, vgroup:%d sid:%d", pCreate->tableId, pVgroup->vgId, sid); mTrace("thandle:%p, create ctable:%s, vgroup:%d sid:%d ahandle:%p", pMsg->thandle, pCreate->tableId, pVgroup->vgId, sid, pMsg);
rpcRsp.code = mgmtCreateChildTable(pCreate, contLen, pVgroup, sid, &pDCreate, &pTable); code = mgmtCreateChildTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
} else { } else {
mTrace("table:%s, start to create normal table, vgroup:%d sid:%d", pCreate->tableId, pVgroup->vgId, sid); mTrace("thandle:%p, create ntable:%s, vgroup:%d sid:%d ahandle:%p", pMsg->thandle, pCreate->tableId, pVgroup->vgId, sid, pMsg);
rpcRsp.code = mgmtCreateNormalTable(pCreate, contLen, pVgroup, sid, &pDCreate, &pTable); code = mgmtCreateNormalTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
} }
if (rpcRsp.code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mTrace("table:%s, failed to create table in vgroup:%d sid:%d ", pCreate->tableId, pVgroup->vgId, sid); mTrace("thandle:%p, failed to create table:%s in vgroup:%d", pMsg->thandle, pCreate->tableId, pVgroup->vgId);
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
assert(pDCreate != NULL);
assert(pTable != NULL);
SProcessInfo *info = calloc(1, sizeof(SProcessInfo));
info->type = TSDB_PROCESS_CREATE_TABLE;
info->thandle = thandle;
info->ahandle = pTable;
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup); SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
if (isGetMeta) {
info->type = TSDB_PROCESS_CREATE_TABLE_GET_META;
}
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = info, .handle = pMsg,
.pCont = pCreate, .pCont = pCreate,
.contLen = htonl(pDCreate->contLen), .contLen = htonl(pMDCreate->contLen),
.code = 0, .code = 0,
.msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE .msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
}; };
mgmtSendMsgToDnode(&ipSet, &rpcMsg); mgmtSendMsgToDnode(&ipSet, &rpcMsg);
} }
int32_t mgmtCreateTable(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
SDbObj *pDb = mgmtGetDb(pCreate->db);
if (pDb == NULL) {
mError("table:%s, failed to create table, db not selected", pCreate->tableId);
return TSDB_CODE_DB_NOT_SELECTED;
}
STableInfo *pTable = mgmtGetTable(pCreate->tableId);
if (pTable != NULL) {
if (pCreate->igExists) {
mTrace("table:%s, table is already exist, think it success", pCreate->tableId);
return TSDB_CODE_SUCCESS;
} else {
mError("table:%s, failed to create table, table already exist", pCreate->tableId);
return TSDB_CODE_TABLE_ALREADY_EXIST;
}
}
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
assert(pAcct != NULL);
int32_t code = mgmtCheckTableLimit(pAcct, pCreate->numOfColumns);
if (code != TSDB_CODE_SUCCESS) {
mError("table:%s, failed to create table, table num exceed the limit", pCreate->tableId);
return code;
}
if (mgmtCheckExpired()) {
mError("table:%s, failed to create table, grant expired", pCreate->tableId);
return TSDB_CODE_GRANT_EXPIRED;
}
if (pCreate->numOfTags != 0) {
mTrace("table:%s, start to create super table, tags:%d columns:%d",
pCreate->tableId, pCreate->numOfTags, pCreate->numOfColumns);
return mgmtCreateSuperTable(pDb, pCreate);
}
code = mgmtCheckTimeSeries(pCreate->numOfColumns);
if (code != TSDB_CODE_SUCCESS) {
mError("table:%s, failed to create table, timeseries exceed the limit", pCreate->tableId);
return TSDB_CODE_SUCCESS;
}
SVgObj *pVgroup = mgmtGetAvailableVgroup(pDb);
if (pVgroup == NULL) {
mTrace("table:%s, no avaliable vgroup, start to create a new one", pCreate->tableId);
mgmtProcessCreateVgroup(pCreate, contLen, thandle, isGetMeta);
} else {
mTrace("table:%s, try to create table in vgroup:%d", pCreate->tableId, pVgroup->vgId);
mgmtProcessCreateTable(pVgroup, pCreate, contLen, thandle, isGetMeta);
}
return TSDB_CODE_ACTION_IN_PROGRESS;
}
int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore) { int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore) {
STableInfo *pTable = mgmtGetTable(tableId); STableInfo *pTable = mgmtGetTable(tableId);
if (pTable == NULL) { if (pTable == NULL) {
@ -547,114 +436,145 @@ SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable) {
} }
void mgmtSetTableDirty(STableInfo *pTable, bool isDirty) { void mgmtSetTableDirty(STableInfo *pTable, bool isDirty) {
// TODO: if dirty, delete from sdb
pTable->dirty = isDirty; pTable->dirty = isDirty;
} }
void mgmtProcessCreateTableMsg(SRpcMsg *rpcMsg) { void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SCMCreateTableMsg *pCreate = pMsg->pCont;
mTrace("thandle:%p, start to create table:%s", pMsg->thandle, pCreate->tableId);
SCMCreateTableMsg *pCreate = (SCMCreateTableMsg *) rpcMsg->pCont; if (mgmtCheckRedirect(pMsg->thandle)) {
pCreate->numOfColumns = htons(pCreate->numOfColumns); mError("thandle:%p, failed to create table:%s, need redirect", pMsg->thandle, pCreate->tableId);
pCreate->numOfTags = htons(pCreate->numOfTags);
pCreate->sqlLen = htons(pCreate->sqlLen);
SSchema *pSchema = (SSchema*) pCreate->schema;
for (int32_t i = 0; i < pCreate->numOfColumns + pCreate->numOfTags; ++i) {
pSchema->bytes = htons(pSchema->bytes);
pSchema->colId = i;
pSchema++;
}
if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) {
mError("table:%s, failed to create table, need redirect message", pCreate->tableId);
return; return;
} }
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); if (mgmtCheckExpired()) {
if (pUser == NULL) { mError("thandle:%p, failed to create table:%s, grant expired", pCreate->tableId);
mError("table:%s, failed to create table, invalid user", pCreate->tableId); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_GRANT_EXPIRED);
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return; return;
} }
if (!pUser->writeAuth) { if (!pMsg->pUser->writeAuth) {
mError("table:%s, failed to create table, no rights", pCreate->tableId); mError("thandle:%p, failed to create table:%s, no rights", pMsg->thandle, pCreate->tableId);
rpcRsp.code = TSDB_CODE_NO_RIGHTS; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
rpcSendResponse(&rpcRsp);
return; return;
} }
int32_t code = mgmtCreateTable(pCreate, rpcMsg->contLen, rpcMsg->handle, false); SAcctObj *pAcct = pMsg->pUser->pAcct;
if (code != TSDB_CODE_ACTION_IN_PROGRESS) { int32_t code = mgmtCheckTableLimit(pAcct, htons(pCreate->numOfColumns));
rpcRsp.code = code; if (code != TSDB_CODE_SUCCESS) {
rpcSendResponse(&rpcRsp); mError("thandle:%p, failed to create table:%s, exceed the limit", pMsg->thandle, pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
return;
}
SDbObj *pDb = mgmtGetDb(pCreate->db);
if (pDb == NULL) {
mError("thandle:%p, failed to create table:%s, db not selected", pMsg->thandle, pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_DB_NOT_SELECTED);
return;
}
STableInfo *pTable = mgmtGetTable(pCreate->tableId);
if (pTable != NULL) {
if (pCreate->igExists) {
mTrace("thandle:%p, table:%s is already exist", pMsg->thandle, pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SUCCESS);
return;
} else {
mError("thandle:%p, failed to create table:%s, table already exist", pMsg->thandle, pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_TABLE_ALREADY_EXIST);
return;
}
}
if (pCreate->numOfTags != 0) {
mTrace("thandle:%p, start to create super table:%s, tags:%d columns:%d",
pMsg->thandle, pCreate->tableId, pCreate->numOfTags, pCreate->numOfColumns);
code = mgmtCreateSuperTable(pDb, pCreate);
mgmtSendSimpleResp(pMsg->thandle, code);
return;
}
code = mgmtCheckTimeSeries(pCreate->numOfColumns);
if (code != TSDB_CODE_SUCCESS) {
mError("thandle:%p, failed to create table:%s, timeseries exceed the limit", pMsg->thandle, pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, code);
return;
}
SQueuedMsg *newMsg = malloc(sizeof(SQueuedMsg));
memcpy(newMsg, pMsg, sizeof(SQueuedMsg));
pMsg->pCont = NULL;
SVgObj *pVgroup = mgmtGetAvailableVgroup(pDb);
if (pVgroup == NULL) {
mTrace("thandle:%p, table:%s start to create a new vgroup", pMsg->thandle, pCreate->tableId);
mgmtCreateVgroup(pMsg);
} else {
mTrace("thandle:%p, create table:%s in vgroup:%d", pMsg->thandle, pCreate->tableId, pVgroup->vgId);
mgmtCreateTable(pVgroup, pMsg);
} }
} }
void mgmtProcessDropTableMsg(SRpcMsg *rpcMsg) { void mgmtProcessDropTableMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SCMDropTableMsg *pDrop = pMsg->pCont;
SCMDropTableMsg *pDrop = (SCMDropTableMsg *) rpcMsg->pCont;
if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) { if (mgmtCheckRedirect(pMsg->thandle)) {
mError("table:%s, failed to drop table, need redirect message", pDrop->tableId); mError("thandle:%p, failed to drop table:%s, need redirect message", pMsg->thandle, pDrop->tableId);
return; return;
} }
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
mError("table:%s, failed to drop table, invalid user", pDrop->tableId); mError("table:%s, failed to drop table, invalid user", pDrop->tableId);
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return; return;
} }
if (!pUser->writeAuth) { if (!pUser->writeAuth) {
mError("table:%s, failed to drop table, no rights", pDrop->tableId); mError("table:%s, failed to drop table, no rights", pDrop->tableId);
rpcRsp.code = TSDB_CODE_NO_RIGHTS; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
rpcSendResponse(&rpcRsp);
return; return;
} }
SDbObj *pDb = mgmtGetDbByTableId(pDrop->tableId); SDbObj *pDb = mgmtGetDbByTableId(pDrop->tableId);
if (pDb == NULL) { if (pDb == NULL) {
mError("table:%s, failed to drop table, db not selected", pDrop->tableId); mError("table:%s, failed to drop table, db not selected", pDrop->tableId);
rpcRsp.code = TSDB_CODE_DB_NOT_SELECTED; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_DB_NOT_SELECTED);
rpcSendResponse(&rpcRsp);
return; return;
} }
int32_t code = mgmtDropTable(pDb, pDrop->tableId, pDrop->igNotExists); int32_t code = mgmtDropTable(pDb, pDrop->tableId, pDrop->igNotExists);
if (code != TSDB_CODE_ACTION_IN_PROGRESS) { if (code != TSDB_CODE_ACTION_IN_PROGRESS) {
rpcRsp.code = code; mgmtSendSimpleResp(pMsg->thandle, code);
rpcSendResponse(&rpcRsp);
} }
} }
void mgmtProcessAlterTableMsg(SRpcMsg *rpcMsg) { void mgmtProcessAlterTableMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) {
if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) {
return; return;
} }
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return; return;
} }
SCMAlterTableMsg *pAlter = (SCMAlterTableMsg *) rpcMsg->pCont; SCMAlterTableMsg *pAlter = pMsg->pCont;
int32_t code;
if (!pUser->writeAuth) { if (!pUser->writeAuth) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} else { } else {
pAlter->type = htons(pAlter->type); pAlter->type = htons(pAlter->type);
pAlter->numOfCols = htons(pAlter->numOfCols); pAlter->numOfCols = htons(pAlter->numOfCols);
if (pAlter->numOfCols > 2) { if (pAlter->numOfCols > 2) {
mError("table:%s error numOfCols:%d in alter table", pAlter->tableId, pAlter->numOfCols); mError("table:%s error numOfCols:%d in alter table", pAlter->tableId, pAlter->numOfCols);
rpcRsp.code = TSDB_CODE_APP_ERROR; code = TSDB_CODE_APP_ERROR;
} else { } else {
SDbObj *pDb = mgmtGetDb(pAlter->db); SDbObj *pDb = mgmtGetDb(pAlter->db);
if (pDb) { if (pDb) {
@ -662,17 +582,17 @@ void mgmtProcessAlterTableMsg(SRpcMsg *rpcMsg) {
pAlter->schema[i].bytes = htons(pAlter->schema[i].bytes); pAlter->schema[i].bytes = htons(pAlter->schema[i].bytes);
} }
rpcRsp.code = mgmtAlterTable(pDb, pAlter); code = mgmtAlterTable(pDb, pAlter);
if (rpcRsp.code == 0) { if (code == 0) {
mLPrint("table:%s is altered by %s", pAlter->tableId, pUser->user); mLPrint("table:%s is altered by %s", pAlter->tableId, pUser->user);
} }
} else { } else {
rpcRsp.code = TSDB_CODE_DB_NOT_SELECTED; code = TSDB_CODE_DB_NOT_SELECTED;
} }
} }
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
} }
void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) { void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) {
@ -707,21 +627,14 @@ void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) {
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
void mgmtProcessTableMetaMsg(SQueuedMsg *pMsg) {
void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) { SCMTableInfoMsg *pInfo = pMsg->pCont;
SRpcMsg rpcRsp;
rpcRsp.handle = rpcMsg->handle;
rpcRsp.pCont = NULL;
rpcRsp.contLen = 0;
SCMTableInfoMsg *pInfo = rpcMsg->pCont;
pInfo->createFlag = htons(pInfo->createFlag); pInfo->createFlag = htons(pInfo->createFlag);
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle); SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle);
if (pUser == NULL) { if (pUser == NULL) {
mError("table:%s, failed to get table meta, invalid user", pInfo->tableId); mError("table:%s, failed to get table meta, invalid user", pInfo->tableId);
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return; return;
} }
@ -729,12 +642,11 @@ void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) {
if (pTable == NULL) { if (pTable == NULL) {
if (pInfo->createFlag != 1) { if (pInfo->createFlag != 1) {
mError("table:%s, failed to get table meta, table not exist", pInfo->tableId); mError("table:%s, failed to get table meta, table not exist", pInfo->tableId);
rpcRsp.code = TSDB_CODE_INVALID_TABLE; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
rpcSendResponse(&rpcRsp);
return; return;
} else { } else {
// on demand create table from super table if table does not exists // on demand create table from super table if table does not exists
if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) { if (mgmtCheckRedirect(pMsg->thandle)) {
mError("table:%s, failed to create table while get meta info, need redirect message", pInfo->tableId); mError("table:%s, failed to create table while get meta info, need redirect message", pInfo->tableId);
return; return;
} }
@ -743,8 +655,7 @@ void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) {
SCMCreateTableMsg *pCreateMsg = rpcMallocCont(contLen); SCMCreateTableMsg *pCreateMsg = rpcMallocCont(contLen);
if (pCreateMsg == NULL) { if (pCreateMsg == NULL) {
mError("table:%s, failed to create table while get meta info, no enough memory", pInfo->tableId); mError("table:%s, failed to create table while get meta info, no enough memory", pInfo->tableId);
rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
rpcSendResponse(&rpcRsp);
return; return;
} }
@ -752,41 +663,34 @@ void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) {
strcpy(pCreateMsg->tableId, pInfo->tableId); strcpy(pCreateMsg->tableId, pInfo->tableId);
mError("table:%s, start to create table while get meta info", pInfo->tableId); mError("table:%s, start to create table while get meta info", pInfo->tableId);
mgmtCreateTable(pCreateMsg, contLen, rpcMsg->handle, true); // mgmtCreateTable(pCreateMsg, contLen, pMsg->thandle, true);
} }
} else { } else {
mgmtProcessGetTableMeta(pTable, rpcMsg->handle); mgmtProcessGetTableMeta(pTable, pMsg->thandle);
} }
} }
void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg) { void mgmtProcessMultiTableMetaMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp;
rpcRsp.handle = rpcMsg->handle;
rpcRsp.pCont = NULL;
rpcRsp.contLen = 0;
SRpcConnInfo connInfo; SRpcConnInfo connInfo;
if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) { if (rpcGetConnInfo(pMsg->thandle, &connInfo) != 0) {
mError("conn:%p is already released while get mulit table meta", rpcMsg->handle); mError("conn:%p is already released while get mulit table meta", pMsg->thandle);
return; return;
} }
bool usePublicIp = (connInfo.serverIp == tsPublicIpInt); bool usePublicIp = (connInfo.serverIp == tsPublicIpInt);
SUserObj *pUser = mgmtGetUser(connInfo.user); SUserObj *pUser = mgmtGetUser(connInfo.user);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return; return;
} }
SCMMultiTableInfoMsg *pInfo = rpcMsg->pCont; SCMMultiTableInfoMsg *pInfo = pMsg->pCont;
pInfo->numOfTables = htonl(pInfo->numOfTables); pInfo->numOfTables = htonl(pInfo->numOfTables);
int32_t totalMallocLen = 4*1024*1024; // first malloc 4 MB, subsequent reallocation as twice int32_t totalMallocLen = 4*1024*1024; // first malloc 4 MB, subsequent reallocation as twice
SMultiTableMeta *pMultiMeta = rpcMallocCont(totalMallocLen); SMultiTableMeta *pMultiMeta = rpcMallocCont(totalMallocLen);
if (pMultiMeta == NULL) { if (pMultiMeta == NULL) {
rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
rpcSendResponse(&rpcRsp);
return; return;
} }
@ -823,29 +727,69 @@ void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg) {
} }
} }
SRpcMsg rpcRsp = {0};
rpcRsp.handle = pMsg->thandle;
rpcRsp.pCont = pMultiMeta; rpcRsp.pCont = pMultiMeta;
rpcRsp.contLen = pMultiMeta->contLen; rpcRsp.contLen = pMultiMeta->contLen;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
void mgmtProcessSuperTableMetaMsg(SRpcMsg *rpcMsg) { void mgmtProcessSuperTableMetaMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SCMSuperTableInfoMsg *pInfo = pMsg->pCont;
SCMSuperTableInfoMsg *pInfo = rpcMsg->pCont;
STableInfo *pTable = mgmtGetSuperTable(pInfo->tableId); STableInfo *pTable = mgmtGetSuperTable(pInfo->tableId);
if (pTable == NULL) { if (pTable == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_TABLE; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
rpcSendResponse(&rpcRsp);
return; return;
} }
SCMSuperTableInfoRsp *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable); SCMSuperTableInfoRsp *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable);
if (pRsp != NULL) { if (pRsp != NULL) {
int32_t msgLen = sizeof(SSuperTableObj) + htonl(pRsp->numOfDnodes) * sizeof(int32_t); int32_t msgLen = sizeof(SSuperTableObj) + htonl(pRsp->numOfDnodes) * sizeof(int32_t);
SRpcMsg rpcRsp = {0};
rpcRsp.handle = pMsg->thandle;
rpcRsp.pCont = pRsp; rpcRsp.pCont = pRsp;
rpcRsp.contLen = msgLen; rpcRsp.contLen = msgLen;
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} else { } else {
rpcRsp.code = TSDB_CODE_INVALID_TABLE; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
rpcSendResponse(&rpcRsp);
} }
} }
static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->handle == NULL) return;
SQueuedMsg *queueMsg = rpcMsg->handle;
queueMsg->received++;
STableInfo *pTable = queueMsg->ahandle;
mTrace("thandle:%p, create table:%s rsp received, ahandle:%p code:%d received:%d",
queueMsg->thandle, pTable->tableId, rpcMsg->handle, rpcMsg->code, queueMsg->received);
if (rpcMsg->code != TSDB_CODE_SUCCESS) {
mgmtSetTableDirty(pTable, true);
//sdbDeleteRow(tsVgroupSdb, pVgroup);
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
mError("table:%s, failed to create in dnode, code:%d, set it dirty", pTable->tableId, rpcMsg->code);
mgmtSetTableDirty(pTable, true);
} else {
mTrace("table:%s, created in dnode", pTable->tableId);
mgmtSetTableDirty(pTable, false);
if (queueMsg->msgType != TSDB_MSG_TYPE_CM_CREATE_TABLE) {
SQueuedMsg *newMsg = calloc(1, sizeof(SQueuedMsg));
newMsg->msgType = queueMsg->msgType;
newMsg->thandle = queueMsg->thandle;
newMsg->pDb = queueMsg->pDb;
newMsg->pUser = queueMsg->pUser;
newMsg->contLen = queueMsg->contLen;
newMsg->pCont = rpcMallocCont(newMsg->contLen);
memcpy(newMsg->pCont, queueMsg->pCont, newMsg->contLen);
mTrace("table:%s, start to process get meta", pTable->tableId);
mgmtAddToShellQueue(newMsg);
} else {
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
}
}
free(queueMsg);
}

102
src/mnode/src/mgmtUser.c
View File

@ -33,9 +33,9 @@ static int32_t mgmtUpdateUser(SUserObj *pUser);
static int32_t mgmtGetUserMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetUserMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCreateUserMsg(SRpcMsg *rpcMsg); static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessAlterUserMsg(SRpcMsg *rpcMsg); static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessDropUserMsg(SRpcMsg *rpcMsg); static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg);
static void *(*mgmtUserActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize); static void *(*mgmtUserActionFp[SDB_MAX_ACTION_TYPES])(void *row, char *str, int32_t size, int32_t *ssize);
static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize); static void *mgmtUserActionInsert(void *row, char *str, int32_t size, int32_t *ssize);
@ -337,52 +337,40 @@ SUserObj *mgmtGetUserFromConn(void *pConn) {
return NULL; return NULL;
} }
static void mgmtProcessCreateUserMsg(SRpcMsg *rpcMsg) { static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) return;
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle);
if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return;
}
int32_t code;
SUserObj *pUser = pMsg->pUser;
if (pUser->superAuth) { if (pUser->superAuth) {
SCMCreateUserMsg *pCreate = rpcMsg->pCont; SCMCreateUserMsg *pCreate = pMsg->pCont;
rpcRsp.code = mgmtCreateUser(pUser->pAcct, pCreate->user, pCreate->pass); code = mgmtCreateUser(pUser->pAcct, pCreate->user, pCreate->pass);
if (rpcRsp.code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
mLPrint("user:%s is created by %s", pCreate->user, pUser->user); mLPrint("user:%s is created by %s", pCreate->user, pUser->user);
} }
} else { } else {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
} }
static void mgmtProcessAlterUserMsg(SRpcMsg *rpcMsg) { static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) return;
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SUserObj *pOperUser = mgmtGetUserFromConn(rpcMsg->handle); int32_t code;
if (pOperUser == NULL) { SUserObj *pOperUser = pMsg->pUser;
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp); SCMAlterUserMsg *pAlter = pMsg->pCont;
return;
}
SCMAlterUserMsg *pAlter = rpcMsg->pCont;
SUserObj *pUser = mgmtGetUser(pAlter->user); SUserObj *pUser = mgmtGetUser(pAlter->user);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return; return;
} }
if (strcmp(pUser->user, "monitor") == 0 || (strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) { if (strcmp(pUser->user, "monitor") == 0 || (strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
rpcSendResponse(&rpcRsp);
return; return;
} }
@ -405,13 +393,13 @@ static void mgmtProcessAlterUserMsg(SRpcMsg *rpcMsg) {
if (hasRight) { if (hasRight) {
memset(pUser->pass, 0, sizeof(pUser->pass)); memset(pUser->pass, 0, sizeof(pUser->pass));
taosEncryptPass((uint8_t*)pAlter->pass, strlen(pAlter->pass), pUser->pass); taosEncryptPass((uint8_t*)pAlter->pass, strlen(pAlter->pass), pUser->pass);
rpcRsp.code = mgmtUpdateUser(pUser); code = mgmtUpdateUser(pUser);
mLPrint("user:%s password is altered by %s, code:%d", pAlter->user, pUser->user, rpcRsp.code); mLPrint("user:%s password is altered by %s, code:%d", pAlter->user, pUser->user, code);
} else { } else {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
@ -454,42 +442,34 @@ static void mgmtProcessAlterUserMsg(SRpcMsg *rpcMsg) {
pUser->writeAuth = 1; pUser->writeAuth = 1;
} }
rpcRsp.code = mgmtUpdateUser(pUser); code = mgmtUpdateUser(pUser);
mLPrint("user:%s privilege is altered by %s, code:%d", pAlter->user, pUser->user, rpcRsp.code); mLPrint("user:%s privilege is altered by %s, code:%d", pAlter->user, pUser->user, code);
} else { } else {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
rpcRsp.code = TSDB_CODE_NO_RIGHTS; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
rpcSendResponse(&rpcRsp);
} }
static void mgmtProcessDropUserMsg(SRpcMsg *rpcMsg) { static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; if (mgmtCheckRedirect(pMsg->thandle)) return;
if (mgmtCheckRedirect(rpcMsg->handle)) return;
SUserObj *pOperUser = mgmtGetUserFromConn(rpcMsg->handle); int32_t code;
if (pOperUser == NULL) { SUserObj *pOperUser = pMsg->pUser;
rpcRsp.code = TSDB_CODE_INVALID_USER;
rpcSendResponse(&rpcRsp);
return ;
}
SCMDropUserMsg *pDrop = rpcMsg->pCont; SCMDropUserMsg *pDrop = pMsg->pCont;
SUserObj *pUser = mgmtGetUser(pDrop->user); SUserObj *pUser = mgmtGetUser(pDrop->user);
if (pUser == NULL) { if (pUser == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_USER; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
rpcSendResponse(&rpcRsp);
return ; return ;
} }
if (strcmp(pUser->user, "monitor") == 0 || (strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) { if (strcmp(pUser->user, "monitor") == 0 || (strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
rpcSendResponse(&rpcRsp);
return ; return ;
} }
@ -511,13 +491,13 @@ static void mgmtProcessDropUserMsg(SRpcMsg *rpcMsg) {
} }
if (hasRight) { if (hasRight) {
rpcRsp.code = mgmtDropUser(pUser->pAcct, pDrop->user); code = mgmtDropUser(pUser->pAcct, pDrop->user);
if (rpcRsp.code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
mLPrint("user:%s is dropped by %s", pDrop->user, pUser->user); mLPrint("user:%s is dropped by %s", pDrop->user, pUser->user);
} }
} else { } else {
rpcRsp.code = TSDB_CODE_NO_RIGHTS; code = TSDB_CODE_NO_RIGHTS;
} }
rpcSendResponse(&rpcRsp); mgmtSendSimpleResp(pMsg->thandle, code);
} }

126
src/mnode/src/mgmtVgroup.c
View File

@ -24,6 +24,7 @@
#include "mgmtDb.h" #include "mgmtDb.h"
#include "mgmtDClient.h" #include "mgmtDClient.h"
#include "mgmtDnode.h" #include "mgmtDnode.h"
#include "mgmtProfile.h"
#include "mgmtShell.h" #include "mgmtShell.h"
#include "mgmtTable.h" #include "mgmtTable.h"
#include "mgmtVgroup.h" #include "mgmtVgroup.h"
@ -42,6 +43,9 @@ static void *mgmtVgroupActionDestroy(void *row, char *str, int32_t size, int32_t
static int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg);
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
static void mgmtVgroupActionInit() { static void mgmtVgroupActionInit() {
SVgObj tObj; SVgObj tObj;
@ -114,6 +118,7 @@ int32_t mgmtInitVgroups() {
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta); mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VGROUP, mgmtRetrieveVgroups); mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VGROUP, mgmtRetrieveVgroups);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP, mgmtProcessCreateVnodeRsp);
mTrace("vgroup is initialized"); mTrace("vgroup is initialized");
return 0; return 0;
@ -123,19 +128,6 @@ SVgObj *mgmtGetVgroup(int32_t vgId) {
return (SVgObj *)sdbGetRow(tsVgroupSdb, &vgId); return (SVgObj *)sdbGetRow(tsVgroupSdb, &vgId);
} }
int32_t mgmtAllocateSid(SDbObj *pDb, SVgObj *pVgroup) {
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid < 0) {
mWarn("table:%s, vgroup:%d run out of ID, num:%d", pDb->name, pVgroup->vgId, taosIdPoolNumOfUsed(pVgroup->idPool));
pDb->vgStatus = TSDB_VG_STATUS_IN_PROGRESS;
mgmtCreateVgroup(pDb);
terrno = TSDB_CODE_ACTION_IN_PROGRESS;
}
terrno = 0;
return sid;
}
/* /*
* TODO: check if there is enough sids * TODO: check if there is enough sids
*/ */
@ -155,21 +147,25 @@ void mgmtProcessVgTimer(void *handle, void *tmrId) {
pDb->vgTimer = NULL; pDb->vgTimer = NULL;
} }
SVgObj *mgmtCreateVgroup(SDbObj *pDb) { void mgmtCreateVgroup(SQueuedMsg *pMsg) {
SDbObj *pDb = pMsg->pDb;
if (pDb == NULL) {
mError("thandle:%p, failed to create vgroup, db not found", pMsg->thandle);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_DB);
return;
}
SVgObj *pVgroup = (SVgObj *)calloc(sizeof(SVgObj), 1); SVgObj *pVgroup = (SVgObj *)calloc(sizeof(SVgObj), 1);
strcpy(pVgroup->dbName, pDb->name); strcpy(pVgroup->dbName, pDb->name);
pVgroup->numOfVnodes = pDb->cfg.replications; pVgroup->numOfVnodes = pDb->cfg.replications;
pVgroup->createdTime = taosGetTimestampMs();
// based on load balance, create a new one
if (mgmtAllocVnodes(pVgroup) != 0) { if (mgmtAllocVnodes(pVgroup) != 0) {
mError("db:%s, no enough free dnode to alloc %d vnodes", pDb->name, pVgroup->numOfVnodes); mError("thandle:%p, db:%s no enough dnode to alloc %d vnodes", pMsg->thandle, pDb->name, pVgroup->numOfVnodes);
free(pVgroup); free(pVgroup);
pDb->vgStatus = TSDB_VG_STATUS_FULL; mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_ENOUGH_DNODES);
taosTmrReset(mgmtProcessVgTimer, 5000, pDb, tsMgmtTmr, &pDb->vgTimer); return;
return NULL;
} }
pVgroup->createdTime = taosGetTimestampMs();
pVgroup->tableList = (STableInfo **) calloc(sizeof(STableInfo *), pDb->cfg.maxSessions); pVgroup->tableList = (STableInfo **) calloc(sizeof(STableInfo *), pDb->cfg.maxSessions);
pVgroup->numOfTables = 0; pVgroup->numOfTables = 0;
pVgroup->idPool = taosInitIdPool(pDb->cfg.maxSessions); pVgroup->idPool = taosInitIdPool(pDb->cfg.maxSessions);
@ -179,11 +175,14 @@ SVgObj *mgmtCreateVgroup(SDbObj *pDb) {
sdbInsertRow(tsVgroupSdb, pVgroup, 0); sdbInsertRow(tsVgroupSdb, pVgroup, 0);
mTrace("vgroup:%d, vgroup is created, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes); mPrint("thandle:%p, vgroup:%d is created in mnode, db:%s replica:%d", pMsg->thandle, pVgroup->vgId, pDb->name,
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) pVgroup->numOfVnodes);
mTrace("vgroup:%d, dnode:%s vnode:%d is created", pVgroup->vgId, taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode); for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
mPrint("thandle:%p, vgroup:%d, dnode:%s vnode:%d", pMsg->thandle, pVgroup->vgId,
taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode);
}
return pVgroup; mgmtSendCreateVgroupMsg(pVgroup, pMsg);
} }
int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup) { int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup) {
@ -514,13 +513,13 @@ SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
SDbObj *pDb = mgmtGetDb(pVgroup->dbName); SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
if (pDb == NULL) return NULL; if (pDb == NULL) return NULL;
SMDCreateVnodeMsg *pVPeers = rpcMallocCont(sizeof(SMDCreateVnodeMsg)); SMDCreateVnodeMsg *pVnode = rpcMallocCont(sizeof(SMDCreateVnodeMsg));
if (pVPeers == NULL) return NULL; if (pVnode == NULL) return NULL;
pVPeers->vnode = htonl(vnode); pVnode->vnode = htonl(vnode);
pVPeers->cfg = pDb->cfg; pVnode->cfg = pDb->cfg;
SVnodeCfg *pCfg = &pVPeers->cfg; SVnodeCfg *pCfg = &pVnode->cfg;
pCfg->vgId = htonl(pVgroup->vgId); pCfg->vgId = htonl(pVgroup->vgId);
pCfg->maxSessions = htonl(pCfg->maxSessions); pCfg->maxSessions = htonl(pCfg->maxSessions);
pCfg->cacheBlockSize = htonl(pCfg->cacheBlockSize); pCfg->cacheBlockSize = htonl(pCfg->cacheBlockSize);
@ -534,13 +533,14 @@ SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
pCfg->replications = (char) pVgroup->numOfVnodes; pCfg->replications = (char) pVgroup->numOfVnodes;
pCfg->rowsInFileBlock = htonl(pCfg->rowsInFileBlock); pCfg->rowsInFileBlock = htonl(pCfg->rowsInFileBlock);
SVnodeDesc *vpeerDesc = pVPeers->vpeerDesc; SVnodeDesc *vpeerDesc = pVnode->vpeerDesc;
for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) { for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {
vpeerDesc[j].ip = htonl(pVgroup->vnodeGid[j].ip); vpeerDesc[j].vgId = htonl(pVgroup->vgId);
vpeerDesc[j].ip = htonl(pVgroup->vnodeGid[j].ip);
vpeerDesc[j].vnode = htonl(pVgroup->vnodeGid[j].vnode); vpeerDesc[j].vnode = htonl(pVgroup->vnodeGid[j].vnode);
} }
return pVPeers; return pVnode;
} }
SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode) { SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode) {
@ -558,7 +558,11 @@ SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode) {
} }
SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) { SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
SRpcIpSet ipSet = {.numOfIps = pVgroup->numOfVnodes, .inUse = 0, .port = tsMnodeDnodePort + 1}; SRpcIpSet ipSet = {
.numOfIps = pVgroup->numOfVnodes,
.inUse = 0,
.port = tsMnodeDnodePort
};
for (int i = 0; i < pVgroup->numOfVnodes; ++i) { for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
ipSet.ip[i] = pVgroup->vnodeGid[i].ip; ipSet.ip[i] = pVgroup->vnodeGid[i].ip;
} }
@ -566,7 +570,12 @@ SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
} }
SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip) { SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip) {
SRpcIpSet ipSet = {.ip[0] = ip, .numOfIps = 1, .inUse = 0, .port = tsMnodeDnodePort + 1}; SRpcIpSet ipSet = {
.ip[0] = ip,
.numOfIps = 1,
.inUse = 0,
.port = tsMnodeDnodePort
};
return ipSet; return ipSet;
} }
@ -574,19 +583,54 @@ void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, vo
mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, vnode, ahandle); mTrace("vgroup:%d, send create vnode:%d msg, ahandle:%p", pVgroup->vgId, vnode, ahandle);
SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup, vnode); SMDCreateVnodeMsg *pCreate = mgmtBuildCreateVnodeMsg(pVgroup, vnode);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = ahandle, .handle = ahandle,
.pCont = pCreate, .pCont = pCreate,
.contLen = pCreate ? sizeof(SMDCreateVnodeMsg) : 0, .contLen = pCreate ? sizeof(SMDCreateVnodeMsg) : 0,
.code = 0, .code = 0,
.msgType = TSDB_MSG_TYPE_MD_CREATE_VNODE .msgType = TSDB_MSG_TYPE_MD_CREATE_VNODE
}; };
mgmtSendMsgToDnode(ipSet, &rpcMsg); mgmtSendMsgToDnode(ipSet, &rpcMsg);
} }
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) { void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
mTrace("vgroup:%d, send create all vnodes msg, handle:%p", pVgroup->vgId, ahandle); mTrace("send create vgroup:%d msg, ahandle:%p", pVgroup->vgId, ahandle);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) { for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip); SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle); mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
} }
}
static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->handle == NULL) return;
SQueuedMsg *queueMsg = rpcMsg->handle;
queueMsg->received++;
if (rpcMsg->code == TSDB_CODE_SUCCESS) {
queueMsg->code = rpcMsg->code;
queueMsg->successed++;
}
SVgObj *pVgroup = queueMsg->ahandle;
mTrace("thandle:%p, vgroup:%d create vnode rsp received, ahandle:%p code:%d received:%d successed:%d expected:%d",
queueMsg->thandle, pVgroup->vgId, rpcMsg->handle, rpcMsg->code, queueMsg->received, queueMsg->successed,
queueMsg->expected);
if (queueMsg->received != queueMsg->expected) return;
if (queueMsg->received == queueMsg->successed) {
SQueuedMsg *newMsg = calloc(1, sizeof(SQueuedMsg));
newMsg->msgType = queueMsg->msgType;
newMsg->thandle = queueMsg->thandle;
newMsg->pDb = queueMsg->pDb;
newMsg->pUser = queueMsg->pUser;
newMsg->contLen = queueMsg->contLen;
newMsg->pCont = rpcMallocCont(newMsg->contLen);
memcpy(newMsg->pCont, queueMsg->pCont, newMsg->contLen);
mgmtAddToShellQueue(newMsg);
} else {
sdbDeleteRow(tsVgroupSdb, pVgroup);
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
}
free(queueMsg);
} }