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

Merge pull request #23135 from taosdata/enh/TD-26532-3.0 

enh: exclude sys tables in log/audit db
This commit is contained in:
dapan1121 2023-10-07 19:26:15 +08:00 committed by GitHub
parent 01284c2509 b9767707cb
commit 9f1b2c9e65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 156 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/libs/monitor/monitor.h
View File

@ -206,6 +206,11 @@ typedef struct {
bool comp;
} SMonCfg;
typedef struct {
int8_t state;
tsem_t sem;
} SDmNotifyHandle;
int32_t monInit(const SMonCfg *pCfg);
void monCleanup();
void monRecordLog(int64_t ts, ELogLevel level, const char *content);

19
source/dnode/mgmt/mgmt_dnode/src/dmWorker.c
View File

@ -53,21 +53,26 @@ static void *dmStatusThreadFp(void *param) {
return NULL;
}
tsem_t dmNotifySem;
SDmNotifyHandle dmNotifyHdl = {.state = 0};
static void *dmNotifyThreadFp(void *param) {
SDnodeMgmt *pMgmt = param;
int64_t lastTime = taosGetTimestampMs();
setThreadName("dnode-notify");
if (tsem_init(&dmNotifySem, 0, 0) != 0) {
if (tsem_init(&dmNotifyHdl.sem, 0, 0) != 0) {
return NULL;
}
bool wait = true;
while (1) {
if (pMgmt->pData->dropped || pMgmt->pData->stopped) break;
tsem_wait(&dmNotifySem);
if (wait) tsem_wait(&dmNotifyHdl.sem);
atomic_store_8(&dmNotifyHdl.state, 1);
dmSendNotifyReq(pMgmt);
if (1 == atomic_val_compare_exchange_8(&dmNotifyHdl.state, 1, 0)) {
wait = true;
continue;
}
wait = false;
}
return NULL;
@ -189,11 +194,11 @@ int32_t dmStartNotifyThread(SDnodeMgmt *pMgmt) {
void dmStopNotifyThread(SDnodeMgmt *pMgmt) {
if (taosCheckPthreadValid(pMgmt->notifyThread)) {
tsem_post(&dmNotifySem);
tsem_post(&dmNotifyHdl.sem);
taosThreadJoin(pMgmt->notifyThread, NULL);
taosThreadClear(&pMgmt->notifyThread);
}
tsem_destroy(&dmNotifySem);
tsem_destroy(&dmNotifyHdl.sem);
}
int32_t dmStartMonitorThread(SDnodeMgmt *pMgmt) {

7
source/dnode/vnode/inc/vnode.h
View File

@ -132,9 +132,10 @@ tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name);
int32_t metaGetCachedTbGroup(void *pVnode, tb_uid_t suid, const uint8_t *pKey, int32_t keyLen, SArray **pList);
int32_t metaPutTbGroupToCache(void *pVnode, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
int32_t payloadLen);
bool metaTbInFilterCache(void *pVnode, tb_uid_t suid, int8_t type);
int32_t metaPutTbToFilterCache(void *pVnode, tb_uid_t suid, int8_t type);
int32_t metaSizeOfTbFilterCache(void *pVnode, int8_t type);
bool metaTbInFilterCache(SMeta *pMeta, const void* key, int8_t type);
int32_t metaPutTbToFilterCache(SMeta *pMeta, const void* key, int8_t type);
int32_t metaSizeOfTbFilterCache(SMeta *pMeta, int8_t type);
int32_t metaInitTbFilterCache(SMeta *pMeta);
int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t *numOfCols);

63
source/dnode/vnode/src/meta/metaCache.c
View File

@ -14,6 +14,13 @@
*/
#include "meta.h"
#ifdef TD_ENTERPRISE
extern const char* tkLogStb[];
extern const char* tkAuditStb[];
extern const int tkLogStbNum;
extern const int tkAuditStbNum;
#endif
#define TAG_FILTER_RES_KEY_LEN 32
#define META_CACHE_BASE_BUCKET 1024
#define META_CACHE_STATS_BUCKET 16
@ -69,6 +76,7 @@ struct SMetaCache {
struct STbFilterCache {
SHashObj* pStb;
SHashObj* pStbName;
} STbFilterCache;
};
@ -178,6 +186,13 @@ int32_t metaCacheOpen(SMeta* pMeta) {
goto _err2;
}
pCache->STbFilterCache.pStbName =
taosHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), false, HASH_NO_LOCK);
if (pCache->STbFilterCache.pStbName == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
}
pMeta->pCache = pCache;
return code;
@ -204,6 +219,7 @@ void metaCacheClose(SMeta* pMeta) {
taosHashCleanup(pMeta->pCache->STbGroupResCache.pTableEntry);
taosHashCleanup(pMeta->pCache->STbFilterCache.pStb);
taosHashCleanup(pMeta->pCache->STbFilterCache.pStbName);
taosMemoryFree(pMeta->pCache);
pMeta->pCache = NULL;
@ -893,30 +909,59 @@ int32_t metaTbGroupCacheClear(SMeta* pMeta, uint64_t suid) {
return TSDB_CODE_SUCCESS;
}
bool metaTbInFilterCache(void* pVnode, tb_uid_t suid, int8_t type) {
SMeta* pMeta = ((SVnode*)pVnode)->pMeta;
bool metaTbInFilterCache(SMeta *pMeta, const void* key, int8_t type) {
if (type == 0 && taosHashGet(pMeta->pCache->STbFilterCache.pStb, key, sizeof(tb_uid_t))) {
return true;
}
if (type == 0 && taosHashGet(pMeta->pCache->STbFilterCache.pStb, &suid, sizeof(suid))) {
if (type == 1 && taosHashGet(pMeta->pCache->STbFilterCache.pStbName, key, strlen(key))) {
return true;
}
return false;
}
int32_t metaPutTbToFilterCache(void* pVnode, tb_uid_t suid, int8_t type) {
SMeta* pMeta = ((SVnode*)pVnode)->pMeta;
int32_t metaPutTbToFilterCache(SMeta *pMeta, const void* key, int8_t type) {
if (type == 0) {
return taosHashPut(pMeta->pCache->STbFilterCache.pStb, &suid, sizeof(suid), NULL, 0);
return taosHashPut(pMeta->pCache->STbFilterCache.pStb, key, sizeof(tb_uid_t), NULL, 0);
}
if (type == 1) {
return taosHashPut(pMeta->pCache->STbFilterCache.pStbName, key, strlen(key), NULL, 0);
}
return 0;
}
int32_t metaSizeOfTbFilterCache(void* pVnode, int8_t type) {
SMeta* pMeta = ((SVnode*)pVnode)->pMeta;
int32_t metaSizeOfTbFilterCache(SMeta *pMeta, int8_t type) {
if (type == 0) {
return taosHashGetSize(pMeta->pCache->STbFilterCache.pStb);
}
return 0;
}
int32_t metaInitTbFilterCache(SMeta* pMeta) {
#ifdef TD_ENTERPRISE
int32_t tbNum = 0;
const char** pTbArr = NULL;
const char* dbName = NULL;
if (!(dbName = strchr(pMeta->pVnode->config.dbname, '.'))) return 0;
if (0 == strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {
tbNum = tkLogStbNum;
pTbArr = (const char**)&tkLogStb;
} else if (0 == strncmp(dbName, "audit", TSDB_DB_NAME_LEN)) {
tbNum = tkAuditStbNum;
pTbArr = (const char**)&tkAuditStb;
}
if (tbNum && pTbArr) {
for (int32_t i = 0; i < tbNum; ++i) {
if (metaPutTbToFilterCache(pMeta, pTbArr[i], 1) != 0) {
return terrno ? terrno : -1;
}
}
}
#else
#endif
return 0;
}

4
source/dnode/vnode/src/meta/metaOpen.c
View File

@ -176,6 +176,10 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
goto _err;
}
if (metaInitTbFilterCache(pMeta) != 0) {
goto _err;
}
metaDebug("vgId:%d, meta is opened", TD_VID(pVnode));
*ppMeta = pMeta;

38
source/dnode/vnode/src/meta/metaTable.c
View File

@ -15,7 +15,7 @@
#include "meta.h"
extern tsem_t dmNotifySem;
extern SDmNotifyHandle dmNotifyHdl;
static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema);
static int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema);
@ -28,7 +28,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME);
static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaUpdateSuidIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry);
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid);
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid, int8_t *pSysTbl);
static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey);
// opt ins_tables query
static int metaUpdateBtimeIdx(SMeta *pMeta, const SMetaEntry *pME);
@ -198,7 +198,11 @@ static inline void metaTimeSeriesNotifyCheck(SMeta *pMeta) {
#if defined(TD_ENTERPRISE) && !defined(_TD_DARWIN_64)
int64_t nTimeSeries = metaGetTimeSeriesNum(pMeta, 0);
int64_t deltaTS = nTimeSeries - pMeta->pVnode->config.vndStats.numOfReportedTimeSeries;
if (deltaTS > tsTimeSeriesThreshold) tsem_post(&dmNotifySem);
if (deltaTS > tsTimeSeriesThreshold) {
if (0 == atomic_val_compare_exchange_8(&dmNotifyHdl.state, 1, 2)) {
tsem_post(&dmNotifyHdl.sem);
}
}
#endif
}
@ -303,7 +307,7 @@ int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq, SArray *tb
for (int32_t iChild = 0; iChild < taosArrayGetSize(tbUidList); iChild++) {
tb_uid_t uid = *(tb_uid_t *)taosArrayGet(tbUidList, iChild);
metaDropTableByUid(pMeta, uid, NULL, NULL);
metaDropTableByUid(pMeta, uid, NULL, NULL, NULL);
}
// drop super table
@ -392,6 +396,7 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
nStbEntry.stbEntry.schemaTag = pReq->schemaTag;
int32_t deltaCol = pReq->schemaRow.nCols - oStbEntry.stbEntry.schemaRow.nCols;
bool updStat = deltaCol != 0 && !metaTbInFilterCache(pMeta, pReq->name, 1);
metaWLock(pMeta);
// compare two entry
@ -407,15 +412,16 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
// metaStatsCacheDrop(pMeta, nStbEntry.uid);
if (deltaCol != 0) {
if (updStat) {
metaUpdateStbStats(pMeta, pReq->suid, 0, deltaCol);
}
metaULock(pMeta);
if (deltaCol != 0) {
if (updStat) {
int64_t ctbNum;
metaGetStbStats(pMeta->pVnode, pReq->suid, &ctbNum, NULL);
pMeta->pVnode->config.vndStats.numOfTimeSeries += (ctbNum * deltaCol);
metaTimeSeriesNotifyCheck(pMeta);
}
_exit:
@ -760,6 +766,8 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs
}
metaReaderClear(&mr);
bool sysTbl = (pReq->type == TSDB_CHILD_TABLE) && metaTbInFilterCache(pMeta, pReq->ctb.stbName, 1);
// build SMetaEntry
SVnodeStats *pStats = &pMeta->pVnode->config.vndStats;
me.version = ver;
@ -796,9 +804,12 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs
#endif
++pStats->numOfCTables;
if (!sysTbl) {
int32_t nCols = 0;
metaGetStbStats(pMeta->pVnode, me.ctbEntry.suid, 0, &nCols);
pStats->numOfTimeSeries += nCols - 1;
}
metaWLock(pMeta);
metaUpdateStbStats(pMeta, me.ctbEntry.suid, 1, 0);
@ -852,6 +863,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
int rc = 0;
tb_uid_t uid = 0;
tb_uid_t suid = 0;
int8_t sysTbl = 0;
int type;
rc = tdbTbGet(pMeta->pNameIdx, pReq->name, strlen(pReq->name) + 1, &pData, &nData);
@ -862,12 +874,12 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
uid = *(tb_uid_t *)pData;
metaWLock(pMeta);
rc = metaDropTableByUid(pMeta, uid, &type, &suid);
rc = metaDropTableByUid(pMeta, uid, &type, &suid, &sysTbl);
metaULock(pMeta);
if (rc < 0) goto _exit;
if (type == TSDB_CHILD_TABLE) {
if (!sysTbl && type == TSDB_CHILD_TABLE) {
int32_t nCols = 0;
SVnodeStats *pStats = &pMeta->pVnode->config.vndStats;
if (metaGetStbStats(pMeta->pVnode, suid, NULL, &nCols) == 0) {
@ -898,9 +910,10 @@ void metaDropTables(SMeta *pMeta, SArray *tbUids) {
for (int i = 0; i < taosArrayGetSize(tbUids); ++i) {
tb_uid_t uid = *(tb_uid_t *)taosArrayGet(tbUids, i);
tb_uid_t suid = 0;
int8_t sysTbl = 0;
int type;
metaDropTableByUid(pMeta, uid, &type, &suid);
if (type == TSDB_CHILD_TABLE && suid != 0 && suidHash) {
metaDropTableByUid(pMeta, uid, &type, &suid, &sysTbl);
if (!sysTbl && type == TSDB_CHILD_TABLE && suid != 0 && suidHash) {
int64_t *pVal = tSimpleHashGet(suidHash, &suid, sizeof(tb_uid_t));
if (pVal) {
nCtbDropped = *pVal + 1;
@ -1059,7 +1072,7 @@ static int metaDeleteTtl(SMeta *pMeta, const SMetaEntry *pME) {
return ttlMgrDeleteTtl(pMeta->pTtlMgr, &ctx);
}
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid) {
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *pSuid, int8_t* pSysTbl) {
void *pData = NULL;
int nData = 0;
int rc = 0;
@ -1088,7 +1101,6 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *p
void *tData = NULL;
int tLen = 0;
if (tdbTbGet(pMeta->pUidIdx, &e.ctbEntry.suid, sizeof(tb_uid_t), &tData, &tLen) == 0) {
STbDbKey tbDbKey = {.uid = e.ctbEntry.suid, .version = ((SUidIdxVal *)tData)[0].version};
if (tdbTbGet(pMeta->pTbDb, &tbDbKey, sizeof(tbDbKey), &tData, &tLen) == 0) {
@ -1098,6 +1110,8 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type, tb_uid_t *p
tDecoderInit(&tdc, tData, tLen);
metaDecodeEntry(&tdc, &stbEntry);
if (pSysTbl) *pSysTbl = metaTbInFilterCache(pMeta, stbEntry.name, 1) ? 1 : 0;
SSchema *pTagColumn = NULL;
SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
if (pTagSchema->nCols == 1 && pTagSchema->pSchema[0].type == TSDB_DATA_TYPE_JSON) {

41
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -566,8 +566,7 @@ int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num) {
}
#ifdef TD_ENTERPRISE
#define TK_LOG_STB_NUM 19
static const char *tkLogStb[TK_LOG_STB_NUM] = {"cluster_info",
const char *tkLogStb[] = {"cluster_info",
"data_dir",
"dnodes_info",
"d_info",
@ -586,22 +585,36 @@ static const char *tkLogStb[TK_LOG_STB_NUM] = {"cluster_info",
"temp_dir",
"vgroups_info",
"vnodes_role"};
const char *tkAuditStb[] = {"operations"};
const int tkLogStbNum = ARRAY_SIZE(tkLogStb);
const int tkAuditStbNum = ARRAY_SIZE(tkAuditStb);
// exclude stbs of taoskeeper log
static int32_t vnodeGetTimeSeriesBlackList(SVnode *pVnode) {
char *dbName = strchr(pVnode->config.dbname, '.');
if (!dbName || 0 != strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {
return 0;
int32_t tbSize = 0;
int32_t tbNum = 0;
const char **pTbArr = NULL;
const char *dbName = NULL;
if (!(dbName = strchr(pVnode->config.dbname, '.'))) return 0;
if (0 == strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {
tbNum = tkLogStbNum;
pTbArr = (const char **)&tkLogStb;
} else if (0 == strncmp(dbName, "audit", TSDB_DB_NAME_LEN)) {
tbNum = tkAuditStbNum;
pTbArr = (const char **)&tkAuditStb;
}
int32_t tbSize = metaSizeOfTbFilterCache(pVnode, 0);
if (tbSize < TK_LOG_STB_NUM) {
for (int32_t i = 0; i < TK_LOG_STB_NUM; ++i) {
tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, tkLogStb[i]);
if (tbNum && pTbArr) {
tbSize = metaSizeOfTbFilterCache(pVnode->pMeta, 0);
if (tbSize < tbNum) {
for (int32_t i = 0; i < tbNum; ++i) {
tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, pTbArr[i]);
if (suid != 0) {
metaPutTbToFilterCache(pVnode, suid, 0);
metaPutTbToFilterCache(pVnode->pMeta, &suid, 0);
}
}
tbSize = metaSizeOfTbFilterCache(pVnode, 0);
tbSize = metaSizeOfTbFilterCache(pVnode->pMeta, 0);
}
}
return tbSize;
@ -611,7 +624,7 @@ static int32_t vnodeGetTimeSeriesBlackList(SVnode *pVnode) {
static bool vnodeTimeSeriesFilter(void *arg1, void *arg2) {
SVnode *pVnode = (SVnode *)arg1;
if (metaTbInFilterCache(pVnode, *(tb_uid_t *)(arg2), 0)) {
if (metaTbInFilterCache(pVnode->pMeta, arg2, 0)) {
return true;
}
return false;
@ -626,9 +639,9 @@ int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num) {
}
int32_t tbFilterSize = 0;
#ifdef TD_ENTERPRISE
#ifdef TD_ENTERPRISE
tbFilterSize = vnodeGetTimeSeriesBlackList(pVnode);
#endif
#endif
if ((!tbFilterSize && vnodeGetStbIdList(pVnode, 0, suidList) < 0) ||
(tbFilterSize && vnodeGetStbIdListByFilter(pVnode, 0, suidList, vnodeTimeSeriesFilter, pVnode) < 0)) {