diff --git a/include/libs/index/index.h b/include/libs/index/index.h index 47eb97cc3a..453b49e4c6 100644 --- a/include/libs/index/index.h +++ b/include/libs/index/index.h @@ -29,6 +29,12 @@ typedef struct SIndexOpts SIndexOpts; typedef struct SIndexMultiTermQuery SIndexMultiTermQuery; typedef struct SArray SIndexMultiTerm; +typedef struct SIndex SIndexJson; +typedef struct SIndexTerm SIndexJsonTerm; +typedef struct SIndexOpts SIndexJsonOpts; +typedef struct SIndexMultiTermQuery SIndexJsonMultiTermQuery; +typedef struct SArray SIndexJsonMultiTerm; + typedef enum { ADD_VALUE, // add index colume value DEL_VALUE, // delete index column value @@ -39,24 +45,108 @@ typedef enum { } SIndexOperOnColumn; typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType; -typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX = 3 } EIndexQueryType; +typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX = 3, QUERY_RANGE = 4 } EIndexQueryType; + /* - * @param: oper - * + * create multi query + * @param oper (input, relation between querys) */ SIndexMultiTermQuery* indexMultiTermQueryCreate(EIndexOperatorType oper); -void indexMultiTermQueryDestroy(SIndexMultiTermQuery* pQuery); -int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EIndexQueryType type); + /* - * @param: - * @param: + * destroy multi query + * @param pQuery (input, multi-query-object to be destory) + */ + +void indexMultiTermQueryDestroy(SIndexMultiTermQuery* pQuery); +/* + * add query to multi query + * @param pQuery (input, multi-query-object) + * @param term (input, single query term) + * @param type (input, single query type) + * @return error code + */ +int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EIndexQueryType type); +/* + * open index + * @param opt (input, index opt) + * @param path (input, index path) + * @param index (output, index object) + * @return error code + */ +int indexOpen(SIndexOpts* opt, const char* path, SIndex** index); +/* + * close index + * @param index (input, index to be closed) + * @return error code */ -int indexOpen(SIndexOpts* opt, const char* path, SIndex** index); void indexClose(SIndex* index); -int indexPut(SIndex* index, SIndexMultiTerm* terms, uint64_t uid); -int indexDelete(SIndex* index, SIndexMultiTermQuery* query); -int indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result); -int indexRebuild(SIndex* index, SIndexOpts* opt); + +/* + * insert terms into index + * @param index (input, index object) + * @param term (input, terms inserted into index) + * @param uid (input, uid of terms) + * @return error code + */ +int indexPut(SIndex* index, SIndexMultiTerm* terms, uint64_t uid); +/* + * delete terms that meet query condition + * @param index (input, index object) + * @param query (input, condition query to deleted) + * @return error code + */ + +int indexDelete(SIndex* index, SIndexMultiTermQuery* query); +/* + * search index + * @param index (input, index object) + * @param query (input, multi query condition) + * @param result(output, query result) + * @return error code + */ +int indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result); +/* + * rebuild index + * @param index (input, index object) + * @parma opt (input, rebuild index opts) + * @return error code + */ +int indexRebuild(SIndex* index, SIndexOpts* opt); + +/* + * open index + * @param opt (input,index json opt) + * @param path (input, index json path) + * @param index (output, index json object) + * @return error code + */ +int tIndexJsonOpen(SIndexJsonOpts* opts, const char* path, SIndexJson** index); +/* + * close index + * @param index (input, index to be closed) + * @return void + */ + +void tIndexJsonClose(SIndexJson* index); + +/* + * insert terms into index + * @param index (input, index object) + * @param term (input, terms inserted into index) + * @param uid (input, uid of terms) + * @return error code + */ +int tIndexJsonPut(SIndexJson* index, SIndexJsonMultiTerm* terms, uint64_t uid); +/* + * search index + * @param index (input, index object) + * @param query (input, multi query condition) + * @param result(output, query result) + * @return error code + */ + +int tIndexJsonSearch(SIndexJson* index, SIndexJsonMultiTermQuery* query, SArray* result); /* * @param * @param diff --git a/include/libs/monitor/monitor.h b/include/libs/monitor/monitor.h index 4b7c6a9bc9..b3b6091b95 100644 --- a/include/libs/monitor/monitor.h +++ b/include/libs/monitor/monitor.h @@ -23,6 +23,11 @@ extern "C" { #endif +#define MON_STATUS_LEN 8 +#define MON_ROLE_LEN 9 +#define MON_VER_LEN 12 +#define MON_LOG_LEN 1024 + typedef struct { int32_t dnode_id; char dnode_ep[TSDB_EP_LEN]; @@ -31,19 +36,19 @@ typedef struct { typedef struct { int32_t dnode_id; char dnode_ep[TSDB_EP_LEN]; - char status[8]; + char status[MON_STATUS_LEN]; } SMonDnodeDesc; typedef struct { int32_t mnode_id; char mnode_ep[TSDB_EP_LEN]; - char role[8]; + char role[MON_ROLE_LEN]; } SMonMnodeDesc; typedef struct { char first_ep[TSDB_EP_LEN]; int32_t first_ep_dnode_id; - char version[12]; + char version[MON_VER_LEN]; float master_uptime; // day int32_t monitor_interval; // sec int32_t vgroups_total; @@ -57,19 +62,18 @@ typedef struct { typedef struct { int32_t dnode_id; - int8_t vnode_online; - char vnode_role[8]; + char vnode_role[MON_ROLE_LEN]; } SMonVnodeDesc; typedef struct { int32_t vgroup_id; + char database_name[TSDB_DB_NAME_LEN]; + int32_t tables_num; + char status[MON_STATUS_LEN]; SMonVnodeDesc vnodes[TSDB_MAX_REPLICA]; } SMonVgroupDesc; typedef struct { - char database_name[TSDB_DB_NAME_LEN]; - int32_t tables_num; - int8_t status; SArray *vgroups; // array of SMonVgroupDesc } SMonVgroupInfo; @@ -107,7 +111,7 @@ typedef struct { int32_t errors; int32_t vnodes_num; int32_t masters; - int32_t has_mnode; + int8_t has_mnode; } SMonDnodeInfo; typedef struct { @@ -117,13 +121,15 @@ typedef struct { } SMonDiskDesc; typedef struct { - SArray *disks; // array of SMonDiskDesc + SArray *datadirs; // array of SMonDiskDesc + SMonDiskDesc logdir; + SMonDiskDesc tempdir; } SMonDiskInfo; typedef struct { int64_t ts; int8_t level; - char content[1024]; + char content[MON_LOG_LEN]; } SMonLogItem; typedef struct SMonInfo SMonInfo; diff --git a/source/dnode/mgmt/impl/inc/dndEnv.h b/source/dnode/mgmt/impl/inc/dndEnv.h index cbd5eb5827..724ba30155 100644 --- a/source/dnode/mgmt/impl/inc/dndEnv.h +++ b/source/dnode/mgmt/impl/inc/dndEnv.h @@ -137,6 +137,9 @@ typedef struct SDnode { SStartupReq startup; } SDnode; + +int32_t dndGetDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo); + #ifdef __cplusplus } #endif diff --git a/source/dnode/mgmt/impl/src/dndEnv.c b/source/dnode/mgmt/impl/src/dndEnv.c index 2539443b41..ae08d6387c 100644 --- a/source/dnode/mgmt/impl/src/dndEnv.c +++ b/source/dnode/mgmt/impl/src/dndEnv.c @@ -323,3 +323,5 @@ void dndCleanup() { taosStopCacheRefreshWorker(); dInfo("dnode env is cleaned up"); } + +int32_t dndGetDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo) { return 0; } \ No newline at end of file diff --git a/source/dnode/mgmt/impl/src/dndMgmt.c b/source/dnode/mgmt/impl/src/dndMgmt.c index 4bf8b82043..268ed7e362 100644 --- a/source/dnode/mgmt/impl/src/dndMgmt.c +++ b/source/dnode/mgmt/impl/src/dndMgmt.c @@ -474,21 +474,25 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) { rpcSendResponse(&rpcRsp); } -void dndGetBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { +static int32_t dndGetBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { pInfo->dnode_id = dndGetDnodeId(pDnode); tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN); + return 0; } +static int32_t dndGetDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) { return 0; } + static void dndSendMonitorReport(SDnode *pDnode) { - if (!tsEnableMonitor || tsMonitorFqdn[0] == 0) return; + if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return; + dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort); + SMonInfo *pMonitor = monCreateMonitorInfo(); if (pMonitor == NULL) return; - dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort); - SMonBasicInfo basicInfo = {0}; - dndGetBasicInfo(pDnode, &basicInfo); - monSetBasicInfo(pMonitor, &basicInfo); + if (dndGetBasicInfo(pDnode, &basicInfo) == 0) { + monSetBasicInfo(pMonitor, &basicInfo); + } SMonClusterInfo clusterInfo = {0}; SMonVgroupInfo vgroupInfo = {0}; @@ -499,6 +503,16 @@ static void dndSendMonitorReport(SDnode *pDnode) { monSetGrantInfo(pMonitor, &grantInfo); } + SMonDnodeInfo dnodeInfo = {0}; + if (dndGetDnodeInfo(pDnode, &dnodeInfo) == 0) { + monSetDnodeInfo(pMonitor, &dnodeInfo); + } + + SMonDiskInfo diskInfo = {0}; + if (dndGetDiskInfo(pDnode, &diskInfo) == 0) { + monSetDiskInfo(pMonitor, &diskInfo); + } + monSendReport(pMonitor); monCleanupMonitorInfo(pMonitor); } diff --git a/source/libs/index/inc/indexInt.h b/source/libs/index/inc/indexInt.h index 59ed5b79f5..0fcaf11087 100644 --- a/source/libs/index/inc/indexInt.h +++ b/source/libs/index/inc/indexInt.h @@ -122,30 +122,53 @@ typedef struct TFileCacheKey { int indexFlushCacheToTFile(SIndex* sIdx, void*); int32_t indexSerialCacheKey(ICacheKey* key, char* buf); +// int32_t indexSerialKey(ICacheKey* key, char* buf); +// int32_t indexSerialTermKey(SIndexTerm* itm, char* buf); -#define indexFatal(...) \ - do { \ - if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); } \ +#define indexFatal(...) \ + do { \ + if (sDebugFlag & DEBUG_FATAL) { \ + taosPrintLog("index FATAL ", 255, __VA_ARGS__); \ + } \ } while (0) -#define indexError(...) \ - do { \ - if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("index ERROR ", 255, __VA_ARGS__); } \ +#define indexError(...) \ + do { \ + if (sDebugFlag & DEBUG_ERROR) { \ + taosPrintLog("index ERROR ", 255, __VA_ARGS__); \ + } \ } while (0) -#define indexWarn(...) \ - do { \ - if (sDebugFlag & DEBUG_WARN) { taosPrintLog("index WARN ", 255, __VA_ARGS__); } \ +#define indexWarn(...) \ + do { \ + if (sDebugFlag & DEBUG_WARN) { \ + taosPrintLog("index WARN ", 255, __VA_ARGS__); \ + } \ } while (0) -#define indexInfo(...) \ - do { \ - if (sDebugFlag & DEBUG_INFO) { taosPrintLog("index ", 255, __VA_ARGS__); } \ +#define indexInfo(...) \ + do { \ + if (sDebugFlag & DEBUG_INFO) { \ + taosPrintLog("index ", 255, __VA_ARGS__); \ + } \ } while (0) -#define indexDebug(...) \ - do { \ - if (sDebugFlag & DEBUG_DEBUG) { taosPrintLog("index ", sDebugFlag, __VA_ARGS__); } \ +#define indexDebug(...) \ + do { \ + if (sDebugFlag & DEBUG_DEBUG) { \ + taosPrintLog("index ", sDebugFlag, __VA_ARGS__); \ + } \ } while (0) -#define indexTrace(...) \ - do { \ - if (sDebugFlag & DEBUG_TRACE) { taosPrintLog("index ", sDebugFlag, __VA_ARGS__); } \ +#define indexTrace(...) \ + do { \ + if (sDebugFlag & DEBUG_TRACE) { \ + taosPrintLog("index ", sDebugFlag, __VA_ARGS__); \ + } \ + } while (0) + +#define INDEX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0) +#define INDEX_TYPE_GET_TYPE(ty) (ty & 0x0F) +#define INDEX_TYPE_ADD_EXTERN_TYPE(ty, exTy) \ + do { \ + uint8_t oldTy = ty; \ + ty = (ty >> 4) | exTy; \ + ty = (ty << 4) | oldTy; \ } while (0) #ifdef __cplusplus diff --git a/source/libs/index/inc/index_comm.h b/source/libs/index/inc/index_comm.h new file mode 100644 index 0000000000..0d8418ba65 --- /dev/null +++ b/source/libs/index/inc/index_comm.h @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#ifndef _TD_INDEX_COMM_H_ +#define _TD_INDEX_COMM_H_ + +#ifdef __cplusplus +extern "C" { +#endif + +extern char JSON_COLUMN[]; +extern char JSON_VALUE_DELIM; + +char* indexPackJsonData(SIndexTerm* itm); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/source/libs/index/src/index.c b/source/libs/index/src/index.c index 5147734a85..168e819073 100644 --- a/source/libs/index/src/index.c +++ b/source/libs/index/src/index.c @@ -2,8 +2,8 @@ * Copyright (c) 2019 TAOS Data, Inc. * * This program is free software: you can use, redistribute, and/or modify - * it under the terms of the GNU Affero General Public License, version 3 - * or later ("AGPL"), as published by the Free Software Foundation. + * it under the terms of the GNU Affero General Public License, version 3 * or later ("AGPL"), as published by the Free + * Software Foundation. * * This program is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or @@ -30,6 +30,8 @@ void* indexQhandle = NULL; +static char JSON_COLUMN[] = "JSON"; + void indexInit() { // refactor later indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, INDEX_NUM_OF_THREADS, "index"); @@ -63,6 +65,9 @@ static int indexGenTFile(SIndex* index, IndexCache* cache, SArray* batch); static void indexMergeCacheAndTFile(SArray* result, IterateValue* icache, IterateValue* iTfv); static void indexMergeSameKey(SArray* result, TFileValue* tv); +// static int32_t indexSerialTermKey(SIndexTerm* itm, char* buf); +// int32_t indexSerialKey(ICacheKey* key, char* buf); + int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) { pthread_once(&isInit, indexInit); SIndex* sIdx = calloc(1, sizeof(SIndex)); @@ -148,7 +153,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) { SIndexTerm* p = taosArrayGetP(fVals, i); char buf[128] = {0}; - ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName)}; + ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName), .colType = p->colType}; int32_t sz = indexSerialCacheKey(&key, buf); IndexCache** cache = taosHashGet(index->colObj, buf, sz); @@ -163,7 +168,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) { SIndexTerm* p = taosArrayGetP(fVals, i); char buf[128] = {0}; - ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName)}; + ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName), .colType = p->colType}; int32_t sz = indexSerialCacheKey(&key, buf); IndexCache** cache = taosHashGet(index->colObj, buf, sz); @@ -330,8 +335,9 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result IndexCache* cache = NULL; char buf[128] = {0}; - ICacheKey key = {.suid = term->suid, .colName = term->colName, .nColName = strlen(term->colName)}; - int32_t sz = indexSerialCacheKey(&key, buf); + ICacheKey key = { + .suid = term->suid, .colName = term->colName, .nColName = strlen(term->colName), .colType = term->colType}; + int32_t sz = indexSerialCacheKey(&key, buf); pthread_mutex_lock(&sIdx->mtx); IndexCache** pCache = taosHashGet(sIdx->colObj, buf, sz); @@ -555,11 +561,17 @@ END: } int32_t indexSerialCacheKey(ICacheKey* key, char* buf) { + bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(key->colType, TSDB_DATA_TYPE_JSON); + char* p = buf; SERIALIZE_MEM_TO_BUF(buf, key, suid); SERIALIZE_VAR_TO_BUF(buf, '_', char); // SERIALIZE_MEM_TO_BUF(buf, key, colType); // SERIALIZE_VAR_TO_BUF(buf, '_', char); - SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName); + if (hasJson) { + SERIALIZE_STR_VAR_TO_BUF(buf, JSON_COLUMN, strlen(JSON_COLUMN)); + } else { + SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName); + } return buf - p; } diff --git a/source/libs/index/src/index_cache.c b/source/libs/index/src/index_cache.c index d6a7141825..599bac3fe6 100644 --- a/source/libs/index/src/index_cache.c +++ b/source/libs/index/src/index_cache.c @@ -14,6 +14,7 @@ */ #include "index_cache.h" +#include "index_comm.h" #include "index_util.h" #include "tcompare.h" #include "tsched.h" @@ -44,8 +45,9 @@ IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, in indexError("failed to create index cache"); return NULL; }; + cache->mem = indexInternalCacheCreate(type); - cache->colName = tstrdup(colName); + cache->colName = INDEX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? tstrdup(JSON_COLUMN) : tstrdup(colName); cache->type = type; cache->index = idx; cache->version = 0; @@ -207,11 +209,11 @@ static void indexCacheMakeRoomForWrite(IndexCache* cache) { } } } - int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) { if (cache == NULL) { return -1; } + bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON); IndexCache* pCache = cache; indexCacheRef(pCache); @@ -222,8 +224,12 @@ int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) { } // set up key ct->colType = term->colType; - ct->colVal = (char*)calloc(1, sizeof(char) * (term->nColVal + 1)); - memcpy(ct->colVal, term->colVal, term->nColVal); + if (hasJson) { + ct->colVal = indexPackJsonData(term); + } else { + ct->colVal = (char*)calloc(1, sizeof(char) * (term->nColVal + 1)); + memcpy(ct->colVal, term->colVal, term->nColVal); + } ct->version = atomic_add_fetch_32(&pCache->version, 1); // set value ct->uid = uid; @@ -294,13 +300,22 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, SArray* result, STermV SIndexTerm* term = query->term; EIndexQueryType qtype = query->qType; - CacheTerm ct = {.colVal = term->colVal, .version = atomic_load_32(&pCache->version)}; + + bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON); + char* p = term->colVal; + if (hasJson) { + p = indexPackJsonData(term); + } + CacheTerm ct = {.colVal = p, .version = atomic_load_32(&pCache->version)}; int ret = indexQueryMem(mem, &ct, qtype, result, s); if (ret == 0 && *s != kTypeDeletion) { // continue search in imm ret = indexQueryMem(imm, &ct, qtype, result, s); } + if (hasJson) { + tfree(p); + } indexMemUnRef(mem); indexMemUnRef(imm); @@ -367,6 +382,8 @@ static int32_t indexCacheTermCompare(const void* l, const void* r) { } static MemTable* indexInternalCacheCreate(int8_t type) { + type = INDEX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? TSDB_DATA_TYPE_BINARY : type; + MemTable* tbl = calloc(1, sizeof(MemTable)); indexMemRef(tbl); if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) { @@ -389,9 +406,6 @@ static bool indexCacheIteratorNext(Iterate* itera) { IterateValue* iv = &itera->val; iterateValueDestroy(iv, false); - // IterateValue* iv = &itera->val; - // IterateValue tIterVal = {.colVal = NULL, .val = taosArrayInit(1, sizeof(uint64_t))}; - bool next = tSkipListIterNext(iter); if (next) { SSkipListNode* node = tSkipListIterGet(iter); @@ -411,10 +425,6 @@ static bool indexCacheIteratorNext(Iterate* itera) { taosArrayPush(iv->val, &ct->uid); } - // IterateValue* iv = &itera->val; - // iterateValueDestroy(iv, true); - //*iv = tIterVal; - return next; } diff --git a/source/libs/index/src/index_comm.c b/source/libs/index/src/index_comm.c new file mode 100644 index 0000000000..4f3cbaa4da --- /dev/null +++ b/source/libs/index/src/index_comm.c @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 * or later ("AGPL"), as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include "index.h" +#include "indexInt.h" + +char JSON_COLUMN[] = "JSON"; +char JSON_VALUE_DELIM = '&'; + +char* indexPackJsonData(SIndexTerm* itm) { + /* + * |<-----colname---->|<-----dataType---->|<--------colVal---------->| + * |<-----string----->|<-----uint8_t----->|<----depend on dataType-->| + */ + uint8_t ty = INDEX_TYPE_GET_TYPE(itm->colType); + + int32_t sz = itm->nColName + itm->nColVal + sizeof(uint8_t) + sizeof(JSON_VALUE_DELIM) * 2 + 1; + char* buf = (char*)calloc(1, sz); + char* p = buf; + + memcpy(p, itm->colName, itm->nColName); + p += itm->nColName; + + memcpy(p, &JSON_VALUE_DELIM, sizeof(JSON_VALUE_DELIM)); + p += sizeof(JSON_VALUE_DELIM); + + memcpy(p, &ty, sizeof(ty)); + p += sizeof(ty); + + memcpy(p, &JSON_VALUE_DELIM, sizeof(JSON_VALUE_DELIM)); + p += sizeof(JSON_VALUE_DELIM); + + memcpy(p, itm->colVal, itm->nColVal); + + return buf; +} diff --git a/source/libs/index/src/index_json.c b/source/libs/index/src/index_json.c new file mode 100644 index 0000000000..de88ff3c8a --- /dev/null +++ b/source/libs/index/src/index_json.c @@ -0,0 +1,44 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 * or later ("AGPL"), as published by the Free + * Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +#include "index.h" +#include "indexInt.h" + +int tIndexJsonOpen(SIndexJsonOpts *opts, const char *path, SIndexJson **index) { + // handle + return indexOpen(opts, path, index); +} +int tIndexJsonPut(SIndexJson *index, SIndexJsonMultiTerm *terms, uint64_t uid) { + for (int i = 0; i < taosArrayGetSize(terms); i++) { + SIndexJsonTerm *p = taosArrayGetP(terms, i); + INDEX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON); + } + return indexPut(index, terms, uid); + // handle put +} + +int tIndexJsonSearch(SIndexJson *index, SIndexJsonMultiTermQuery *tq, SArray *result) { + SArray *terms = tq->query; + for (int i = 0; i < taosArrayGetSize(terms); i++) { + SIndexJsonTerm *p = taosArrayGetP(terms, i); + INDEX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON); + } + return indexSearch(index, tq, result); + // handle search +} + +void tIndexJsonClose(SIndexJson *index) { + return indexClose(index); + // handle close +} diff --git a/source/libs/index/src/index_tfile.c b/source/libs/index/src/index_tfile.c index e44f8fc1c3..0947c796b2 100644 --- a/source/libs/index/src/index_tfile.c +++ b/source/libs/index/src/index_tfile.c @@ -15,6 +15,7 @@ p * #include "index_tfile.h" #include "index.h" +#include "index_comm.h" #include "index_fst.h" #include "index_fst_counting_writer.h" #include "index_util.h" @@ -186,13 +187,20 @@ void tfileReaderDestroy(TFileReader* reader) { int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* result) { SIndexTerm* term = query->term; + bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON); EIndexQueryType qtype = query->qType; int ret = -1; // refactor to callback later if (qtype == QUERY_TERM) { uint64_t offset; - FstSlice key = fstSliceCreate(term->colVal, term->nColVal); + char* p = term->colVal; + uint64_t sz = term->nColVal; + if (hasJson) { + p = indexPackJsonData(term); + sz = strlen(p); + } + FstSlice key = fstSliceCreate(p, sz); if (fstGet(reader->fst, &key, &offset)) { indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex", term->suid, term->colName, term->colVal); @@ -202,10 +210,17 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* resul term->colVal); } fstSliceDestroy(&key); + if (hasJson) { + free(p); + } } else if (qtype == QUERY_PREFIX) { // handle later // - } else { + } else if (qtype == QUERY_SUFFIX) { + // handle later + } else if (qtype == QUERY_REGEX) { + // handle later + } else if (qtype == QUERY_RANGE) { // handle later } tfileReaderUnRef(reader); @@ -260,6 +275,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) { __compar_fn_t fn; int8_t colType = tw->header.colType; + colType = INDEX_TYPE_GET_TYPE(colType); if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) { fn = tfileStrCompare; } else { @@ -557,6 +573,8 @@ static int tfileWriteHeader(TFileWriter* writer) { static int tfileWriteData(TFileWriter* write, TFileValue* tval) { TFileHeader* header = &write->header; uint8_t colType = header->colType; + + colType = INDEX_TYPE_GET_TYPE(colType); if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) { FstSlice key = fstSliceCreate((uint8_t*)(tval->colVal), (size_t)strlen(tval->colVal)); if (fstBuilderInsert(write->fb, key, tval->offset)) { @@ -586,11 +604,10 @@ static int tfileReaderLoadHeader(TFileReader* reader) { int64_t nread = reader->ctx->readFrom(reader->ctx, buf, sizeof(buf), 0); if (nread == -1) { - indexError("actual Read: %d, to read: %d, errno: %d, filename: %s", (int)(nread), (int)sizeof(buf), - errno, reader->ctx->file.buf); + indexError("actual Read: %d, to read: %d, errno: %d, filename: %s", (int)(nread), (int)sizeof(buf), errno, + reader->ctx->file.buf); } else { - indexInfo("actual Read: %d, to read: %d, filename: %s", (int)(nread), (int)sizeof(buf), - reader->ctx->file.buf); + indexInfo("actual Read: %d, to read: %d, filename: %s", (int)(nread), (int)sizeof(buf), reader->ctx->file.buf); } // assert(nread == sizeof(buf)); memcpy(&reader->header, buf, sizeof(buf)); diff --git a/source/libs/index/test/CMakeLists.txt b/source/libs/index/test/CMakeLists.txt index bed42be3e5..1ebf85368b 100644 --- a/source/libs/index/test/CMakeLists.txt +++ b/source/libs/index/test/CMakeLists.txt @@ -2,6 +2,7 @@ add_executable(indexTest "") add_executable(fstTest "") add_executable(fstUT "") add_executable(UtilUT "") +add_executable(jsonUT "") target_sources(indexTest PRIVATE @@ -21,6 +22,10 @@ target_sources(UtilUT "utilUT.cc" ) +target_sources(jsonUT + PRIVATE + "jsonUT.cc" +) target_include_directories ( indexTest PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/index" @@ -43,6 +48,12 @@ target_include_directories ( UtilUT "${CMAKE_SOURCE_DIR}/include/libs/index" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) + +target_include_directories (jsonUT + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/index" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries (indexTest os util @@ -73,6 +84,13 @@ target_link_libraries (UtilUT index ) +target_link_libraries (jsonUT + os + util + common + gtest_main + index +) #add_test( # NAME index_test diff --git a/source/libs/index/test/fstTest.cc b/source/libs/index/test/fstTest.cc index 698945ae14..cb3206a611 100644 --- a/source/libs/index/test/fstTest.cc +++ b/source/libs/index/test/fstTest.cc @@ -301,13 +301,18 @@ void validateTFile(char* arg) { } } -void iterTFileReader(char* path, char* ver) { - int version = atoi(ver); - TFileReader* reader = tfileReaderOpen(path, 0, version, "tag1"); - Iterate* iter = tfileIteratorCreate(reader); - bool tn = iter ? iter->next(iter) : false; - int count = 0; - int termCount = 0; +void iterTFileReader(char* path, char* uid, char* colName, char* ver) { + // tfInit(); + + uint64_t suid = atoi(uid); + int version = atoi(ver); + + TFileReader* reader = tfileReaderOpen(path, suid, version, colName); + + Iterate* iter = tfileIteratorCreate(reader); + bool tn = iter ? iter->next(iter) : false; + int count = 0; + int termCount = 0; while (tn == true) { count++; IterateValue* cv = iter->getValue(iter); @@ -323,9 +328,9 @@ void iterTFileReader(char* path, char* ver) { int main(int argc, char* argv[]) { // tool to check all kind of fst test // if (argc > 1) { validateTFile(argv[1]); } - if (argc > 2) { - // opt - iterTFileReader(argv[1], argv[2]); + if (argc > 4) { + // path suid colName ver + iterTFileReader(argv[1], argv[2], argv[3], argv[4]); } // checkFstCheckIterator(); // checkFstLongTerm(); diff --git a/source/libs/index/test/fstUT.cc b/source/libs/index/test/fstUT.cc index 2d3d2f8894..6e9c88bc79 100644 --- a/source/libs/index/test/fstUT.cc +++ b/source/libs/index/test/fstUT.cc @@ -213,21 +213,21 @@ class FstEnv : public ::testing::Test { TEST_F(FstEnv, writeNormal) { fst->CreateWriter(); - std::string str("aa"); + std::string str("11"); for (int i = 0; i < 10; i++) { - str[0] = 'a' + i; + str[0] = '1' + i; str.resize(2); assert(fst->Put(str, i) == true); } // order failed - assert(fst->Put("aa", 1) == false); + assert(fst->Put("11", 1) == false); fst->DestroyWriter(); fst->CreateReader(); uint64_t val; - assert(fst->Get("a", &val) == false); - assert(fst->Get("aa", &val) == true); + assert(fst->Get("1", &val) == false); + assert(fst->Get("11", &val) == true); assert(val == 0); std::vector rlt; @@ -235,3 +235,19 @@ TEST_F(FstEnv, writeNormal) { assert(fst->Search(ctx, rlt) == true); } TEST_F(FstEnv, WriteMillonrRecord) {} +TEST_F(FstEnv, writeAbNormal) { + fst->CreateWriter(); + std::string str1("voltage&\b&ab"); + std::string str2("voltbge&\b&ab"); + + fst->Put(str1, 1); + fst->Put(str2, 2); + + fst->DestroyWriter(); + + fst->CreateReader(); + uint64_t val; + assert(fst->Get("1", &val) == false); + assert(fst->Get("voltage&\b&ab", &val) == true); + assert(val == 1); +} diff --git a/source/libs/index/test/jsonUT.cc b/source/libs/index/test/jsonUT.cc new file mode 100644 index 0000000000..e5c79d137f --- /dev/null +++ b/source/libs/index/test/jsonUT.cc @@ -0,0 +1,135 @@ +#include +#include +#include +#include +#include +#include +#include "index.h" +#include "indexInt.h" +#include "index_cache.h" +#include "index_fst.h" +#include "index_fst_counting_writer.h" +#include "index_fst_util.h" +#include "index_tfile.h" +#include "index_util.h" +#include "tglobal.h" +#include "tskiplist.h" +#include "tutil.h" + +static std::string dir = "/tmp/json"; +class JsonEnv : public ::testing::Test { + protected: + virtual void SetUp() { + taosRemoveDir(dir.c_str()); + taosMkDir(dir.c_str()); + printf("set up\n"); + opts = indexOptsCreate(); + int ret = tIndexJsonOpen(opts, dir.c_str(), &index); + assert(ret == 0); + } + virtual void TearDown() { + tIndexJsonClose(index); + indexOptsDestroy(opts); + printf("destory\n"); + } + SIndexJsonOpts* opts; + SIndexJson* index; +}; + +TEST_F(JsonEnv, testWrite) { + { + std::string colName("test"); + std::string colVal("ab"); + SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SIndexMultiTerm* terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + for (size_t i = 0; i < 100; i++) { + tIndexJsonPut(index, terms, i); + } + indexMultiTermDestroy(terms); + } + { + std::string colName("voltage"); + std::string colVal("ab1"); + SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SIndexMultiTerm* terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + for (size_t i = 0; i < 100; i++) { + tIndexJsonPut(index, terms, i); + } + indexMultiTermDestroy(terms); + } + { + std::string colName("voltage"); + std::string colVal("123"); + SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SIndexMultiTerm* terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + for (size_t i = 0; i < 100; i++) { + tIndexJsonPut(index, terms, i); + } + indexMultiTermDestroy(terms); + } + { + std::string colName("test"); + std::string colVal("ab"); + + SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); + SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SArray* result = taosArrayInit(1, sizeof(uint64_t)); + indexMultiTermQueryAdd(mq, q, QUERY_TERM); + tIndexJsonSearch(index, mq, result); + assert(100 == taosArrayGetSize(result)); + indexMultiTermQueryDestroy(mq); + } +} +TEST_F(JsonEnv, testWriteMillonData) { + { + std::string colName("test"); + std::string colVal("ab"); + SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SIndexMultiTerm* terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + for (size_t i = 0; i < 100; i++) { + tIndexJsonPut(index, terms, i); + } + indexMultiTermDestroy(terms); + } + { + std::string colName("voltagefdadfa"); + std::string colVal("abxxxxxxxxxxxx"); + SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SIndexMultiTerm* terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + for (size_t i = 0; i < 1000000; i++) { + tIndexJsonPut(index, terms, i); + } + indexMultiTermDestroy(terms); + } + { + std::string colName("test"); + std::string colVal("ab"); + + SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); + SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); + + SArray* result = taosArrayInit(1, sizeof(uint64_t)); + indexMultiTermQueryAdd(mq, q, QUERY_TERM); + tIndexJsonSearch(index, mq, result); + assert(100 == taosArrayGetSize(result)); + indexMultiTermQueryDestroy(mq); + } +} diff --git a/source/libs/monitor/src/monitor.c b/source/libs/monitor/src/monitor.c index 811ee40dc8..ecf9da218b 100644 --- a/source/libs/monitor/src/monitor.c +++ b/source/libs/monitor/src/monitor.c @@ -23,7 +23,7 @@ static SMonitor tsMonitor = {0}; int32_t monInit(const SMonCfg *pCfg) { - tsMonitor.logs = taosArrayInit(16, sizeof(SMonInfo)); + tsMonitor.logs = taosArrayInit(16, sizeof(SMonLogItem)); if (tsMonitor.logs == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -44,7 +44,7 @@ void monCleanup() { void monAddLogItem(SMonLogItem *pItem) { taosWLockLatch(&tsMonitor.lock); int32_t size = taosArrayGetSize(tsMonitor.logs); - if (size > tsMonitor.maxLogs) { + if (size >= tsMonitor.maxLogs) { uInfo("too many logs for monitor"); } else { taosArrayPush(tsMonitor.logs, pItem); @@ -54,7 +54,10 @@ void monAddLogItem(SMonLogItem *pItem) { SMonInfo *monCreateMonitorInfo() { SMonInfo *pMonitor = calloc(1, sizeof(SMonInfo)); - if (pMonitor == NULL) return NULL; + if (pMonitor == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } taosWLockLatch(&tsMonitor.lock); pMonitor->logs = taosArrayDup(tsMonitor.logs); @@ -77,41 +80,257 @@ void monCleanupMonitorInfo(SMonInfo *pMonitor) { free(pMonitor); } +void monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo) { + SJson *pJson = pMonitor->pJson; + int64_t ms = taosGetTimestampMs(); + char buf[40] = {0}; + taosFormatUtcTime(buf, sizeof(buf), ms, TSDB_TIME_PRECISION_MILLI); + + tjsonAddStringToObject(pJson, "ts", buf); + tjsonAddDoubleToObject(pJson, "dnode_id", pInfo->dnode_id); + tjsonAddStringToObject(pJson, "dnode_ep", pInfo->dnode_ep); +} + +void monSetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo) { + SJson *pJson = tjsonCreateObject(); + if (pJson == NULL) return; + if (tjsonAddItemToObject(pMonitor->pJson, "cluster_info", pJson) != 0) { + tjsonDelete(pJson); + return; + } + + tjsonAddStringToObject(pJson, "first_ep", pInfo->first_ep); + tjsonAddDoubleToObject(pJson, "first_ep_dnode_id", pInfo->first_ep_dnode_id); + tjsonAddStringToObject(pJson, "version", pInfo->version); + tjsonAddDoubleToObject(pJson, "master_uptime", pInfo->master_uptime); + tjsonAddDoubleToObject(pJson, "monitor_interval", pInfo->monitor_interval); + tjsonAddDoubleToObject(pJson, "vgroups_total", pInfo->vgroups_total); + tjsonAddDoubleToObject(pJson, "vgroups_alive", pInfo->vgroups_alive); + tjsonAddDoubleToObject(pJson, "vnodes_total", pInfo->vnodes_total); + tjsonAddDoubleToObject(pJson, "vnodes_alive", pInfo->vnodes_alive); + tjsonAddDoubleToObject(pJson, "connections_total", pInfo->connections_total); + + SJson *pDnodesJson = tjsonAddArrayToObject(pJson, "dnodes"); + if (pDnodesJson == NULL) return; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->dnodes); ++i) { + SJson *pDnodeJson = tjsonCreateObject(); + if (pDnodeJson == NULL) continue; + + SMonDnodeDesc *pDnodeDesc = taosArrayGet(pInfo->dnodes, i); + tjsonAddDoubleToObject(pDnodeJson, "dnode_id", pDnodeDesc->dnode_id); + tjsonAddStringToObject(pDnodeJson, "dnode_ep", pDnodeDesc->dnode_ep); + tjsonAddStringToObject(pDnodeJson, "status", pDnodeDesc->status); + + if (tjsonAddItemToArray(pDnodesJson, pDnodeJson) != 0) tjsonDelete(pDnodeJson); + } + + SJson *pMnodesJson = tjsonAddArrayToObject(pJson, "mnodes"); + if (pMnodesJson == NULL) return; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->dnodes); ++i) { + SJson *pMnodeJson = tjsonCreateObject(); + if (pMnodeJson == NULL) continue; + + SMonMnodeDesc *pMnodeDesc = taosArrayGet(pInfo->dnodes, i); + tjsonAddDoubleToObject(pMnodeJson, "mnode_id", pMnodeDesc->mnode_id); + tjsonAddStringToObject(pMnodeJson, "mnode_ep", pMnodeDesc->mnode_ep); + tjsonAddStringToObject(pMnodeJson, "role", pMnodeDesc->role); + + if (tjsonAddItemToArray(pMnodesJson, pMnodeJson) != 0) tjsonDelete(pMnodeJson); + } +} + +void monSetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo) { + SJson *pJson = tjsonAddArrayToObject(pMonitor->pJson, "vgroup_infos"); + if (pJson == NULL) return; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->vgroups); ++i) { + SJson *pVgroupJson = tjsonCreateObject(); + if (pVgroupJson == NULL) continue; + if (tjsonAddItemToArray(pJson, pVgroupJson) != 0) { + tjsonDelete(pVgroupJson); + continue; + } + + SMonVgroupDesc *pVgroupDesc = taosArrayGet(pInfo->vgroups, i); + tjsonAddDoubleToObject(pVgroupJson, "vgroup_id", pVgroupDesc->vgroup_id); + tjsonAddStringToObject(pVgroupJson, "database_name", pVgroupDesc->database_name); + tjsonAddDoubleToObject(pVgroupJson, "tables_num", pVgroupDesc->tables_num); + tjsonAddStringToObject(pVgroupJson, "status", pVgroupDesc->status); + + SJson *pVnodesJson = tjsonAddArrayToObject(pVgroupJson, "vnodes"); + if (pVnodesJson == NULL) continue; + + for (int32_t j = 0; j < TSDB_MAX_REPLICA; ++j) { + SMonVnodeDesc *pVnodeDesc = &pVgroupDesc->vnodes[j]; + if (pVnodeDesc->dnode_id <= 0) continue; + + SJson *pVnodeJson = tjsonCreateObject(); + if (pVnodeJson == NULL) continue; + + tjsonAddDoubleToObject(pVnodeJson, "dnode_id", pVnodeDesc->dnode_id); + tjsonAddStringToObject(pVnodeJson, "vnode_role", pVnodeDesc->vnode_role); + + if (tjsonAddItemToArray(pVnodesJson, pVnodeJson) != 0) tjsonDelete(pVnodeJson); + } + } +} + +void monSetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo) { + SJson *pJson = tjsonCreateObject(); + if (pJson == NULL) return; + if (tjsonAddItemToObject(pMonitor->pJson, "grant_info", pJson) != 0) { + tjsonDelete(pJson); + return; + } + + tjsonAddDoubleToObject(pJson, "expire_time", pInfo->expire_time); + tjsonAddDoubleToObject(pJson, "timeseries_used", pInfo->timeseries_used); + tjsonAddDoubleToObject(pJson, "timeseries_total", pInfo->timeseries_total); +} + +void monSetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo) { + SJson *pJson = tjsonCreateObject(); + if (pJson == NULL) return; + if (tjsonAddItemToObject(pMonitor->pJson, "dnode_info", pJson) != 0) { + tjsonDelete(pJson); + return; + } + + tjsonAddDoubleToObject(pJson, "uptime", pInfo->uptime); + tjsonAddDoubleToObject(pJson, "cpu_engine", pInfo->cpu_engine); + tjsonAddDoubleToObject(pJson, "cpu_system", pInfo->cpu_system); + tjsonAddDoubleToObject(pJson, "cpu_cores", pInfo->cpu_cores); + tjsonAddDoubleToObject(pJson, "mem_engine", pInfo->mem_engine); + tjsonAddDoubleToObject(pJson, "mem_system", pInfo->mem_system); + tjsonAddDoubleToObject(pJson, "mem_total", pInfo->mem_total); + tjsonAddDoubleToObject(pJson, "disk_engine", pInfo->disk_engine); + tjsonAddDoubleToObject(pJson, "disk_used", pInfo->disk_used); + tjsonAddDoubleToObject(pJson, "disk_total", pInfo->disk_total); + tjsonAddDoubleToObject(pJson, "net_in", pInfo->net_in); + tjsonAddDoubleToObject(pJson, "net_out", pInfo->net_out); + tjsonAddDoubleToObject(pJson, "io_read", pInfo->io_read); + tjsonAddDoubleToObject(pJson, "io_write", pInfo->io_write); + tjsonAddDoubleToObject(pJson, "io_read_disk", pInfo->io_read_disk); + tjsonAddDoubleToObject(pJson, "io_write_disk", pInfo->io_write_disk); + tjsonAddDoubleToObject(pJson, "req_select", pInfo->req_select); + tjsonAddDoubleToObject(pJson, "req_select_rate", pInfo->req_select_rate); + tjsonAddDoubleToObject(pJson, "req_insert", pInfo->req_insert); + tjsonAddDoubleToObject(pJson, "req_insert_success", pInfo->req_insert_success); + tjsonAddDoubleToObject(pJson, "req_insert_rate", pInfo->req_insert_rate); + tjsonAddDoubleToObject(pJson, "req_insert_batch", pInfo->req_insert_batch); + tjsonAddDoubleToObject(pJson, "req_insert_batch_success", pInfo->req_insert_batch_success); + tjsonAddDoubleToObject(pJson, "req_insert_batch_rate", pInfo->req_insert_batch_rate); + tjsonAddDoubleToObject(pJson, "errors", pInfo->errors); + tjsonAddDoubleToObject(pJson, "vnodes_num", pInfo->vnodes_num); + tjsonAddDoubleToObject(pJson, "masters", pInfo->masters); + tjsonAddDoubleToObject(pJson, "has_mnode", pInfo->has_mnode); +} + +void monSetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo) { + SJson *pJson = tjsonCreateObject(); + if (pJson == NULL) return; + if (tjsonAddItemToObject(pMonitor->pJson, "disk_infos", pJson) != 0) { + tjsonDelete(pJson); + return; + } + + SJson *pDatadirsJson = tjsonAddArrayToObject(pJson, "datadir"); + if (pDatadirsJson == NULL) return; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->datadirs); ++i) { + SJson *pDatadirJson = tjsonCreateObject(); + if (pDatadirJson == NULL) continue; + + SMonDiskDesc *pDatadirDesc = taosArrayGet(pInfo->datadirs, i); + if (tjsonAddStringToObject(pDatadirJson, "name", pDatadirDesc->name) != 0) tjsonDelete(pDatadirJson); + if (tjsonAddDoubleToObject(pDatadirJson, "level", pDatadirDesc->level) != 0) tjsonDelete(pDatadirJson); + if (tjsonAddDoubleToObject(pDatadirJson, "avail", pDatadirDesc->size.avail) != 0) tjsonDelete(pDatadirJson); + if (tjsonAddDoubleToObject(pDatadirJson, "used", pDatadirDesc->size.used) != 0) tjsonDelete(pDatadirJson); + if (tjsonAddDoubleToObject(pDatadirJson, "total", pDatadirDesc->size.total) != 0) tjsonDelete(pDatadirJson); + + if (tjsonAddItemToArray(pDatadirsJson, pDatadirJson) != 0) tjsonDelete(pDatadirJson); + } + + SJson *pLogdirJson = tjsonCreateObject(); + if (pLogdirJson == NULL) return; + if (tjsonAddItemToObject(pJson, "logdir", pLogdirJson) != 0) return; + tjsonAddStringToObject(pLogdirJson, "name", pInfo->logdir.name); + tjsonAddDoubleToObject(pLogdirJson, "avail", pInfo->logdir.size.avail); + tjsonAddDoubleToObject(pLogdirJson, "used", pInfo->logdir.size.used); + tjsonAddDoubleToObject(pLogdirJson, "total", pInfo->logdir.size.total); + + SJson *pTempdirJson = tjsonCreateObject(); + if (pTempdirJson == NULL) return; + if (tjsonAddItemToObject(pJson, "tempdir", pTempdirJson) != 0) return; + tjsonAddStringToObject(pTempdirJson, "name", pInfo->tempdir.name); + tjsonAddDoubleToObject(pTempdirJson, "avail", pInfo->tempdir.size.avail); + tjsonAddDoubleToObject(pTempdirJson, "used", pInfo->tempdir.size.used); + tjsonAddDoubleToObject(pTempdirJson, "total", pInfo->tempdir.size.total); +} + +static void monSetLogInfo(SMonInfo *pMonitor) { + SJson *pJson = tjsonCreateObject(); + if (pJson == NULL) return; + if (tjsonAddItemToObject(pMonitor->pJson, "log_infos", pJson) != 0) { + tjsonDelete(pJson); + return; + } + + SJson *pLogsJson = tjsonAddArrayToObject(pJson, "logs"); + if (pLogsJson == NULL) return; + + for (int32_t i = 0; i < taosArrayGetSize(pMonitor->logs); ++i) { + SJson *pLogJson = tjsonCreateObject(); + if (pLogJson == NULL) continue; + + SMonLogItem *pLogItem = taosArrayGet(pMonitor->logs, i); + + char buf[40] = {0}; + taosFormatUtcTime(buf, sizeof(buf), pLogItem->ts, TSDB_TIME_PRECISION_MILLI); + + tjsonAddStringToObject(pLogJson, "ts", buf); + tjsonAddDoubleToObject(pLogJson, "level", pLogItem->level); + tjsonAddStringToObject(pLogJson, "content", pLogItem->content); + + if (tjsonAddItemToArray(pLogsJson, pLogJson) != 0) tjsonDelete(pLogJson); + } + + SJson *pSummaryJson = tjsonAddArrayToObject(pJson, "summary"); + if (pSummaryJson == NULL) return; + + SJson *pLogError = tjsonCreateObject(); + if (pLogError == NULL) return; + tjsonAddStringToObject(pLogError, "level", "error"); + tjsonAddDoubleToObject(pLogError, "total", 1); + if (tjsonAddItemToArray(pSummaryJson, pLogError) != 0) tjsonDelete(pLogError); + + SJson *pLogInfo = tjsonCreateObject(); + if (pLogInfo == NULL) return; + tjsonAddStringToObject(pLogInfo, "level", "info"); + tjsonAddDoubleToObject(pLogInfo, "total", 1); + if (tjsonAddItemToArray(pSummaryJson, pLogInfo) != 0) tjsonDelete(pLogInfo); + + SJson *pLogDebug = tjsonCreateObject(); + if (pLogDebug == NULL) return; + tjsonAddStringToObject(pLogDebug, "level", "debug"); + tjsonAddDoubleToObject(pLogDebug, "total", 1); + if (tjsonAddItemToArray(pSummaryJson, pLogDebug) != 0) tjsonDelete(pLogDebug); + + SJson *pLogTrace = tjsonCreateObject(); + if (pLogTrace == NULL) return; + tjsonAddStringToObject(pLogTrace, "level", "trace"); + tjsonAddDoubleToObject(pLogTrace, "total", 1); + if (tjsonAddItemToArray(pSummaryJson, pLogTrace) != 0) tjsonDelete(pLogTrace); +} + void monSendReport(SMonInfo *pMonitor) { + monSetLogInfo(pMonitor); + char *pCont = tjsonToString(pMonitor->pJson); if (pCont != NULL) { taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont)); free(pCont); } } - -void monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo) { - SJson *pJson = pMonitor->pJson; - tjsonAddDoubleToObject(pJson, "dnode_id", pInfo->dnode_id); - tjsonAddStringToObject(pJson, "dnode_ep", pInfo->dnode_ep); - - int64_t ms = taosGetTimestampMs(); - char buf[40] = {0}; - taosFormatUtcTime(buf, sizeof(buf), ms, TSDB_TIME_PRECISION_MILLI); - tjsonAddStringToObject(pJson, "ts", buf); -} - -void monSetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo) { - -} - -void monSetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo) { - -} - -void monSetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo) { - -} - -void monSetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo) { - -} - -void monSetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo) { - -} diff --git a/source/libs/monitor/test/monTest.cpp b/source/libs/monitor/test/monTest.cpp index a1805d0c9c..ad48ed5407 100644 --- a/source/libs/monitor/test/monTest.cpp +++ b/source/libs/monitor/test/monTest.cpp @@ -13,21 +13,262 @@ #include "os.h" #include "monitor.h" +#include "tglobal.h" class MonitorTest : public ::testing::Test { protected: - static void SetUpTestSuite() { root = "/tmp/monTest"; } - static void TearDownTestSuite() {} + static void SetUpTestSuite() { + SMonCfg cfg; + cfg.maxLogs = 2; + cfg.port = 80; + cfg.server = "localhost"; + monInit(&cfg); + } + + static void TearDownTestSuite() { monCleanup(); } public: void SetUp() override {} void TearDown() override {} - static const char *root; + void GetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo); + void GetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo); + void GetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo); + void GetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo); + void GetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo); + void GetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo); + void AddLogInfo1(); + void AddLogInfo2(); }; -const char *MonitorTest::root; - -TEST_F(MonitorTest, 01_Open_Close) { - +void MonitorTest::GetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo) { + pInfo->dnode_id = 1; + strcpy(pInfo->dnode_ep, "localhost"); +} + +void MonitorTest::GetClusterInfo(SMonInfo *pMonitor, SMonClusterInfo *pInfo) { + strcpy(pInfo->first_ep, "localhost:6030"); + pInfo->first_ep_dnode_id = 1; + strcpy(pInfo->version, "3.0.0.0"); + pInfo->master_uptime = 1; + pInfo->monitor_interval = 2; + pInfo->vgroups_total = 3; + pInfo->vgroups_alive = 43; + pInfo->vnodes_total = 5; + pInfo->vnodes_alive = 6; + pInfo->connections_total = 7; + + pInfo->dnodes = taosArrayInit(4, sizeof(SMonDnodeDesc)); + SMonDnodeDesc d1 = {0}; + d1.dnode_id = 1; + strcpy(d1.dnode_ep, "localhost:6030"); + strcpy(d1.status, "ready"); + taosArrayPush(pInfo->dnodes, &d1); + SMonDnodeDesc d2 = {0}; + d2.dnode_id = 2; + strcpy(d2.dnode_ep, "localhost:7030"); + strcpy(d2.status, "offline"); + taosArrayPush(pInfo->dnodes, &d2); + + pInfo->mnodes = taosArrayInit(4, sizeof(SMonMnodeDesc)); + SMonMnodeDesc m1 = {0}; + m1.mnode_id = 1; + strcpy(m1.mnode_ep, "localhost:6030"); + strcpy(m1.role, "master"); + taosArrayPush(pInfo->mnodes, &m1); + SMonMnodeDesc m2 = {0}; + m2.mnode_id = 2; + strcpy(m2.mnode_ep, "localhost:7030"); + strcpy(m2.role, "unsynced"); + taosArrayPush(pInfo->mnodes, &m2); +} + +void MonitorTest::GetVgroupInfo(SMonInfo *pMonitor, SMonVgroupInfo *pInfo) { + pInfo->vgroups = taosArrayInit(4, sizeof(SMonVgroupDesc)); + + SMonVgroupDesc vg1 = {0}; + vg1.vgroup_id = 1; + strcpy(vg1.database_name, "d1"); + vg1.tables_num = 4; + strcpy(vg1.status, "ready"); + vg1.vnodes[0].dnode_id = 1; + strcpy(vg1.vnodes[0].vnode_role, "master"); + vg1.vnodes[1].dnode_id = 2; + strcpy(vg1.vnodes[1].vnode_role, "slave"); + taosArrayPush(pInfo->vgroups, &vg1); + + SMonVgroupDesc vg2 = {0}; + vg2.vgroup_id = 2; + strcpy(vg2.database_name, "d2"); + vg2.tables_num = 5; + strcpy(vg2.status, "offline"); + vg2.vnodes[0].dnode_id = 1; + strcpy(vg2.vnodes[0].vnode_role, "master"); + vg2.vnodes[1].dnode_id = 2; + strcpy(vg2.vnodes[1].vnode_role, "unsynced"); + taosArrayPush(pInfo->vgroups, &vg2); + + SMonVgroupDesc vg3 = {0}; + vg3.vgroup_id = 3; + strcpy(vg3.database_name, "d3"); + vg3.tables_num = 6; + strcpy(vg3.status, "ready"); + vg3.vnodes[0].dnode_id = 1; + strcpy(vg3.vnodes[0].vnode_role, "master"); + taosArrayPush(pInfo->vgroups, &vg3); +} + +void MonitorTest::GetGrantInfo(SMonInfo *pMonitor, SMonGrantInfo *pInfo) { + pInfo->expire_time = 1234567; + pInfo->timeseries_total = 234567; + pInfo->timeseries_used = 34567; +} + +void MonitorTest::GetDnodeInfo(SMonInfo *pMonitor, SMonDnodeInfo *pInfo) { + pInfo->uptime = 1.2; + pInfo->cpu_engine = 2.1; + pInfo->cpu_system = 2.1; + pInfo->cpu_cores = 2; + pInfo->mem_engine = 3.1; + pInfo->mem_system = 3.2; + pInfo->mem_total = 3.3; + pInfo->disk_engine = 4.1; + pInfo->disk_used = 4.2; + pInfo->disk_total = 4.3; + pInfo->net_in = 5.1; + pInfo->net_out = 5.2; + pInfo->io_read = 6.1; + pInfo->io_write = 6.2; + pInfo->io_read_disk = 7.1; + pInfo->io_write_disk = 7.2; + pInfo->req_select = 8; + pInfo->req_select_rate = 8.1; + pInfo->req_insert = 9; + pInfo->req_insert_success = 10; + pInfo->req_insert_rate = 10.1; + pInfo->req_insert_batch = 11; + pInfo->req_insert_batch_success = 12; + pInfo->req_insert_batch_rate = 12.3; + pInfo->errors = 4; + pInfo->vnodes_num = 5; + pInfo->masters = 6; + pInfo->has_mnode = 1; +} + +void MonitorTest::GetDiskInfo(SMonInfo *pMonitor, SMonDiskInfo *pInfo) { + pInfo->datadirs = taosArrayInit(2, sizeof(SMonDiskDesc)); + SMonDiskDesc d1 = {0}; + strcpy(d1.name, "/t1/d1/d"); + d1.level = 0; + d1.size.avail = 11; + d1.size.total = 12; + d1.size.used = 13; + taosArrayPush(pInfo->datadirs, &d1); + + SMonDiskDesc d2 = {0}; + strcpy(d2.name, "/t2d2/d"); + d2.level = 2; + d2.size.avail = 21; + d2.size.total = 22; + d2.size.used = 23; + taosArrayPush(pInfo->datadirs, &d2); + + SMonDiskDesc d3 = {0}; + strcpy(d3.name, "/t3/d3/d"); + d3.level = 3; + d3.size.avail = 31; + d3.size.total = 32; + d3.size.used = 33; + taosArrayPush(pInfo->datadirs, &d3); + + strcpy(pInfo->logdir.name, "/log/dir/d"); + pInfo->logdir.size.avail = 41; + pInfo->logdir.size.total = 42; + pInfo->logdir.size.used = 43; + + strcpy(pInfo->tempdir.name, "/data/dir/d"); + pInfo->tempdir.size.avail = 51; + pInfo->tempdir.size.total = 52; + pInfo->tempdir.size.used = 53; +} + +void MonitorTest::AddLogInfo1() { + SMonLogItem log1 = {0}; + log1.ts = taosGetTimestampMs(); + log1.level = 1; + strcpy(log1.content, "1 -------------------------- a"); + monAddLogItem(&log1); + + SMonLogItem log2 = {0}; + log2.ts = taosGetTimestampMs(); + log2.level = 1; + strcpy(log2.content, "1 ------------------------ b"); + monAddLogItem(&log2); + + SMonLogItem log3 = {0}; + log3.ts = taosGetTimestampMs(); + log3.level = 1; + strcpy(log3.content, "1 ------- c"); + monAddLogItem(&log3); +} + +void MonitorTest::AddLogInfo2() { + SMonLogItem log1; + log1.ts = taosGetTimestampMs(); + log1.level = 01; + strcpy(log1.content, "2 ------- a"); + monAddLogItem(&log1); + + SMonLogItem log2; + log2.ts = taosGetTimestampMs(); + log2.level = 0; + strcpy(log2.content, "2 ------- b"); + monAddLogItem(&log2); +} + +TEST_F(MonitorTest, 01_Full) { + AddLogInfo1(); + + SMonInfo *pMonitor = monCreateMonitorInfo(); + if (pMonitor == NULL) return; + + SMonBasicInfo basicInfo = {0}; + GetBasicInfo(pMonitor, &basicInfo); + monSetBasicInfo(pMonitor, &basicInfo); + + SMonClusterInfo clusterInfo = {0}; + SMonVgroupInfo vgroupInfo = {0}; + SMonGrantInfo grantInfo = {0}; + GetClusterInfo(pMonitor, &clusterInfo); + GetVgroupInfo(pMonitor, &vgroupInfo); + GetGrantInfo(pMonitor, &grantInfo); + monSetClusterInfo(pMonitor, &clusterInfo); + monSetVgroupInfo(pMonitor, &vgroupInfo); + monSetGrantInfo(pMonitor, &grantInfo); + + SMonDnodeInfo dnodeInfo = {0}; + GetDnodeInfo(pMonitor, &dnodeInfo); + monSetDnodeInfo(pMonitor, &dnodeInfo); + + SMonDiskInfo diskInfo = {0}; + GetDiskInfo(pMonitor, &diskInfo); + monSetDiskInfo(pMonitor, &diskInfo); + + monSendReport(pMonitor); + monCleanupMonitorInfo(pMonitor); + + taosArrayDestroy(clusterInfo.dnodes); + taosArrayDestroy(clusterInfo.mnodes); + taosArrayDestroy(vgroupInfo.vgroups); + taosArrayDestroy(diskInfo.datadirs); +} + +TEST_F(MonitorTest, 02_Log) { + AddLogInfo2(); + + SMonInfo *pMonitor = monCreateMonitorInfo(); + if (pMonitor == NULL) return; + + monSendReport(pMonitor); + monCleanupMonitorInfo(pMonitor); } diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index f0db054797..c7b6ca2a2c 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -286,15 +286,17 @@ void uvOnWriteCb(uv_write_t* req, int status) { transClearBuffer(&conn->readBuf); if (status == 0) { tTrace("server conn %p data already was written on stream", conn); - assert(taosArrayGetSize(conn->srvMsgs) >= 1); - SSrvMsg* msg = taosArrayGetP(conn->srvMsgs, 0); - taosArrayRemove(conn->srvMsgs, 0); - destroySmsg(msg); + if (conn->srvMsgs != NULL) { + assert(taosArrayGetSize(conn->srvMsgs) >= 1); + SSrvMsg* msg = taosArrayGetP(conn->srvMsgs, 0); + taosArrayRemove(conn->srvMsgs, 0); + destroySmsg(msg); - // send second data, just use for push - if (taosArrayGetSize(conn->srvMsgs) > 0) { - msg = (SSrvMsg*)taosArrayGetP(conn->srvMsgs, 0); - uvStartSendRespInternal(msg); + // send second data, just use for push + if (taosArrayGetSize(conn->srvMsgs) > 0) { + msg = (SSrvMsg*)taosArrayGetP(conn->srvMsgs, 0); + uvStartSendRespInternal(msg); + } } } else { tError("server conn %p failed to write data, %s", conn, uv_err_name(status)); @@ -615,7 +617,7 @@ static void destroyConn(SSrvConn* conn, bool clear) { SSrvMsg* msg = taosArrayGetP(conn->srvMsgs, i); destroySmsg(msg); } - taosArrayDestroy(conn->srvMsgs); + conn->srvMsgs = taosArrayDestroy(conn->srvMsgs); QUEUE_REMOVE(&conn->queue); if (clear) {