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

[TD-93] add ref to users

This commit is contained in:
slguan 2020-04-03 14:20:57 +08:00
parent e068bf335b
commit 07fe4defd4
7 changed files with 51 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/inc/mnode.h
View File

@ -181,7 +181,6 @@ typedef struct _user_obj {
int8_t writeAuth;
int8_t reserved[13];
int8_t updateEnd[1];
struct _user_obj *prev, *next;
struct _acctObj * pAcct;
SQqueryList * pQList; // query list
SStreamList * pSList; // stream list
@ -216,7 +215,6 @@ typedef struct _acctObj {
int8_t updateEnd[1];
SAcctInfo acctInfo;
SDbObj * pHead;
SUserObj * pUser;
pthread_mutex_t mutex;
} SAcctObj;

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

@ -30,6 +30,8 @@ typedef enum {
int32_t acctInit();
void acctCleanUp();
SAcctObj *acctGetAcct(char *acctName);
void acctIncRef(SAcctObj *pAcct);
void acctDecRef(SAcctObj *pAcct);
int32_t acctCheck(SAcctObj *pAcct, EAcctGrantType type);
void acctAddDb(SAcctObj *pAcct, SDbObj *pDb);

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

@ -28,7 +28,7 @@ void mgmtCleanUpDbs();
SDbObj *mgmtGetDb(char *db);
SDbObj *mgmtGetDbByTableId(char *db);
void mgmtIncDbRef(SDbObj *pDb);
void mgmtDecDbRef(SDbObj *pDb)
void mgmtDecDbRef(SDbObj *pDb);
bool mgmtCheckIsMonitorDB(char *db, char *monitordb);
void mgmtDropAllDbs(SAcctObj *pAcct);

33
src/mnode/src/mgmtAcct.c
View File

@ -18,6 +18,8 @@
#include "taoserror.h"
#include "mnode.h"
#include "mgmtAcct.h"
#include "mgmtDb.h"
#include "mgmtUser.h"
#ifndef _ACCOUNT
static SAcctObj tsAcctObj = {0};
@ -30,6 +32,8 @@ int32_t acctInit() {
void acctCleanUp() {}
SAcctObj *acctGetAcct(char *acctName) { return &tsAcctObj; }
void acctIncRef(SAcctObj *pAcct) {}
void acctDecRef(SAcctObj *pAcct) {}
int32_t acctCheck(SAcctObj *pAcct, EAcctGrantType type) { return TSDB_CODE_SUCCESS; }
#endif
@ -46,6 +50,8 @@ void acctAddDb(SAcctObj *pAcct, SDbObj *pDb) {
pAcct->pHead = pDb;
pAcct->acctInfo.numOfDbs++;
pthread_mutex_unlock(&pAcct->mutex);
mgmtIncDbRef(pDb);
}
void acctRemoveDb(SAcctObj *pAcct, SDbObj *pDb) {
@ -64,37 +70,24 @@ void acctRemoveDb(SAcctObj *pAcct, SDbObj *pDb) {
pAcct->acctInfo.numOfDbs--;
pthread_mutex_unlock(&pAcct->mutex);
mgmtDecDbRef(pDb);
}
void acctAddUser(SAcctObj *pAcct, SUserObj *pUser) {
pthread_mutex_lock(&pAcct->mutex);
pUser->next = pAcct->pUser;
pUser->prev = NULL;
if (pAcct->pUser) {
pAcct->pUser->prev = pUser;
}
pAcct->pUser = pUser;
pAcct->acctInfo.numOfUsers++;
pUser->pAcct = pAcct;
pthread_mutex_unlock(&pAcct->mutex);
mgmtIncUserRef(pUser);
}
void acctRemoveUser(SAcctObj *pAcct, SUserObj *pUser) {
pthread_mutex_lock(&pAcct->mutex);
if (pUser->prev) {
pUser->prev->next = pUser->next;
}
if (pUser->next) {
pUser->next->prev = pUser->prev;
}
if (pUser->prev == NULL) {
pAcct->pUser = pUser->next;
}
pAcct->acctInfo.numOfUsers--;
pUser->pAcct = NULL;
pthread_mutex_unlock(&pAcct->mutex);
mgmtDecUserRef(pUser);
}

2
src/mnode/src/mgmtSdb.c
View File

@ -325,7 +325,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
.pObj = pMetaRow
};
sdbDecRef(pTable, pMetaRow);
(*pTable->destroyFp)(&oper);
(*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, rowHead->data);
pTable->numOfRows--;
sdbTrace("table:%s, version:%" PRId64 " numOfRows:%d, read deleted record:%s",
@ -342,7 +341,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
.pObj = pMetaRow
};
sdbDecRef(pTable, pMetaRow);
(*pTable->destroyFp)(&oper);
(*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, rowHead->data);
int32_t code = (*pTable->decodeFp)(&oper);

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

@ -117,6 +117,7 @@ void mgmtAddShellShowRetrieveHandle(uint8_t msgType, SShowRetrieveFp fp) {
void mgmtProcessTranRequest(SSchedMsg *sched) {
SQueuedMsg *queuedMsg = sched->msg;
(*tsMgmtProcessShellMsgFp[queuedMsg->msgType])(queuedMsg);
mgmtDecUserRef(queuedMsg->pUser);
rpcFreeCont(queuedMsg->pCont);
free(queuedMsg);
}
@ -167,6 +168,7 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
queuedMsg.pUser = pUser;
queuedMsg.usePublicIp = usePublicIp;
(*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(&queuedMsg);
mgmtDecUserRef(pUser);
rpcFreeCont(rpcMsg->pCont);
} else {
SQueuedMsg *queuedMsg = calloc(1, sizeof(SQueuedMsg));
@ -354,9 +356,11 @@ static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secr
SUserObj *pUser = mgmtGetUser(user);
if (pUser == NULL) {
*secret = 0;
mgmtDecUserRef(pUser);
return TSDB_CODE_INVALID_USER;
} else {
memcpy(secret, pUser->pass, TSDB_KEY_LEN);
mgmtDecUserRef(pUser);
return TSDB_CODE_SUCCESS;
}
}
@ -372,28 +376,19 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
}
int32_t code;
SUserObj *pUser = mgmtGetUser(connInfo.user);
if (pUser == NULL) {
code = TSDB_CODE_INVALID_USER;
goto connect_over;
}
if (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS) {
code = TSDB_CODE_GRANT_EXPIRED;
goto connect_over;
}
SAcctObj *pAcct = acctGetAcct(pUser->acct);
if (pAcct == NULL) {
code = TSDB_CODE_INVALID_ACCT;
goto connect_over;
}
code = taosCheckVersion(pConnectMsg->clientVersion, version, 3);
if (code != TSDB_CODE_SUCCESS) {
goto connect_over;
}
SUserObj *pUser = pMsg->pUser;
SAcctObj *pAcct = pUser->pAcct;
if (pConnectMsg->db[0]) {
char dbName[TSDB_TABLE_ID_LEN * 3] = {0};
sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db);
@ -415,7 +410,7 @@ static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
pConnectRsp->writeAuth = pUser->writeAuth;
pConnectRsp->superAuth = pUser->superAuth;
if (connInfo.serverIp == tsPublicIpInt) {
if (pMsg->usePublicIp) {
mgmtGetMnodePublicIpList(&pConnectRsp->ipList);
} else {
mgmtGetMnodePrivateIpList(&pConnectRsp->ipList);

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

@ -25,14 +25,10 @@
#include "mgmtShell.h"
#include "mgmtUser.h"
void * tsUserSdb = NULL;
static void * tsUserSdb = NULL;
static int32_t tsUserUpdateSize = 0;
static int32_t mgmtDropUser(SAcctObj *pAcct, char *name);
static int32_t mgmtUpdateUser(SUserObj *pUser);
static int32_t mgmtGetUserMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg);
static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg);
@ -48,7 +44,7 @@ static int32_t mgmtUserActionInsert(SSdbOperDesc *pOper) {
if (pAcct != NULL) {
acctAddUser(pAcct, pUser);
mgmtIncUserRef(pUser);
acctDecRef(pAcct);
}
else {
mError("user:%s, acct:%s info not exist in sdb", pUser->user, pUser->acct);
@ -64,7 +60,7 @@ static int32_t mgmtUserActionDelete(SSdbOperDesc *pOper) {
if (pAcct != NULL) {
acctRemoveUser(pAcct, pUser);
mgmtDecUserRef(pUser);
acctDecRef(pAcct);
}
return TSDB_CODE_SUCCESS;
@ -122,6 +118,7 @@ int32_t mgmtInitUsers() {
mgmtCreateUser(pAcct, "root", "taosdata");
mgmtCreateUser(pAcct, "monitor", tsInternalPass);
mgmtCreateUser(pAcct, "_root", tsInternalPass);
acctDecRef(pAcct);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_USER, mgmtProcessCreateUserMsg);
mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_USER, mgmtProcessAlterUserMsg);
@ -159,7 +156,6 @@ static int32_t mgmtUpdateUser(SUserObj *pUser) {
int32_t code = sdbUpdateRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
tfree(pUser);
code = TSDB_CODE_SDB_ERROR;
}
@ -176,9 +172,10 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
return TSDB_CODE_INVALID_MSG;
}
SUserObj *pUser = (SUserObj *)sdbGetRow(tsUserSdb, name);
SUserObj *pUser = mgmtGetUser(name);
if (pUser != NULL) {
mTrace("user:%s is already there", name);
mgmtDecUserRef(pUser);
return TSDB_CODE_USER_ALREADY_EXIST;
}
@ -214,19 +211,7 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
return code;
}
static int32_t mgmtDropUser(SAcctObj *pAcct, char *name) {
SUserObj *pUser;
pUser = (SUserObj *)sdbGetRow(tsUserSdb, name);
if (pUser == NULL) {
mWarn("user:%s is not there", name);
return TSDB_CODE_INVALID_USER;
}
if (strcmp(pAcct->user, pUser->acct) != 0) {
return TSDB_CODE_NO_RIGHTS;
}
static int32_t mgmtDropUser(SUserObj *pUser) {
SSdbOperDesc oper = {
.type = SDB_OPER_TYPE_GLOBAL,
.table = tsUserSdb,
@ -278,9 +263,9 @@ static int32_t mgmtGetUserMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCon
}
pShow->numOfRows = pUser->pAcct->acctInfo.numOfUsers;
pShow->pNode = pUser->pAcct->pUser;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
mgmtDecUserRef(pUser);
return 0;
}
@ -291,10 +276,9 @@ static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void
char *pWrite;
while (numOfRows < rows) {
pUser = (SUserObj *)pShow->pNode;
pShow->pNode = sdbFetchRow(tsUserSdb, pShow->pNode, (void **) &pUser);
if (pUser == NULL) break;
pShow->pNode = (void *)pUser->next;
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
@ -316,6 +300,7 @@ static int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void
cols++;
numOfRows++;
mgmtDecUserRef(pUser);
}
pShow->numOfReads += numOfRows;
return numOfRows;
@ -367,6 +352,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
if (strcmp(pUser->user, "monitor") == 0 || (strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
mgmtDecUserRef(pUser);
return;
}
@ -396,10 +382,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
}
mgmtSendSimpleResp(pMsg->thandle, code);
return;
}
if ((pAlter->flag & TSDB_ALTER_USER_PRIVILEGES) != 0) {
} else if ((pAlter->flag & TSDB_ALTER_USER_PRIVILEGES) != 0) {
bool hasRight = false;
if (strcmp(pUser->user, "root") == 0) {
@ -441,10 +424,11 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
}
mgmtSendSimpleResp(pMsg->thandle, code);
return;
} else {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
}
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
mgmtDecUserRef(pUser);
}
static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
@ -463,6 +447,7 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
if (strcmp(pUser->user, "monitor") == 0 || strcmp(pUser->user, pUser->acct) == 0 ||
(strcmp(pUser->user + 1, pUser->acct) == 0 && pUser->user[0] == '_')) {
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
mgmtDecUserRef(pUser);
return ;
}
@ -474,9 +459,7 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
} else if (strcmp(pUser->user, pOperUser->user) == 0) {
hasRight = false;
} else if (pOperUser->superAuth) {
if (strcmp(pUser->user, "root") == 0) {
hasRight = false;
} else if (strcmp(pOperUser->acct, pUser->acct) != 0) {
if (strcmp(pOperUser->acct, pUser->acct) != 0) {
hasRight = false;
} else {
hasRight = true;
@ -484,22 +467,23 @@ static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg) {
}
if (hasRight) {
code = mgmtDropUser(pUser->pAcct, pDrop->user);
code = mgmtDropUser(pUser);
if (code == TSDB_CODE_SUCCESS) {
mLPrint("user:%s is dropped by %s, result:%d", pUser->user, pOperUser->user, tstrerror(code));
mLPrint("user:%s is dropped by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
}
} else {
code = TSDB_CODE_NO_RIGHTS;
}
mgmtSendSimpleResp(pMsg->thandle, code);
mgmtDecUserRef(pUser);
}
void mgmtDropAllUsers(SAcctObj *pAcct) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfUsers = 0;
int32_t acctNameLen = strlen(pAcct->user);
void * pNode = NULL;
void * pLastNode = NULL;
int32_t numOfUsers = 0;
int32_t acctNameLen = strlen(pAcct->user);
SUserObj *pUser = NULL;
while (1) {
@ -516,8 +500,9 @@ void mgmtDropAllUsers(SAcctObj *pAcct) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfUsers++;
continue;
}
mgmtDecUserRef(pUser);
}
mTrace("acct:%s, all users:%d is dropped from sdb", pAcct->user, numOfUsers);