Skip to content

T
TDengine

taosdata / TDengine
11 个月 前同步成功

通知 1179
Star 22014
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 f54cdb02 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #12176 from taosdata/feature/dnode 

fix: alter db replications
上级 4d99b003 b38f1802
隐藏空白更改
内联 并排
Showing with 403 addition and 139 deletion
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ f54cdb02
......@@ -293,14 +293,10 @@ void vmInitMsgHandle(SMgmtWrapper *pWrapper) {
dmSetMsgHandle(pWrapper, TDMT_VND_TASK_MERGE_EXEC, vmProcessMergeMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_TASK_WRITE_EXEC, vmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_STREAM_TRIGGER, vmProcessFetchMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_ALTER_VNODE, vmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_COMPACT_VNODE, vmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
// dmSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
// dmSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
// sync integration
dmSetMsgHandle(pWrapper, TDMT_VND_SYNC_TIMEOUT, (NodeMsgFp)vmProcessSyncMsg, DEFAULT_HANDLE);
......
source/dnode/mnode/impl/inc/mndAuth.h
浏览文件 @ f54cdb02
......@@ -31,6 +31,7 @@ int32_t mndCheckDropUserAuth(SUserObj *pOperUser);
int32_t mndCheckNodeAuth(SUserObj *pOperUser);
int32_t mndCheckFuncAuth(SUserObj *pOperUser);
int32_t mndCheckTransAuth(SUserObj *pOperUser);
int32_t mndCheckCreateDbAuth(SUserObj *pOperUser);
int32_t mndCheckAlterDropCompactDbAuth(SUserObj *pOperUser, SDbObj *pDb);
......
source/dnode/mnode/impl/inc/mndVgroup.h
浏览文件 @ f54cdb02
......@@ -27,10 +27,14 @@ void mndCleanupVgroup(SMnode *pMnode);
SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId);
void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup);
SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
SEpSet mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup);
int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId);
int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
SArray *mndBuildDnodesArray(SMnode *pMnode);
int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray);
int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2);
void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
void *mndBuildAlterVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
......
source/dnode/mnode/impl/src/mndAuth.c
浏览文件 @ f54cdb02
......@@ -74,10 +74,7 @@ static int32_t mndProcessAuthReq(SNodeMsg *pReq) {
}
int32_t mndCheckCreateUserAuth(SUserObj *pOperUser) {
if (pOperUser->superUser) {
return 0;
}
if (pOperUser->superUser) return 0;
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
......@@ -118,28 +115,25 @@ int32_t mndCheckAlterUserAuth(SUserObj *pOperUser, SUserObj *pUser, SDbObj *pDb,
}
int32_t mndCheckDropUserAuth(SUserObj *pOperUser) {
if (pOperUser->superUser) {
return 0;
}
if (pOperUser->superUser) return 0;
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
int32_t mndCheckNodeAuth(SUserObj *pOperUser) {
if (pOperUser->superUser) {
return 0;
}
if (pOperUser->superUser) return 0;
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
int32_t mndCheckFuncAuth(SUserObj *pOperUser) {
if (pOperUser->superUser) {
return 0;
}
if (pOperUser->superUser) return 0;
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
int32_t mndCheckTransAuth(SUserObj *pOperUser) {
if (pOperUser->superUser) return 0;
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ f54cdb02
......@@ -261,6 +261,104 @@ void mndReleaseDb(SMnode *pMnode, SDbObj *pDb) {
sdbRelease(pSdb, pDb);
}
static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
bool isRedo) {
STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_VNODE;
action.acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED;
if (isRedo) {
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
} else {
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
}
return 0;
}
static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
bool isRedo) {
STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildAlterVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_VND_ALTER_VNODE;
if (isRedo) {
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
} else {
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
}
return 0;
}
static int32_t mndAddDropVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
bool isRedo) {
STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
if (isRedo) {
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
} else {
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
}
return 0;
}
static int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
char *pos = strstr(dbName, TS_PATH_DELIMITER);
if (pos == NULL) {
......@@ -278,6 +376,8 @@ static int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
}
static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
if (pCfg->numOfVgroups < TSDB_MIN_VNODES_PER_DB || pCfg->numOfVgroups > TSDB_MAX_VNODES_PER_DB) return -1;
if (pCfg->numOfStables < TSDB_DB_STREAM_MODE_OFF || pCfg->numOfStables > TSDB_DB_STREAM_MODE_ON) return -1;
if (pCfg->buffer < TSDB_MIN_BUFFER_PER_VNODE || pCfg->buffer > TSDB_MAX_BUFFER_PER_VNODE) return -1;
......@@ -299,10 +399,12 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
if (pCfg->compression < TSDB_MIN_COMP_LEVEL || pCfg->compression > TSDB_MAX_COMP_LEVEL) return -1;
if (pCfg->replications < TSDB_MIN_DB_REPLICA || pCfg->replications > TSDB_MAX_DB_REPLICA) return -1;
if (pCfg->replications > mndGetDnodeSize(pMnode)) return -1;
if (pCfg->replications != 1 && pCfg->replications != 3) return -1;
if (pCfg->strict < TSDB_DB_STRICT_OFF || pCfg->strict > TSDB_DB_STRICT_ON) return -1;
if (pCfg->strict > pCfg->replications) return -1;
if (pCfg->cacheLastRow < TSDB_MIN_DB_CACHE_LAST_ROW || pCfg->cacheLastRow > TSDB_MAX_DB_CACHE_LAST_ROW) return -1;
if (pCfg->hashMethod != 1) return -1;
terrno = 0;
return TSDB_CODE_SUCCESS;
}
......@@ -381,24 +483,8 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
SVgObj *pVgroup = pVgroups + vg;
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
STransAction action = {0};
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_CREATE_VNODE;
action.acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
return -1;
}
}
......@@ -412,24 +498,8 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
SVgObj *pVgroup = pVgroups + vg;
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
STransAction action = {0};
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, false) != 0) {
return -1;
}
}
......@@ -482,7 +552,6 @@ static int32_t mndCreateDb(SMnode *pMnode, SNodeMsg *pReq, SCreateDbReq *pCreate
}
if (mndCheckDbCfg(pMnode, &dbObj.cfg) != 0) {
terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
mError("db:%s, failed to create since %s", pCreate->db, terrstr());
return -1;
}
......@@ -570,37 +639,37 @@ _OVER:
static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
terrno = TSDB_CODE_MND_DB_OPTION_UNCHANGED;
if (pAlter->buffer >= 0 && pAlter->buffer != pDb->cfg.buffer) {
if (pAlter->buffer > 0 && pAlter->buffer != pDb->cfg.buffer) {
pDb->cfg.buffer = pAlter->buffer;
terrno = 0;
}
if (pAlter->pages >= 0 && pAlter->pages != pDb->cfg.pages) {
if (pAlter->pages > 0 && pAlter->pages != pDb->cfg.pages) {
pDb->cfg.pages = pAlter->pages;
terrno = 0;
}
if (pAlter->pageSize >= 0 && pAlter->pageSize != pDb->cfg.pageSize) {
if (pAlter->pageSize > 0 && pAlter->pageSize != pDb->cfg.pageSize) {
pDb->cfg.pageSize = pAlter->pageSize;
terrno = 0;
}
if (pAlter->daysPerFile >= 0 && pAlter->daysPerFile != pDb->cfg.daysPerFile) {
if (pAlter->daysPerFile > 0 && pAlter->daysPerFile != pDb->cfg.daysPerFile) {
pDb->cfg.daysPerFile = pAlter->daysPerFile;
terrno = 0;
}
if (pAlter->daysToKeep0 >= 0 && pAlter->daysToKeep0 != pDb->cfg.daysToKeep0) {
if (pAlter->daysToKeep0 > 0 && pAlter->daysToKeep0 != pDb->cfg.daysToKeep0) {
pDb->cfg.daysToKeep0 = pAlter->daysToKeep0;
terrno = 0;
}
if (pAlter->daysToKeep1 >= 0 && pAlter->daysToKeep1 != pDb->cfg.daysToKeep1) {
if (pAlter->daysToKeep1 > 0 && pAlter->daysToKeep1 != pDb->cfg.daysToKeep1) {
pDb->cfg.daysToKeep1 = pAlter->daysToKeep1;
terrno = 0;
}
if (pAlter->daysToKeep2 >= 0 && pAlter->daysToKeep2 != pDb->cfg.daysToKeep2) {
if (pAlter->daysToKeep2 > 0 && pAlter->daysToKeep2 != pDb->cfg.daysToKeep2) {
pDb->cfg.daysToKeep2 = pAlter->daysToKeep2;
terrno = 0;
}
......@@ -625,8 +694,9 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
terrno = 0;
}
if (pAlter->replications >= 0 && pAlter->replications != pDb->cfg.replications) {
if (pAlter->replications > 0 && pAlter->replications != pDb->cfg.replications) {
pDb->cfg.replications = pAlter->replications;
pDb->vgVersion++;
terrno = 0;
}
......@@ -651,35 +721,57 @@ static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *p
return 0;
}
static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
STransAction action = {0};
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildAlterVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
if (pVgroup->replica <= 0 || pVgroup->replica == pDb->cfg.replications) {
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
return -1;
}
}
} else {
SVgObj newVgroup = {0};
memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
if (newVgroup.replica < pDb->cfg.replications) {
mInfo("db:%s, vgId:%d, will add 2 vnodes, vn:0 dnode:%d", pVgroup->dbName, pVgroup->vgId,
pVgroup->vnodeGid[0].dnodeId);
if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) {
mError("db:%s, failed to add vnode to vgId:%d since %s", pDb->name, newVgroup.vgId, terrstr());
return -1;
}
newVgroup.replica = pDb->cfg.replications;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[0], true) != 0) return -1;
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
} else {
mInfo("db:%s, vgId:%d, will remove 2 vnodes", pVgroup->dbName, pVgroup->vgId);
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_VND_ALTER_VNODE;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
SVnodeGid del1 = {0};
SVnodeGid del2 = {0};
if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del1, &del2) != 0) {
mError("db:%s, failed to remove vnode from vgId:%d since %s", pDb->name, newVgroup.vgId, terrstr());
return -1;
}
newVgroup.replica = pDb->cfg.replications;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[0], true) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
}
SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
if (pVgRaw == NULL) return -1;
if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
if (sdbSetRawStatus(pVgRaw, SDB_STATUS_READY) != 0) return -1;
}
return 0;
}
static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
SArray *pArray = mndBuildDnodesArray(pMnode);
while (1) {
SVgObj *pVgroup = NULL;
......@@ -687,9 +779,10 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
if (pIter == NULL) break;
if (pVgroup->dbUid == pNew->uid) {
if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup) != 0) {
if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup, pArray) != 0) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup);
taosArrayDestroy(pArray);
return -1;
}
}
......@@ -697,6 +790,7 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
sdbRelease(pSdb, pVgroup);
}
taosArrayDestroy(pArray);
return 0;
}
......@@ -726,6 +820,7 @@ static int32_t mndProcessAlterDbReq(SNodeMsg *pReq) {
SDbObj *pDb = NULL;
SUserObj *pUser = NULL;
SAlterDbReq alterReq = {0};
SDbObj dbObj = {0};
if (tDeserializeSAlterDbReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &alterReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
......@@ -749,15 +844,17 @@ static int32_t mndProcessAlterDbReq(SNodeMsg *pReq) {
goto _OVER;
}
SDbObj dbObj = {0};
memcpy(&dbObj, pDb, sizeof(SDbObj));
dbObj.cfg.numOfRetensions = 0;
dbObj.cfg.pRetensions = NULL;
if (dbObj.cfg.pRetensions != NULL) {
dbObj.cfg.pRetensions = taosArrayDup(pDb->cfg.pRetensions);
if (dbObj.cfg.pRetensions == NULL) goto _OVER;
}
code = mndSetDbCfgFromAlterDbReq(&dbObj, &alterReq);
if (code != 0) {
goto _OVER;
}
if (code != 0) goto _OVER;
code = mndCheckDbCfg(pMnode, &dbObj.cfg);
if (code != 0) goto _OVER;
dbObj.cfgVersion++;
dbObj.updateTime = taosGetTimestampMs();
......@@ -771,6 +868,7 @@ _OVER:
mndReleaseDb(pMnode, pDb);
mndReleaseUser(pMnode, pUser);
taosArrayDestroy(dbObj.cfg.pRetensions);
return code;
}
......@@ -899,24 +997,8 @@ static int32_t mndSetDropDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD
static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
STransAction action = {0};
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
return -1;
}
}
......@@ -988,6 +1070,17 @@ static int32_t mndDropDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb) {
/*if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;*/
if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
SUserObj *pUser = mndAcquireUser(pMnode, pDb->createUser);
if (pUser != NULL) {
pUser->authVersion++;
SSdbRaw *pCommitRaw = mndUserActionEncode(pUser);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
goto _OVER;
}
sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
}
int32_t rspLen = 0;
void *pRsp = NULL;
if (mndBuildDropDbRsp(pDb, &rspLen, &pRsp, false) < 0) goto _OVER;
......
source/dnode/mnode/impl/src/mndSma.c
浏览文件 @ f54cdb02
......@@ -21,7 +21,7 @@
#include "mndInfoSchema.h"
#include "mndMnode.h"
#include "mndShow.h"
#include "mndStb.c"
#include "mndStb.h"
#include "mndStream.h"
#include "mndTrans.h"
#include "mndUser.h"
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ f54cdb02
......@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "mndStb.h"
#include "mndAuth.h"
#include "mndDb.h"
......
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @ f54cdb02
......@@ -1336,8 +1336,7 @@ static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
goto _OVER;
}
if (!pUser->superUser) {
terrno = TSDB_CODE_MND_NO_RIGHTS;
if (mndCheckTransAuth(pUser) != 0) {
goto _OVER;
}
......
source/dnode/mnode/impl/src/mndUser.c
浏览文件 @ f54cdb02
......@@ -106,6 +106,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
SDB_SET_INT64(pRaw, dataPos, pUser->createdTime, _OVER)
SDB_SET_INT64(pRaw, dataPos, pUser->updateTime, _OVER)
SDB_SET_INT8(pRaw, dataPos, pUser->superUser, _OVER)
SDB_SET_INT32(pRaw, dataPos, pUser->authVersion, _OVER)
SDB_SET_INT32(pRaw, dataPos, numOfReadDbs, _OVER)
SDB_SET_INT32(pRaw, dataPos, numOfWriteDbs, _OVER)
......@@ -161,6 +162,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT64(pRaw, dataPos, &pUser->createdTime, _OVER)
SDB_GET_INT64(pRaw, dataPos, &pUser->updateTime, _OVER)
SDB_GET_INT8(pRaw, dataPos, &pUser->superUser, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pUser->authVersion, _OVER)
int32_t numOfReadDbs = 0;
int32_t numOfWriteDbs = 0;
......@@ -588,7 +590,7 @@ _OVER:
return code;
}
static int32_t mndSetUserAuthRsp(SMnode *pMnode, SUserObj *pUser, SGetUserAuthRsp *pRsp) {
static int32_t mndSetUserAuthRsp(SMnode *pMnode, SUserObj *pUser, SGetUserAuthRsp *pRsp) {
memcpy(pRsp->user, pUser->user, TSDB_USER_LEN);
pRsp->superAuth = pUser->superUser;
pRsp->version = pUser->authVersion;
......@@ -601,7 +603,7 @@ static int32_t mndSetUserAuthRsp(SMnode *pMnode, SUserObj *pUser, SGetUser
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
while (1) {
......@@ -659,7 +661,7 @@ static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
code = 0;
_OVER:
mndReleaseUser(pMnode, pUser);
tFreeSGetUserAuthRsp(&authRsp);
......@@ -711,7 +713,8 @@ static void mndCancelGetNextUser(SMnode *pMnode, void *pIter) {
sdbCancelFetch(pSdb, pIter);
}
int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp, int32_t *pRspLen) {
int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp,
int32_t *pRspLen) {
SUserAuthBatchRsp batchRsp = {0};
batchRsp.pArray = taosArrayInit(numOfUses, sizeof(SGetUserAuthRsp));
if (batchRsp.pArray == NULL) {
......@@ -731,7 +734,7 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
mndReleaseUser(pMnode, pUser);
continue;
}
SGetUserAuthRsp rsp = {0};
code = mndSetUserAuthRsp(pMnode, pUser, &rsp);
if (code) {
......@@ -740,7 +743,6 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
goto _OVER;
}
taosArrayPush(batchRsp.pArray, &rsp);
mndReleaseUser(pMnode, pUser);
}
......@@ -748,7 +750,7 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
if (taosArrayGetSize(batchRsp.pArray) <= 0) {
*ppRsp = NULL;
*pRspLen = 0;
tFreeSUserAuthBatchRsp(&batchRsp);
return 0;
}
......@@ -772,10 +774,7 @@ _OVER:
*ppRsp = NULL;
*pRspLen = 0;
tFreeSUserAuthBatchRsp(&batchRsp);
return code;
}
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ f54cdb02
......@@ -370,7 +370,7 @@ static bool mndBuildDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2
return true;
}
static SArray *mndBuildDnodesArray(SMnode *pMnode) {
SArray *mndBuildDnodesArray(SMnode *pMnode) {
SSdb *pSdb = pMnode->pSdb;
int32_t numOfDnodes = mndGetDnodeSize(pMnode);
......@@ -421,7 +421,7 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
}
mDebug("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
mInfo("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
pDnode->numOfVnodes++;
}
......@@ -440,12 +440,10 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
}
pArray = mndBuildDnodesArray(pMnode);
if (pArray == NULL) {
goto _OVER;
}
if (pArray == NULL) goto _OVER;
mDebug("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);
mInfo("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);
int32_t allocedVgroups = 0;
int32_t maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
......@@ -483,7 +481,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
*ppVgroups = pVgroups;
code = 0;
mDebug("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
mInfo("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
_OVER:
if (code != 0) taosMemoryFree(pVgroups);
......@@ -491,6 +489,88 @@ _OVER:
return code;
}
int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SDnodeObj *pDnode = taosArrayGet(pArray, i);
mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
}
int32_t maxPos = 1;
for (int32_t d = 0; d < taosArrayGetSize(pArray); ++d) {
SDnodeObj *pDnode = taosArrayGet(pArray, d);
bool used = false;
for (int32_t vn = 0; vn < maxPos; ++vn) {
if (pDnode->id == pVgroup->vnodeGid[vn].dnodeId) {
used = true;
break;
}
}
if (used) continue;
if (pDnode == NULL || pDnode->numOfVnodes > pDnode->numOfSupportVnodes) {
terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
return -1;
}
SVnodeGid *pVgid = &pVgroup->vnodeGid[maxPos];
pVgid->dnodeId = pDnode->id;
pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
pDnode->numOfVnodes++;
mInfo("db:%s, vgId:%d, vn:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId);
maxPos++;
if (maxPos == 3) return 0;
}
terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
return -1;
}
int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2) {
int32_t removedNum = 0;
taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SDnodeObj *pDnode = taosArrayGet(pArray, i);
mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
}
for (int32_t d = taosArrayGetSize(pArray) - 1; d >= 0; --d) {
SDnodeObj *pDnode = taosArrayGet(pArray, d);
for (int32_t vn = 0; vn < TSDB_MAX_REPLICA; ++vn) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
if (pVgid->dnodeId == pDnode->id) {
if (removedNum == 0) *del1 = *pVgid;
if (removedNum == 1) *del2 = *pVgid;
mInfo("db:%s, vgId:%d, vn:%d dnode:%d is removed", pVgroup->dbName, pVgroup->vgId, vn, pVgid->dnodeId);
memset(pVgid, 0, sizeof(SVnodeGid));
removedNum++;
pDnode->numOfVnodes--;
if (removedNum == 2) goto _OVER;
}
}
}
_OVER:
if (removedNum != 2) return -1;
for (int32_t vn = 1; vn < TSDB_MAX_REPLICA; ++vn) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
if (pVgid->dnodeId != 0) {
memcpy(&pVgroup->vnodeGid[0], pVgid, sizeof(SVnodeGid));
memset(pVgid, 0, sizeof(SVnodeGid));
}
}
mInfo("db:%s, vgId:%d, dnode:%d is keeped", pVgroup->dbName, pVgroup->vgId, pVgroup->vnodeGid[0].dnodeId);
return 0;
}
SEpSet mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup) {
SEpSet epset = {0};
......
source/dnode/mnode/impl/test/db/db.cpp
浏览文件 @ f54cdb02
......@@ -73,14 +73,19 @@ TEST_F(MndTestDb, 02_Create_Alter_Drop_Db) {
{
SAlterDbReq alterdbReq = {0};
strcpy(alterdbReq.db, "1.d1");
alterdbReq.buffer = 12;
alterdbReq.daysToKeep0 = 300;
alterdbReq.daysToKeep1 = 400;
alterdbReq.daysToKeep2 = 500;
alterdbReq.pageSize = -1;
alterdbReq.pages = -1;
alterdbReq.daysPerFile = -1;
alterdbReq.daysToKeep0 = -1;
alterdbReq.daysToKeep1 = -1;
alterdbReq.daysToKeep2 = -1;
alterdbReq.fsyncPeriod = 4000;
alterdbReq.walLevel = 2;
alterdbReq.strict = 2;
alterdbReq.strict = 1;
alterdbReq.cacheLastRow = 1;
alterdbReq.replications = 1;
int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
void* pReq = rpcMallocCont(contLen);
......
source/dnode/mnode/impl/test/dnode/mdnode.cpp
浏览文件 @ f54cdb02
......@@ -266,4 +266,96 @@ TEST_F(MndTestDnode, 05_Create_Drop_Restart_Dnode) {
taosMsleep(1300);
test.SendShowReq(TSDB_MGMT_TABLE_DNODE, "dnodes", "");
EXPECT_EQ(test.GetShowRows(), 4);
// alter replica
#if 0
{
SCreateDbReq createReq = {0};
strcpy(createReq.db, "1.d2");
createReq.numOfVgroups = 2;
createReq.buffer = -1;
createReq.pageSize = -1;
createReq.pages = -1;
createReq.daysPerFile = 1000;
createReq.daysToKeep0 = 3650;
createReq.daysToKeep1 = 3650;
createReq.daysToKeep2 = 3650;
createReq.minRows = 100;
createReq.maxRows = 4096;
createReq.fsyncPeriod = 3000;
createReq.walLevel = 1;
createReq.precision = 0;
createReq.compression = 2;
createReq.replications = 1;
createReq.strict = 1;
createReq.cacheLastRow = 0;
createReq.ignoreExist = 1;
createReq.numOfStables = 0;
createReq.numOfRetensions = 0;
int32_t contLen = tSerializeSCreateDbReq(NULL, 0, &createReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSCreateDbReq(pReq, contLen, &createReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DB, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
test.SendShowReq(TSDB_MGMT_TABLE_DB, "user_databases", "");
EXPECT_EQ(test.GetShowRows(), 3);
}
{
SAlterDbReq alterdbReq = {0};
strcpy(alterdbReq.db, "1.d2");
alterdbReq.buffer = 12;
alterdbReq.pageSize = -1;
alterdbReq.pages = -1;
alterdbReq.daysPerFile = -1;
alterdbReq.daysToKeep0 = -1;
alterdbReq.daysToKeep1 = -1;
alterdbReq.daysToKeep2 = -1;
alterdbReq.fsyncPeriod = 4000;
alterdbReq.walLevel = 2;
alterdbReq.strict = 1;
alterdbReq.cacheLastRow = 1;
alterdbReq.replications = 3;
int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSAlterDbReq(pReq, contLen, &alterdbReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_DB, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
}
{
SAlterDbReq alterdbReq = {0};
strcpy(alterdbReq.db, "1.d2");
alterdbReq.buffer = 12;
alterdbReq.pageSize = -1;
alterdbReq.pages = -1;
alterdbReq.daysPerFile = -1;
alterdbReq.daysToKeep0 = -1;
alterdbReq.daysToKeep1 = -1;
alterdbReq.daysToKeep2 = -1;
alterdbReq.fsyncPeriod = 4000;
alterdbReq.walLevel = 2;
alterdbReq.strict = 1;
alterdbReq.cacheLastRow = 1;
alterdbReq.replications = 1;
int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSAlterDbReq(pReq, contLen, &alterdbReq);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_DB, pReq, contLen);
ASSERT_NE(pRsp, nullptr);
ASSERT_EQ(pRsp->code, 0);
}
#endif
}
taos-tools @ 2f3dfddd
比较 59e0ebaf...2f3dfddd
Subproject commit 59e0ebaf4905e4cb6d95a01c58b3fa507abc5a20
Subproject commit 2f3dfddd4d9a869e706ba3cf98fb6d769404cd7c
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或