diff --git a/include/common/tmsg.h b/include/common/tmsg.h index a8053d885403a94fa3486e39525209866a59a117..8741af54ec8520475e4b55b046a95b12c9e6f9b1 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -424,7 +424,9 @@ int32_t tDeserializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq* typedef struct { char user[TSDB_USER_LEN]; + int32_t version; int8_t superAuth; + SHashObj* createdDbs; SHashObj* readDbs; SHashObj* writeDbs; } SGetUserAuthRsp; diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index 30d1bd0a518415d98b7358337c94834aa70716a2..5fc9e08de46bb739e295199d228edfb1d4725ab3 100644 --- a/include/libs/catalog/catalog.h +++ b/include/libs/catalog/catalog.h @@ -40,6 +40,9 @@ enum { CTG_DBG_STB_RENT_NUM, }; +#define USER_AUTH_READ 1 +#define USER_AUTH_WRITE 2 +#define USER_AUTH_ALL 4 typedef struct SCatalogReq { SArray *pTableName; // element is SNAME @@ -57,6 +60,7 @@ typedef struct SMetaData { typedef struct SCatalogCfg { uint32_t maxTblCacheNum; uint32_t maxDBCacheNum; + uint32_t maxUserCacheNum; uint32_t dbRentSec; uint32_t stbRentSec; } SCatalogCfg; @@ -225,6 +229,8 @@ int32_t catalogGetIndexInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo** pInfo); +int32_t catalogGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth); + /** * Destroy catalog and relase all resources diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index 918892b78627cdaa0c6658fba9349abb0f511635..d6aa03aecbf561732ae5eb7dbfce84a0c01370fb 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/source/libs/catalog/inc/catalogInt.h @@ -54,6 +54,7 @@ enum { CTG_ACT_REMOVE_DB, CTG_ACT_REMOVE_STB, CTG_ACT_REMOVE_TBL, + CTG_ACT_UPDATE_USER, CTG_ACT_MAX }; @@ -95,8 +96,18 @@ typedef struct SCtgRentMgmt { SCtgRentSlot *slots; } SCtgRentMgmt; +typedef struct SCtgUserAuth { + int32_t version; + SRWLatch lock; + bool superUser; + SHashObj *createdDbs; + SHashObj *readDbs; + SHashObj *writeDbs; +} SCtgUserAuth; + typedef struct SCatalog { uint64_t clusterId; + SHashObj *userCache; //key:user, value:SCtgUserAuth SHashObj *dbCache; //key:dbname, value:SCtgDBCache SCtgRentMgmt dbRent; SCtgRentMgmt stbRent; @@ -124,6 +135,8 @@ typedef struct SCtgCacheStat { uint64_t vgMissNum; uint64_t tblHitNum; uint64_t tblMissNum; + uint64_t userHitNum; + uint64_t userMissNum; } SCtgCacheStat; typedef struct SCatalogStat { @@ -169,6 +182,11 @@ typedef struct SCtgRemoveTblMsg { uint64_t dbId; } SCtgRemoveTblMsg; +typedef struct SCtgUpdateUserMsg { + SCatalog* pCtg; + SGetUserAuthRsp userAuth; +} SCtgUpdateTblMsg; + typedef struct SCtgMetaAction { int32_t act; @@ -234,6 +252,8 @@ typedef struct SCtgAction { #define CTG_FLAG_SYS_DB 0x8 #define CTG_FLAG_FORCE_UPDATE 0x10 +#define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v)) + #define CTG_FLAG_IS_STB(_flag) ((_flag) & CTG_FLAG_STB) #define CTG_FLAG_IS_NOT_STB(_flag) ((_flag) & CTG_FLAG_NOT_STB) #define CTG_FLAG_IS_UNKNOWN_STB(_flag) ((_flag) & CTG_FLAG_UNKNOWN_STB) diff --git a/source/libs/catalog/src/catalog.c b/source/libs/catalog/src/catalog.c index 6f1f34a57b2fe56f45be909ff6a42b19726a79f6..39a784d00c9f510583e504a2910e443564f1b3cb 100644 --- a/source/libs/catalog/src/catalog.c +++ b/source/libs/catalog/src/catalog.c @@ -51,6 +51,11 @@ SCtgAction gCtgAction[CTG_ACT_MAX] = {{ CTG_ACT_REMOVE_TBL, "remove tbMeta", ctgActRemoveTbl + }, + { + CTG_ACT_UPDATE_USER, + "update user", + ctgActUpdateUser } }; @@ -357,6 +362,30 @@ _return: CTG_RET(code); } +int32_t ctgPushUpdateUserMsgInQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq) { + int32_t code = 0; + SCtgMetaAction action= {.act = CTG_ACT_UPDATE_USER, .syncReq = syncReq}; + SCtgUpdateUserMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateUserMsg)); + if (NULL == msg) { + ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateUserMsg)); + CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR); + } + + msg->pCtg = pCtg; + msg->userAuth = *pAuth; + + action.data = msg; + + CTG_ERR_JRET(ctgPushAction(pCtg, &action)); + + return TSDB_CODE_SUCCESS; + +_return: + + taosMemoryFreeClear(msg); + + CTG_RET(code); +} int32_t ctgAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) { CTG_LOCK(CTG_READ, &dbCache->vgLock); @@ -687,6 +716,43 @@ int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEp return TSDB_CODE_SUCCESS; } +int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *user, SGetUserAuthRsp *authRsp) { + char *msg = NULL; + int32_t msgLen = 0; + + ctgDebug("try to get user auth from mnode, user:%s", user); + + int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)]((void *)user, &msg, 0, &msgLen); + if (code) { + ctgError("Build get user auth msg failed, code:%x, db:%s", code, user); + CTG_ERR_RET(code); + } + + SRpcMsg rpcMsg = { + .msgType = TDMT_MND_GET_USER_AUTH, + .pCont = msg, + .contLen = msgLen, + }; + + SRpcMsg rpcRsp = {0}; + + rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp); + if (TSDB_CODE_SUCCESS != rpcRsp.code) { + ctgError("error rsp for get user auth, error:%s, user:%s", tstrerror(rpcRsp.code), user); + CTG_ERR_RET(rpcRsp.code); + } + + code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)](authRsp, rpcRsp.pCont, rpcRsp.contLen); + if (code) { + ctgError("Process get user auth rsp failed, code:%x, user:%s", code, user); + CTG_ERR_RET(code); + } + + ctgDebug("Got user auth from mnode, user:%s", user); + + return TSDB_CODE_SUCCESS; +} + int32_t ctgIsTableMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) { if (NULL == pCtg->dbCache) { @@ -859,6 +925,55 @@ int32_t ctgGetTableTypeFromCache(SCatalog* pCtg, const char* dbFName, const char return TSDB_CODE_SUCCESS; } +int32_t ctgGetUserDbAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, bool *inCache, int32_t *auth) { + if (NULL == pCtg->userCache) { + ctgDebug("empty user auth cache, user:%s", user); + goto _return; + } + + SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, user, strlen(user)); + if (NULL == pUser) { + ctgDebug("user not in cache, user:%s", user); + goto _return; + } + + *inCache = true; + + ctgDebug("Got user from cache, user:%s", user); + CTG_CACHE_STAT_ADD(userHitNum, 1); + + if (pUser->superUser) { + CTG_FLAG_SET(auth, USER_AUTH_ALL); + return TSDB_CODE_SUCCESS; + } + + CTG_LOCK(CTG_READ, &pUser->lock); + if (pUser->createdDbs && taosHashGet(pUser->createdDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_ALL); + CTG_UNLOCK(CTG_READ, &pUser->lock); + return TSDB_CODE_SUCCESS; + } + + if (pUser->readDbs && taosHashGet(pUser->readDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_READ); + } + + if (pUser->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_WRITE); + } + + CTG_UNLOCK(CTG_READ, &pUser->lock); + + return TSDB_CODE_SUCCESS; + +_return: + + *inCache = false; + CTG_CACHE_STAT_ADD(userMissNum, 1); + + return TSDB_CODE_SUCCESS; +} + int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) { SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName}; char *msg = NULL; @@ -1952,6 +2067,44 @@ _return: CTG_RET(code); } +int32_t ctgGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) { + bool inCache = false; + int32_t code = 0; + *auth = 0; + + CTG_ERR_RET(ctgGetUserDbAuthFromCache(pCtg, user, dbFName, &inCache, auth)); + + if (inCache) { + return TSDB_CODE_SUCCESS; + } + + SGetUserAuthRsp authRsp = {0}; + CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pRpc, pMgmtEps, user, &authRsp)); + + if (authRsp.superAuth) { + CTG_FLAG_SET(auth, USER_AUTH_ALL); + goto _return; + } + + if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_ALL); + goto _return; + } + + if (authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_READ); + } + + if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName))) { + CTG_FLAG_SET(auth, USER_AUTH_WRITE); + } + +_return: + + ctgPushUpdateUserMsgInQueue(pCtg, &authRsp, false); + + return TSDB_CODE_SUCCESS; +} int32_t ctgActUpdateVg(SCtgMetaAction *action) { @@ -2121,6 +2274,67 @@ _return: CTG_RET(code); } +int32_t ctgActUpdateUser(SCtgMetaAction *action) { + int32_t code = 0; + SCtgUpdateUserMsg *msg = action->data; + SCatalog* pCtg = msg->pCtg; + + if (NULL == pCtg->userCache) { + pCtg->userCache = taosHashInit(gCtgMgmt.cfg.maxUserCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK); + if (NULL == pCtg->userCache) { + ctgError("taosHashInit %d user cache failed", gCtgMgmt.cfg.maxUserCacheNum); + CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); + } + } + + SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user)); + if (NULL == pUser) { + SCtgUserAuth userAuth = {0}; + + userAuth.version = msg->userAuth.version; + userAuth.superUser = msg->userAuth.superAuth; + userAuth.createdDbs = msg->userAuth.createdDbs; + userAuth.readDbs = msg->userAuth.readDbs; + userAuth.writeDbs = msg->userAuth.writeDbs; + + if (taosHashPut(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user), &userAuth, sizeof(userAuth))) { + ctgError("taosHashPut user %s to cache failed", msg->userAuth.user); + CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); + } + + return TSDB_CODE_SUCCESS; + } + + pUser->version = msg->userAuth.version; + + CTG_LOCK(CTG_WRITE, &pUser->lock); + + taosHashCleanup(pUser->createdDbs); + pUser->createdDbs = msg->userAuth.createdDbs; + msg->userAuth.createdDbs = NULL; + + taosHashCleanup(pUser->readDbs); + pUser->readDbs = msg->userAuth.readDbs; + msg->userAuth.readDbs = NULL; + + taosHashCleanup(pUser->writeDbs); + pUser->writeDbs = msg->userAuth.writeDbs; + msg->userAuth.writeDbs = NULL; + + CTG_UNLOCK(CTG_WRITE, &pUser->lock); + +_return: + + + taosHashCleanup(msg->userAuth.createdDbs); + taosHashCleanup(msg->userAuth.readDbs); + taosHashCleanup(msg->userAuth.writeDbs); + + taosMemoryFreeClear(msg); + + CTG_RET(code); +} + void* ctgUpdateThreadFunc(void* param) { setThreadName("catalog"); @@ -2880,6 +3094,21 @@ _return: CTG_API_LEAVE(code); } +int32_t catalogGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) { + CTG_API_ENTER(); + + if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == user || NULL == dbFName || NULL == auth) { + CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT); + } + + int32_t code = 0; + CTG_ERR_JRET(ctgGetUserDbAuth(pCtg, pRpc, pMgmtEps, user, dbFName, auth)); + +_return: + + CTG_API_LEAVE(code); +} + void catalogDestroy(void) { qInfo("start to destroy catalog"); diff --git a/source/libs/qcom/src/querymsg.c b/source/libs/qcom/src/querymsg.c index 286e7d3d44d5c1f2d406b417e3846a0414f97988..822c214fe58bba136f773523da8ec7b0bb4ab768 100644 --- a/source/libs/qcom/src/querymsg.c +++ b/source/libs/qcom/src/querymsg.c @@ -181,6 +181,25 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3 return TSDB_CODE_SUCCESS; } +int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { + if (NULL == msg || NULL == msgLen) { + return TSDB_CODE_TSC_INVALID_INPUT; + } + + SGetUserAuthReq req = {0}; + strncpy(req.user, input, sizeof(req.user)); + + int32_t bufLen = tSerializeSGetUserAuthReq(NULL, 0, &req); + void *pBuf = rpcMallocCont(bufLen); + tSerializeSGetUserAuthReq(pBuf, bufLen, &req); + + *msg = pBuf; + *msgLen = bufLen; + + return TSDB_CODE_SUCCESS; +} + + int32_t queryProcessUseDBRsp(void *output, char *msg, int32_t msgSize) { SUseDbOutput *pOut = output; SUseDbRsp usedbRsp = {0}; @@ -419,6 +438,20 @@ int32_t queryProcessRetrieveFuncRsp(void *output, char *msg, int32_t msgSize) { return TSDB_CODE_SUCCESS; } +int32_t queryProcessGetUserAuthRsp(void *output, char *msg, int32_t msgSize) { + if (NULL == output || NULL == msg || msgSize <= 0) { + return TSDB_CODE_TSC_INVALID_INPUT; + } + + if (tDeserializeSGetUserAuthRsp(msg, msgSize, (SGetUserAuthRsp *)output) != 0) { + qError("tDeserializeSGetUserAuthRsp failed, msgSize:%d", msgSize); + return TSDB_CODE_INVALID_MSG; + } + + return TSDB_CODE_SUCCESS; +} + + void initQueryModuleMsgHandle() { queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryBuildTableMetaReqMsg; queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryBuildTableMetaReqMsg; @@ -427,6 +460,8 @@ void initQueryModuleMsgHandle() { queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_DB_CFG)] = queryBuildGetDBCfgMsg; queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_INDEX)] = queryBuildGetIndexMsg; queryBuildMsg[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)] = queryBuildRetrieveFuncMsg; + queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)] = queryBuildGetUserAuthMsg; + queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryProcessTableMetaRsp; queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryProcessTableMetaRsp; @@ -435,6 +470,7 @@ void initQueryModuleMsgHandle() { queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_DB_CFG)] = queryProcessGetDbCfgRsp; queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_INDEX)] = queryProcessGetIndexRsp; queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)] = queryProcessRetrieveFuncRsp; + queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)] = queryProcessGetUserAuthRsp; } #pragma GCC diagnostic pop