From d9af07de6fe590e7960f1a8dddc5890b7ffd3b2b Mon Sep 17 00:00:00 2001 From: dmchen Date: Wed, 24 Jan 2024 09:44:44 +0000 Subject: [PATCH] new table structure --- include/common/tglobal.h | 3 + include/libs/monitor/monitor.h | 7 +- include/libs/monitorfw/taos_collector.h | 2 + .../libs/monitorfw/taos_collector_registry.h | 6 +- include/libs/monitorfw/taos_gauge.h | 154 ++++ include/libs/monitorfw/taos_metric_sample.h | 2 + include/libs/monitorfw/taos_monitor.h | 2 + include/libs/monitorfw/taos_monitor_util.h | 31 + include/util/tdef.h | 6 + source/client/src/clientEnv.c | 5 + source/client/src/selectMonitor.c | 2 +- source/client/src/slowQueryMonitor.c | 2 +- source/common/src/tglobal.c | 14 +- source/dnode/mgmt/mgmt_dnode/inc/dmInt.h | 1 + source/dnode/mgmt/mgmt_dnode/src/dmInt.c | 1 + source/dnode/mgmt/mgmt_dnode/src/dmWorker.c | 11 + source/dnode/mgmt/node_mgmt/inc/dmMgmt.h | 1 + source/dnode/mgmt/node_mgmt/src/dmEnv.c | 1 + source/dnode/mgmt/node_mgmt/src/dmMonitor.c | 18 +- source/dnode/mgmt/node_util/inc/dmUtil.h | 1 + source/dnode/mnode/impl/src/mndDnode.c | 124 +++- source/dnode/mnode/impl/src/mndMain.c | 3 + source/dnode/vnode/src/inc/vnodeInt.h | 24 + source/dnode/vnode/src/vnd/vnodeOpen.c | 22 + source/dnode/vnode/src/vnd/vnodeSvr.c | 69 +- source/libs/monitor/inc/monInt.h | 15 + source/libs/monitor/src/clientMonitor.c | 7 +- source/libs/monitor/src/monFramework.c | 656 ++++++++++++++++++ source/libs/monitor/src/monMain.c | 118 ++-- source/libs/monitor/test/monTest.cpp | 4 +- .../monitorfw/inc/taos_collector_registry_t.h | 2 +- .../inc/taos_metric_formatter_custom_i.h | 31 + .../libs/monitorfw/inc/taos_metric_sample_t.h | 8 +- source/libs/monitorfw/inc/taos_metric_t.h | 2 +- .../libs/monitorfw/inc/taos_monitor_util_i.h | 26 + source/libs/monitorfw/src/taos_collector.c | 6 + .../monitorfw/src/taos_collector_registry.c | 89 ++- source/libs/monitorfw/src/taos_gauge.c | 100 +++ source/libs/monitorfw/src/taos_metric.c | 9 +- .../monitorfw/src/taos_metric_formatter.c | 2 +- .../src/taos_metric_formatter_custom.c | 234 +++++++ .../libs/monitorfw/src/taos_metric_sample.c | 53 +- source/libs/monitorfw/src/taos_monitor_util.c | 110 +++ 43 files changed, 1878 insertions(+), 106 deletions(-) create mode 100644 include/libs/monitorfw/taos_gauge.h create mode 100644 include/libs/monitorfw/taos_monitor_util.h create mode 100644 source/libs/monitor/src/monFramework.c create mode 100644 source/libs/monitorfw/inc/taos_metric_formatter_custom_i.h create mode 100644 source/libs/monitorfw/inc/taos_monitor_util_i.h create mode 100644 source/libs/monitorfw/src/taos_gauge.c create mode 100644 source/libs/monitorfw/src/taos_metric_formatter_custom.c create mode 100644 source/libs/monitorfw/src/taos_monitor_util.c diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 3a7ff40125..9b8d63b327 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -105,6 +105,9 @@ extern char tsMonitorFqdn[]; extern uint16_t tsMonitorPort; extern int32_t tsMonitorMaxLogs; extern bool tsMonitorComp; +extern bool tsMonitorLogProtocol; +extern int32_t tsMonitorIntervalForBasic; +extern bool tsMonitorForceV2; // audit extern bool tsEnableAudit; diff --git a/include/libs/monitor/monitor.h b/include/libs/monitor/monitor.h index 1f3d21777d..4c6ffa0add 100644 --- a/include/libs/monitor/monitor.h +++ b/include/libs/monitor/monitor.h @@ -100,6 +100,7 @@ typedef struct { int32_t mnode_id; char mnode_ep[TSDB_EP_LEN]; char role[MON_ROLE_LEN]; + int32_t syncState; } SMonMnodeDesc; typedef struct { @@ -125,6 +126,7 @@ typedef struct { typedef struct { int32_t dnode_id; char vnode_role[MON_ROLE_LEN]; + int32_t syncState; } SMonVnodeDesc; typedef struct { @@ -221,9 +223,8 @@ void monSetVmInfo(SMonVmInfo *pInfo); void monSetQmInfo(SMonQmInfo *pInfo); void monSetSmInfo(SMonSmInfo *pInfo); void monSetBmInfo(SMonBmInfo *pInfo); -void monSendReport(); -void monSendPromReport(); -void monSendContent(char *pCont); +void monGenAndSendReport(); +void monGenAndSendReportBasic(); void tFreeSMonMmInfo(SMonMmInfo *pInfo); void tFreeSMonVmInfo(SMonVmInfo *pInfo); diff --git a/include/libs/monitorfw/taos_collector.h b/include/libs/monitorfw/taos_collector.h index 918395ae72..8fe304ed7d 100644 --- a/include/libs/monitorfw/taos_collector.h +++ b/include/libs/monitorfw/taos_collector.h @@ -76,6 +76,8 @@ int taos_collector_destroy_generic(void *gen); */ int taos_collector_add_metric(taos_collector_t *self, taos_metric_t *metric); +taos_metric_t* taos_collector_get_metric(taos_collector_t *self, char *metric_name); + /** * @brief The collect function is responsible for doing any work involving a set of metrics and then returning them * for metric exposition. diff --git a/include/libs/monitorfw/taos_collector_registry.h b/include/libs/monitorfw/taos_collector_registry.h index 915b132b11..063e8afb50 100644 --- a/include/libs/monitorfw/taos_collector_registry.h +++ b/include/libs/monitorfw/taos_collector_registry.h @@ -86,6 +86,8 @@ taos_metric_t *taos_collector_registry_must_register_metric(taos_metric_t *metri */ int taos_collector_registry_register_metric(taos_metric_t *metric); +taos_metric_t *taos_collector_registry_get_metric(char* metric_name); + /** * @brief Register a collector with the given registry. Returns a non-zero integer value on failure. * @param self The target taos_collector_registry_t* @@ -105,7 +107,9 @@ int taos_collector_registry_register_collector(taos_collector_registry_t *self, */ const char *taos_collector_registry_bridge(taos_collector_registry_t *self, char *ts, char *format); -int taos_collector_registry_clear_out(taos_collector_registry_t *self); +int taos_collector_registry_clear_batch(taos_collector_registry_t *self); + +const char *taos_collector_registry_bridge_new(taos_collector_registry_t *self, char *ts, char *format, char** prom_str); /** *@brief Validates that the given metric name complies with the specification: diff --git a/include/libs/monitorfw/taos_gauge.h b/include/libs/monitorfw/taos_gauge.h new file mode 100644 index 0000000000..a33661dfe4 --- /dev/null +++ b/include/libs/monitorfw/taos_gauge.h @@ -0,0 +1,154 @@ +/* + * 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 . + */ + +/** + * @file taos_gauge.h + * @brief https://prometheus.io/docs/concepts/metric_types/#gauge + */ + +#ifndef TAOS_GAUGE_H +#define TAOS_GAUGE_H + +#include + +#include "taos_metric.h" + +/** + * @brief A prometheus gauge. + * + * References + * * See https://prometheus.io/docs/concepts/metric_types/#gauge + */ +typedef taos_metric_t taos_gauge_t; + +/** + * @brief Constructs a taos_gauge_t* + * @param name The name of the metric + * @param help The metric description + * @param label_key_count The number of labels associated with the given metric. Pass 0 if the metric does not + * require labels. + * @param label_keys A collection of label keys. The number of keys MUST match the value passed as label_key_count. If + * no labels are required, pass NULL. Otherwise, it may be convenient to pass this value as a + * literal. + * @return The constructed taos_guage_t* + * + * // An example with labels + * taos_gauge_new("foo", "foo is a gauge with labels", 2, (const char**) { "one", "two" }); + * + * // An example without labels + * taos_gauge_new("foo", "foo is a gauge without labels", 0, NULL); + */ +taos_gauge_t *taos_gauge_new(const char *name, const char *help, size_t label_key_count, const char **label_keys); + +/** + * @brief Destroys a taos_gauge_t*. You must set self to NULL after destruction. A non-zero integer value will be + * returned on failure. + * @param self The target taos_gauge_t* + * @return A non-zero integer value upon failure + */ +int taos_gauge_destroy(taos_gauge_t *self); + +/** + * @brief Increment the taos_gauge_t* by 1. + * @param self The target taos_gauger_t* + * @param label_values The label values associated with the metric sample being updated. The number of labels must + * match the value passed to label_key_count in the gauge's constructor. If no label values are + * necessary, pass NULL. Otherwise, It may be convenient to pass this value as a literal. + * @return A non-zero integer value upon failure + * *Example* + * + * // An example with labels + * taos_gauge_inc(foo_gauge, (const char**) { "bar", "bang" }); + * + * // An example without labels + * taos_gauge_inc(foo_gauge, NULL); + */ +int taos_gauge_inc(taos_gauge_t *self, const char **label_values); + +/** + * @brief Decrement the taos_gauge_t* by 1. + * @param self The target taos_gauger_t* + * @param label_values The label values associated with the metric sample being updated. The number of labels must + * match the value passed to label_key_count in the gauge's constructor. If no label values are + * necessary, pass NULL. Otherwise, It may be convenient to pass this value as a literal. + * @return A non-zero integer value upon failure. + * *Example* + * + * // An example with labels + * taos_gauge_dec(foo_gauge, (const char**) { "bar", "bang" }); + * + * // An example without labels + * taos_gauge_dec(foo_gauge, NULL); + */ +int taos_gauge_dec(taos_gauge_t *self, const char **label_values); + +/** + * @brief Add the value to the taos_gauge_t*. + * @param self The target taos_gauge_t* + * @param r_value The double to add to the taos_gauge_t passed as self. + * @param label_values The label values associated with the metric sample being updated. The number of labels must + * match the value passed to label_key_count in the gauge's constructor. If no label values are + * necessary, pass NULL. Otherwise, It may be convenient to pass this value as a literal. + * @return A non-zero integer value upon failure. + * + * *Example* + * + * // An example with labels + * taos_gauge_add(foo_gauge 22, (const char**) { "bar", "bang" }); + * + * // An example without labels + * taos_gauge_add(foo_gauge, 22, NULL); + */ +int taos_gauge_add(taos_gauge_t *self, double r_value, const char **label_values); + +/** + * @brief Subtract the value to the taos_gauge. A non-zero integer value will be returned on failure. + * @param self The target taos_gauge_t* + * @param r_value The double to add to the taos_gauge_t passed as self. + * @param label_values The label values associated with the metric sample being updated. The number of labels must + * match the value passed to label_key_count in the gauge's constructor. If no label values are + * necessary, pass NULL. Otherwise, It may be convenient to pass this value as a literal. + * @return A non-zero integer value upon failure. + * + * *Example* + * + * // An example with labels + * taos_gauge_sub(foo_gauge 22, (const char**) { "bar", "bang" }); + * + * // An example without labels + * taos_gauge_sub(foo_gauge, 22, NULL); + */ +int taos_gauge_sub(taos_gauge_t *self, double r_value, const char **label_values); + +/** + * @brief Set the value for the taos_gauge_t* + * @param self The target taos_gauge_t* + * @param r_value The double to which the taos_gauge_t* passed as self will be set + * @param label_values The label values associated with the metric sample being updated. The number of labels must + * match the value passed to label_key_count in the gauge's constructor. If no label values are + * necessary, pass NULL. Otherwise, It may be convenient to pass this value as a literal. + * @return A non-zero integer value upon failure. + * + * *Example* + * + * // An example with labels + * taos_gauge_set(foo_gauge 22, (const char**) { "bar", "bang" }); + * + * // An example without labels + * taos_gauge_set(foo_gauge, 22, NULL); + */ +int taos_gauge_set(taos_gauge_t *self, double r_value, const char **label_values); + +#endif // TAOS_GAUGE_H diff --git a/include/libs/monitorfw/taos_metric_sample.h b/include/libs/monitorfw/taos_metric_sample.h index 1c37f59e32..0283d56ad9 100644 --- a/include/libs/monitorfw/taos_metric_sample.h +++ b/include/libs/monitorfw/taos_metric_sample.h @@ -56,4 +56,6 @@ int taos_metric_sample_sub(taos_metric_sample_t *self, double r_value); */ int taos_metric_sample_set(taos_metric_sample_t *self, double r_value); +int taos_metric_sample_exchange(taos_metric_sample_t *self, double r_value, double* old_value); + #endif // TAOS_METRIC_SAMPLE_H diff --git a/include/libs/monitorfw/taos_monitor.h b/include/libs/monitorfw/taos_monitor.h index ec4f007c45..c8569dd0a0 100644 --- a/include/libs/monitorfw/taos_monitor.h +++ b/include/libs/monitorfw/taos_monitor.h @@ -126,5 +126,7 @@ limitations under the License. #include "taos_map.h" #include "taos_metric.h" #include "taos_metric_sample.h" +#include "taos_monitor_util.h" +#include "taos_gauge.h" #endif // TAOS_INCLUDED \ No newline at end of file diff --git a/include/libs/monitorfw/taos_monitor_util.h b/include/libs/monitorfw/taos_monitor_util.h new file mode 100644 index 0000000000..a42d1ddcef --- /dev/null +++ b/include/libs/monitorfw/taos_monitor_util.h @@ -0,0 +1,31 @@ +/* + * 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 . + */ + +/** + * @file taos_monitor_util.h + * @brief Functions for retrieving metric samples from metrics given an ordered set of labels + */ + +#ifndef TAOS_MONITOR_UTIL_H +#define TAOS_MONITOR_UTIL_H + +#include "taos_metric.h" +#include "tjson.h" + +void taos_monitor_split_str(char** arr, char* str, const char* del); +void taos_monitor_split_str_metric(char** arr, taos_metric_t* metric, const char* del, char** buf); +char* taos_monitor_get_metric_name(taos_metric_t* metric); + +#endif // TAOS_MONITOR_UTIL_H \ No newline at end of file diff --git a/include/util/tdef.h b/include/util/tdef.h index 875a6f5738..b6f64c368d 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -196,6 +196,8 @@ typedef enum ELogicConditionType { // ACCOUNT is a 32 bit positive integer // this is the length of its string representation, including the terminator zero #define TSDB_ACCT_ID_LEN 11 +#define TSDB_NODE_ID_LEN 11 +#define TSDB_VGROUP_ID_LEN 11 #define TSDB_MAX_COLUMNS 4096 #define TSDB_MIN_COLUMNS 2 // PRIMARY COLUMN(timestamp) + other columns @@ -543,6 +545,10 @@ enum { #define VNODE_TIMEOUT_SEC 60 #define MNODE_TIMEOUT_SEC 60 +#define MONITOR_TABLENAME_LEN 200 +#define MONITOR_TAG_NAME_LEN 100 +#define MONITOR_TAG_VALUE_LEN 300 +#define MONITOR_METRIC_NAME_LEN 100 #ifdef __cplusplus } #endif diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index b3d0023f9a..bb77c530c3 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -18,6 +18,7 @@ #include "clientLog.h" #include "functionMgt.h" #include "os.h" +#include "osSleep.h" #include "query.h" #include "qworker.h" #include "scheduler.h" @@ -82,6 +83,8 @@ static void deregisterRequest(SRequestObj *pRequest) { int32_t num = atomic_sub_fetch_32(&pTscObj->numOfReqs, 1); int32_t reqType = SLOW_LOG_TYPE_OTHERS; + //taosSsleep(3); + int64_t duration = taosGetTimestampUs() - pRequest->metric.start; tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%.2f ms, " @@ -108,6 +111,8 @@ static void deregisterRequest(SRequestObj *pRequest) { } } + duration = 7000000; + if (duration >= (tsSlowLogThreshold * 1000000UL)) { atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1); if (tsSlowLogScope & reqType) { diff --git a/source/client/src/selectMonitor.c b/source/client/src/selectMonitor.c index c1cab23188..58aba8de0b 100644 --- a/source/client/src/selectMonitor.c +++ b/source/client/src/selectMonitor.c @@ -16,7 +16,7 @@ #include "clientMonitor.h" #include "clientLog.h" -const char* selectMonitorName = "slow_query"; +const char* selectMonitorName = "slow_query:slow_query_metric"; const char* selectMonitorHelp = "slow query log when cost > 3s"; const int selectMonitorLabelCount = 1; const char* selectMonitorLabels[] = {"default"}; diff --git a/source/client/src/slowQueryMonitor.c b/source/client/src/slowQueryMonitor.c index 420b66a954..4af00f5360 100644 --- a/source/client/src/slowQueryMonitor.c +++ b/source/client/src/slowQueryMonitor.c @@ -17,7 +17,7 @@ #include "clientLog.h" #include "tglobal.h" -const char* slowQueryName = "slow_query"; +const char* slowQueryName = "slow_query:slow_query_metric"; const char* slowQueryHelp = "slow query log when cost > 3s"; const int slowQueryLabelCount = 1; const char* slowQueryLabels[] = {"cost"}; diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 30596d3e72..cb853c1a75 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -93,6 +93,9 @@ char tsMonitorFqdn[TSDB_FQDN_LEN] = {0}; uint16_t tsMonitorPort = 6043; int32_t tsMonitorMaxLogs = 100; bool tsMonitorComp = false; +bool tsMonitorLogProtocol = false; +int32_t tsMonitorIntervalForBasic = 30; +bool tsMonitorForceV2 = false; // audit bool tsEnableAudit = true; @@ -690,7 +693,11 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "monitorPort", tsMonitorPort, 1, 65056, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddInt32(pCfg, "monitorMaxLogs", tsMonitorMaxLogs, 1, 1000000, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddBool(pCfg, "monitorComp", tsMonitorComp, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; - + if (cfgAddBool(pCfg, "monitorLogProtocol", tsMonitorLogProtocol, CFG_SCOPE_SERVER, CFG_DYN_SERVER) != 0) return -1; + if (cfgAddInt32(pCfg, "monitorIntervalForBasic", tsMonitorIntervalForBasic, 1, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) + return -1; + if (cfgAddBool(pCfg, "monitorForceV2", tsMonitorForceV2, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; + if (cfgAddBool(pCfg, "audit", tsEnableAudit, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1; if (cfgAddBool(pCfg, "auditCreateTable", tsEnableAuditCreateTable, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddInt32(pCfg, "auditInterval", tsAuditInterval, 500, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) @@ -1152,7 +1159,10 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsMonitorMaxLogs = cfgGetItem(pCfg, "monitorMaxLogs")->i32; tsMonitorComp = cfgGetItem(pCfg, "monitorComp")->bval; tsQueryRspPolicy = cfgGetItem(pCfg, "queryRspPolicy")->i32; - + tsMonitorLogProtocol = cfgGetItem(pCfg, "monitorLogProtocol")->bval; + tsMonitorIntervalForBasic = cfgGetItem(pCfg, "monitorIntervalForBasic")->i32; + tsMonitorForceV2 = cfgGetItem(pCfg, "monitorForceV2")->i32; + tsEnableAudit = cfgGetItem(pCfg, "audit")->bval; tsEnableAuditCreateTable = cfgGetItem(pCfg, "auditCreateTable")->bval; tsAuditInterval = cfgGetItem(pCfg, "auditInterval")->i32; diff --git a/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h b/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h index 80502e2662..975246a10f 100644 --- a/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h +++ b/source/dnode/mgmt/mgmt_dnode/inc/dmInt.h @@ -43,6 +43,7 @@ typedef struct SDnodeMgmt { GetMnodeLoadsFp getMnodeLoadsFp; GetQnodeLoadsFp getQnodeLoadsFp; int32_t statusSeq; + SendMonitorReportFp sendMonitorReportFpBasic; } SDnodeMgmt; // dmHandle.c diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmInt.c b/source/dnode/mgmt/mgmt_dnode/src/dmInt.c index b9dd45f1c0..a651fbf060 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmInt.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmInt.c @@ -65,6 +65,7 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { pMgmt->processDropNodeFp = pInput->processDropNodeFp; pMgmt->sendMonitorReportFp = pInput->sendMonitorReportFp; pMgmt->sendAuditRecordsFp = pInput->sendAuditRecordFp; + pMgmt->sendMonitorReportFpBasic = pInput->sendMonitorReportFpBasic; pMgmt->getVnodeLoadsFp = pInput->getVnodeLoadsFp; pMgmt->getVnodeLoadsLiteFp = pInput->getVnodeLoadsLiteFp; pMgmt->getMnodeLoadsFp = pInput->getMnodeLoadsFp; diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index af43804db4..1f2dc575d0 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -81,6 +81,7 @@ static void *dmNotifyThreadFp(void *param) { static void *dmMonitorThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); + int64_t lastTimeForBasic = taosGetTimestampMs(); setThreadName("dnode-monitor"); while (1) { @@ -88,12 +89,22 @@ static void *dmMonitorThreadFp(void *param) { if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; int64_t curTime = taosGetTimestampMs(); + if (curTime < lastTime) lastTime = curTime; float interval = (curTime - lastTime) / 1000.0f; if (interval >= tsMonitorInterval) { (*pMgmt->sendMonitorReportFp)(); lastTime = curTime; } + + if(tsMonitorForceV2){ + if (curTime < lastTimeForBasic) lastTimeForBasic = curTime; + float intervalForBasic = (curTime - lastTimeForBasic) / 1000.0f; + if (intervalForBasic >= tsMonitorIntervalForBasic) { + (*pMgmt->sendMonitorReportFpBasic)(); + lastTimeForBasic = curTime; + } + } } return NULL; diff --git a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h index c646bb4bdd..25e388672e 100644 --- a/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h +++ b/source/dnode/mgmt/node_mgmt/inc/dmMgmt.h @@ -149,6 +149,7 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg); // dmMonitor.c void dmSendMonitorReport(); void dmSendAuditRecords(); +void dmSendMonitorReportBasic(); void dmGetVnodeLoads(SMonVloadInfo *pInfo); void dmGetVnodeLoadsLite(SMonVloadInfo *pInfo); void dmGetMnodeLoads(SMonMloadInfo *pInfo); diff --git a/source/dnode/mgmt/node_mgmt/src/dmEnv.c b/source/dnode/mgmt/node_mgmt/src/dmEnv.c index f9bba19fbb..b343be9426 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmEnv.c +++ b/source/dnode/mgmt/node_mgmt/src/dmEnv.c @@ -398,6 +398,7 @@ SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) { .processDropNodeFp = dmProcessDropNodeReq, .sendMonitorReportFp = dmSendMonitorReport, .sendAuditRecordFp = auditSendRecordsInBatch, + .sendMonitorReportFpBasic = dmSendMonitorReportBasic, .getVnodeLoadsFp = dmGetVnodeLoads, .getVnodeLoadsLiteFp = dmGetVnodeLoadsLite, .getMnodeLoadsFp = dmGetMnodeLoads, diff --git a/source/dnode/mgmt/node_mgmt/src/dmMonitor.c b/source/dnode/mgmt/node_mgmt/src/dmMonitor.c index 69a7956c9c..590c25b936 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmMonitor.c +++ b/source/dnode/mgmt/node_mgmt/src/dmMonitor.c @@ -19,7 +19,8 @@ #include "audit.h" static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { - pInfo->protocol = 1; + //pInfo->protocol = 1; + pInfo->protocol = 2; pInfo->dnode_id = pDnode->data.dnodeId; pInfo->cluster_id = pDnode->data.clusterId; tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN); @@ -106,9 +107,20 @@ void dmSendMonitorReport() { dmGetVmMonitorInfo(pDnode); dmGetQmMonitorInfo(pDnode); dmGetSmMonitorInfo(pDnode); - monSendReport(); + monGenAndSendReport(); +} - monSendPromReport(); +void dmSendMonitorReportBasic() { + if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return; + dTrace("send monitor report to %s:%u", tsMonitorFqdn, tsMonitorPort); + + SDnode *pDnode = dmInstance(); + dmGetDmMonitorInfo(pDnode); + dmGetMmMonitorInfo(pDnode); + //dmGetVmMonitorInfo(pDnode); + //dmGetQmMonitorInfo(pDnode); + //dmGetSmMonitorInfo(pDnode); + monGenAndSendReportBasic(); } //Todo: put this in seperate file in the future diff --git a/source/dnode/mgmt/node_util/inc/dmUtil.h b/source/dnode/mgmt/node_util/inc/dmUtil.h index 4769ef8538..16c464d17c 100644 --- a/source/dnode/mgmt/node_util/inc/dmUtil.h +++ b/source/dnode/mgmt/node_util/inc/dmUtil.h @@ -122,6 +122,7 @@ typedef struct { ProcessDropNodeFp processDropNodeFp; SendMonitorReportFp sendMonitorReportFp; SendAuditRecordsFp sendAuditRecordFp; + SendMonitorReportFp sendMonitorReportFpBasic; GetVnodeLoadsFp getVnodeLoadsFp; GetVnodeLoadsFp getVnodeLoadsLiteFp; GetMnodeLoadsFp getMnodeLoadsFp; diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c index b75753a87f..bd0c29f4c8 100644 --- a/source/dnode/mnode/impl/src/mndDnode.c +++ b/source/dnode/mnode/impl/src/mndDnode.c @@ -14,6 +14,7 @@ */ #define _DEFAULT_SOURCE +#include "tjson.h" #include "mndDnode.h" #include "audit.h" #include "mndCluster.h" @@ -28,6 +29,7 @@ #include "mndVgroup.h" #include "tmisce.h" #include "tunit.h" +#include "taos_monitor.h" #define TSDB_DNODE_VER_NUMBER 2 #define TSDB_DNODE_RESERVE_SIZE 64 @@ -503,9 +505,129 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { goto _OVER; } - monSendContent(pReq->pCont); + if(tsMonitorLogProtocol){ + mInfo("process statis req,\n %s", statisReq.pCont); + } + + SJson* pJson = tjsonParse(statisReq.pCont); + + int32_t ts_size = tjsonGetArraySize(pJson); + + for(int32_t i = 0; i < ts_size; i++){ + SJson* item = tjsonGetArrayItem(pJson, i); + + SJson* tables = tjsonGetObjectItem(item, "tables"); + + int32_t tableSize = tjsonGetArraySize(tables); + for(int32_t i = 0; i < tableSize; i++){ + SJson* table = tjsonGetArrayItem(tables, i); + + char tableName[MONITOR_TABLENAME_LEN] = {0}; + tjsonGetStringValue(table, "name", tableName); + + SJson* metricGroups = tjsonGetObjectItem(table, "metric_groups"); + + int32_t size = tjsonGetArraySize(metricGroups); + for(int32_t i = 0; i < size; i++){ + SJson* item = tjsonGetArrayItem(metricGroups, i); + + SJson* arrayTag = tjsonGetObjectItem(item, "tags"); + + int32_t tagSize = tjsonGetArraySize(arrayTag); + + char** labels = taosMemoryMalloc(sizeof(char*) * tagSize); + char** sample_labels = taosMemoryMalloc(sizeof(char*) * tagSize); + + for(int32_t j = 0; j < tagSize; j++){ + SJson* item = tjsonGetArrayItem(arrayTag, j); + + *(labels + j) = taosMemoryMalloc(MONITOR_TAG_NAME_LEN); + tjsonGetStringValue(item, "name", *(labels + j)); + + *(sample_labels + j) = taosMemoryMalloc(MONITOR_TAG_VALUE_LEN); + tjsonGetStringValue(item, "value", *(sample_labels + j)); + } + + SJson* metrics = tjsonGetObjectItem(item, "metrics"); + + int32_t metricLen = tjsonGetArraySize(metrics); + for(int32_t j = 0; j < metricLen; j++){ + SJson *item = tjsonGetArrayItem(metrics, j); + + char name[MONITOR_METRIC_NAME_LEN] = {0}; + tjsonGetStringValue(item, "name", name); + + double value = 0; + tjsonGetDoubleValue(item, "value", &value); + + double type = 0; + tjsonGetDoubleValue(item, "type", &type); + + int32_t metricNameLen = strlen(name) + strlen(tableName) + 2; + char* metricName = taosMemoryMalloc(metricNameLen); + memset(metricName, 0, metricNameLen); + sprintf(metricName, "%s:%s", tableName, name); + + taos_metric_t* metric = taos_collector_registry_get_metric(metricName); + if(metric == NULL){ + if(type == 0){ + metric = taos_counter_new(metricName, "", tagSize, (const char**)labels); + } + if(type == 1){ + metric = taos_gauge_new(metricName, "", tagSize, (const char**)labels); + } + mTrace("fail to get metric from registry, new one metric:%p", metric); + + if(taos_collector_registry_register_metric(metric) == 1){ + if(type == 0){ + taos_counter_destroy(metric); + } + if(type == 1){ + taos_gauge_destroy(metric); + } + + metric = taos_collector_registry_get_metric(metricName); + + mTrace("fail to register metric, get metric from registry:%p", metric); + } + else{ + mTrace("succeed to register metric:%p", metric); + } + } + else{ + mTrace("get metric from registry:%p", metric); + } + + if(type == 0){ + taos_counter_add(metric, value, (const char**)sample_labels); + } + if(type == 1){ + taos_gauge_set(metric, value, (const char**)sample_labels); + } + + taosMemoryFreeClear(metricName); + } + + for(int32_t j = 0; j < tagSize; j++){ + taosMemoryFreeClear(*(labels + j)); + taosMemoryFreeClear(*(sample_labels + j)); + } + + taosMemoryFreeClear(sample_labels); + taosMemoryFreeClear(labels); + } + } + + } + + code = 0; _OVER: + if(pJson != NULL){ + tjsonDelete(pJson); + pJson = NULL; + } + tFreeSStatisReq(&statisReq); return code; } diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index 38a92b43e7..51abcc4eae 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -859,8 +859,10 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr pClusterInfo->master_uptime = (float)mndGetClusterUpTime(pMnode) / 86400.0f; // pClusterInfo->master_uptime = (ms - pObj->stateStartTime) / (86400000.0f); tstrncpy(desc.role, syncStr(TAOS_SYNC_STATE_LEADER), sizeof(desc.role)); + desc.syncState = TAOS_SYNC_STATE_LEADER; } else { tstrncpy(desc.role, syncStr(pObj->syncState), sizeof(desc.role)); + desc.syncState = pObj->syncState; } taosArrayPush(pClusterInfo->mnodes, &desc); sdbRelease(pSdb, pObj); @@ -891,6 +893,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr SMonVnodeDesc *pVnDesc = &desc.vnodes[i]; pVnDesc->dnode_id = pVgid->dnodeId; tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->syncState), sizeof(pVnDesc->vnode_role)); + pVnDesc->syncState = pVgid->syncState; if (pVgid->syncState == TAOS_SYNC_STATE_LEADER) { tstrncpy(desc.status, "ready", sizeof(desc.status)); pClusterInfo->vgroups_alive++; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index c1a4754b62..eb9c3483d1 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -47,6 +47,8 @@ #include "vnode.h" +#include "taos_monitor.h" + #ifdef __cplusplus extern "C" { #endif @@ -104,6 +106,20 @@ typedef struct SQueryNode SQueryNode; #define VND_INFO_FNAME "vnode.json" #define VND_INFO_FNAME_TMP "vnode_tmp.json" +#define VNODE_METRIC_SQL_COUNT "taoscd_sql_req:count" + +#define VNODE_METRIC_TAG_NAME_SQL_TYPE "sql_type" +#define VNODE_METRIC_TAG_NAME_CLUSTER_ID "cluster_id" +#define VNODE_METRIC_TAG_NAME_DNODE_ID "dnode_id" +#define VNODE_METRIC_TAG_NAME_DNODE_EP "dnode_ep" +#define VNODE_METRIC_TAG_NAME_VGROUP_ID "vgroup_id" +#define VNODE_METRIC_TAG_NAME_USERNAME "username" +#define VNODE_METRIC_TAG_NAME_RESULT "result" + +#define VNODE_METRIC_TAG_VALUE_INSERT_AFFECTED_ROWS "insert_affected_rows" +//#define VNODE_METRIC_TAG_VALUE_INSERT "insert" +#define VNODE_METRIC_TAG_VALUE_DELETE "delete" + // vnd.h typedef int32_t (*_query_reseek_func_t)(void* pQHandle); struct SQueryNode { @@ -438,6 +454,13 @@ typedef struct SVCommitSched { int64_t maxWaitMs; } SVCommitSched; +typedef struct SVMonitorObj{ + char strClusterId[TSDB_CLUSTER_ID_LEN]; + char strDnodeId[TSDB_NODE_ID_LEN]; + char strVgId[TSDB_VGROUP_ID_LEN]; + taos_counter_t *insertCounter; +}SVMonitorObj; + struct SVnode { char* path; SVnodeCfg config; @@ -476,6 +499,7 @@ struct SVnode { int32_t blockSec; int64_t blockSeq; SQHandle* pQuery; + SVMonitorObj monitor; }; #define TD_VID(PVNODE) ((PVNODE)->config.vgId) diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c index 946ce9d278..e1f76b3a25 100644 --- a/source/dnode/vnode/src/vnd/vnodeOpen.c +++ b/source/dnode/vnode/src/vnd/vnodeOpen.c @@ -480,6 +480,28 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC vnodeRollback(pVnode); } + snprintf(pVnode->monitor.strClusterId, TSDB_CLUSTER_ID_LEN, "%"PRId64, pVnode->config.syncCfg.nodeInfo[0].clusterId); + snprintf(pVnode->monitor.strDnodeId, TSDB_NODE_ID_LEN, "%"PRId32, pVnode->config.syncCfg.nodeInfo[0].nodeId); + snprintf(pVnode->monitor.strVgId, TSDB_VGROUP_ID_LEN, "%"PRId32, pVnode->config.vgId); + + if(pVnode->monitor.insertCounter == NULL){ + int32_t label_count = 7; + const char *sample_labels[] = {VNODE_METRIC_TAG_NAME_SQL_TYPE, VNODE_METRIC_TAG_NAME_CLUSTER_ID, + VNODE_METRIC_TAG_NAME_DNODE_ID, VNODE_METRIC_TAG_NAME_DNODE_EP, + VNODE_METRIC_TAG_NAME_VGROUP_ID, VNODE_METRIC_TAG_NAME_USERNAME, + VNODE_METRIC_TAG_NAME_RESULT}; + taos_counter_t *counter = taos_counter_new(VNODE_METRIC_SQL_COUNT, "counter for insert sql", + label_count, sample_labels); + vInfo("vgId:%d, new metric:%p",TD_VID(pVnode), counter); + if(taos_collector_registry_register_metric(counter) == 1){ + taos_counter_destroy(counter); + counter = taos_collector_registry_get_metric(VNODE_METRIC_SQL_COUNT); + vInfo("vgId:%d, get metric from registry:%p",TD_VID(pVnode), counter); + } + pVnode->monitor.insertCounter = counter; + vInfo("vgId:%d, succeed to set metric:%p",TD_VID(pVnode), counter); + } + return pVnode; _err: diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 1c45c93346..d79d239a50 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -21,8 +21,6 @@ #include "vnd.h" #include "vnode.h" #include "vnodeInt.h" -#include "taos_monitor.h" - static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); @@ -32,20 +30,21 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp, SRpcMsg *pOriginRpc); -static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); +static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp, + SRpcMsg *pOriginalMsg); static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); -static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); +static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp, + SRpcMsg *pOriginalMsg); static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessCreateIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessDropIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessCompactVnodeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessConfigChangeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); -taos_counter_t *insert_counter = NULL; extern int32_t vnodeProcessKillCompactReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); extern int32_t vnodeQueryCompactProgress(SVnode *pVnode, SRpcMsg *pMsg); @@ -535,10 +534,10 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg break; /* TSDB */ case TDMT_VND_SUBMIT: - if (vnodeProcessSubmitReq(pVnode, ver, pMsg->pCont, pMsg->contLen, pRsp) < 0) goto _err; + if (vnodeProcessSubmitReq(pVnode, ver, pMsg->pCont, pMsg->contLen, pRsp, pMsg) < 0) goto _err; break; case TDMT_VND_DELETE: - if (vnodeProcessDeleteReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err; + if (vnodeProcessDeleteReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err; break; case TDMT_VND_BATCH_DEL: if (vnodeProcessBatchDeleteReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err; @@ -1482,7 +1481,8 @@ static int32_t vnodeRebuildSubmitReqMsg(SSubmitReq2 *pSubmitReq, void **ppMsg) { return code; } -static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) { +static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp, + SRpcMsg *pOriginalMsg) { int32_t code = 0; terrno = 0; @@ -1700,28 +1700,33 @@ _exit: atomic_add_fetch_64(&pVnode->statis.nInsert, pSubmitRsp->affectedRows); atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, pSubmitRsp->affectedRows); atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1); + + const char *sample_labels[] = {VNODE_METRIC_TAG_VALUE_INSERT_AFFECTED_ROWS, pVnode->monitor.strClusterId, + pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId, + pOriginalMsg->info.conn.user, "Success"}; + taos_counter_add(pVnode->monitor.insertCounter, pSubmitRsp->affectedRows, sample_labels); + if (code == 0) { atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1); code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len); } + /* + if (code == 0) { + atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1); + code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len); - if(insert_counter == NULL){ - int32_t label_count =2; - const char *sample_labels[] = {"vgid", "endpoint"}; - taos_counter_t *counter = taos_counter_new("insert_counter", "counter for insert sql", label_count, sample_labels); - if(taos_collector_registry_register_metric(counter) == 1){ - taos_counter_destroy(counter); - } - else{ - atomic_store_ptr(&insert_counter, counter); - } + const char *batch_sample_labels[] = {VNODE_METRIC_TAG_VALUE_INSERT, pVnode->monitor.strClusterId, + pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId, + pOriginalMsg->info.conn.user, "Success"}; + taos_counter_inc(pVnode->monitor.insertCounter, batch_sample_labels); } - - char vgId[50]; - sprintf(vgId, "%"PRId32, TD_VID(pVnode)); - const char *sample_labels[] = {vgId, tsLocalEp}; - - taos_counter_inc(insert_counter, sample_labels); + else{ + const char *batch_sample_labels[] = {VNODE_METRIC_TAG_VALUE_INSERT, pVnode->monitor.strClusterId, + pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId, + pOriginalMsg->info.conn.user, "Failed"}; + taos_counter_inc(pVnode->monitor.insertCounter, batch_sample_labels); + } + */ // clear taosArrayDestroy(newTbUids); @@ -1977,7 +1982,8 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pRe return 0; } -static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) { +static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp, + SRpcMsg *pOriginalMsg) { int32_t code = 0; SDecoder *pCoder = &(SDecoder){0}; SDeleteRes *pRes = &(SDeleteRes){0}; @@ -2020,6 +2026,19 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, in return code; _err: + if(code == TSDB_CODE_SUCCESS){ + const char *batch_sample_labels[] = {VNODE_METRIC_TAG_VALUE_DELETE, pVnode->monitor.strClusterId, + pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId, + pOriginalMsg->info.conn.user, "Success"}; + taos_counter_inc(pVnode->monitor.insertCounter, batch_sample_labels); + } + else{ + const char *batch_sample_labels[] = {VNODE_METRIC_TAG_VALUE_DELETE, pVnode->monitor.strClusterId, + pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId, + pOriginalMsg->info.conn.user, "Failed"}; + taos_counter_inc(pVnode->monitor.insertCounter, batch_sample_labels); + } + return code; } static int32_t vnodeProcessCreateIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) { diff --git a/source/libs/monitor/inc/monInt.h b/source/libs/monitor/inc/monInt.h index c5219c60b2..7fc718393b 100644 --- a/source/libs/monitor/inc/monInt.h +++ b/source/libs/monitor/inc/monInt.h @@ -20,6 +20,7 @@ #include "query.h" #include "tjson.h" +#include "thash.h" typedef struct { int64_t curTime; @@ -45,8 +46,22 @@ typedef struct { SMonSmInfo smInfo; SMonQmInfo qmInfo; SMonBmInfo bmInfo; + SHashObj *metrics; } SMonitor; +void monGenClusterInfoTable(SMonInfo *pMonitor); +void monGenVgroupInfoTable(SMonInfo *pMonitor); +void monGenDnodeInfoTable(SMonInfo *pMonitor); +void monGenDnodeStatusInfoTable(SMonInfo *pMonitor); +void monGenDataDiskTable(SMonInfo *pMonitor); +void monGenLogDiskTable(SMonInfo *pMonitor); +void monGenMnodeRoleTable(SMonInfo *pMonitor); +void monGenVnodeRoleTable(SMonInfo *pMonitor); + +void monSendPromReport(); +void monInitMonitorFW(); +void monCleanupMonitorFW(); + #ifdef __cplusplus } #endif diff --git a/source/libs/monitor/src/clientMonitor.c b/source/libs/monitor/src/clientMonitor.c index 15d047f3c0..64df693e1e 100644 --- a/source/libs/monitor/src/clientMonitor.c +++ b/source/libs/monitor/src/clientMonitor.c @@ -10,16 +10,17 @@ tmr_h tmrStartHandle; SHashObj* clusterMonitorInfoTable; static const int interval = 1000; // ms -static const int sendBathchSize = 10; +static const int sendBathchSize = 1; int32_t sendReport(ClientMonitor* pMonitor, char* pCont); void generateClusterReport(ClientMonitor* pMonitor, bool send) { char ts[50]; sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); - char* pCont = (char*)taos_collector_registry_bridge(pMonitor->registry, ts, "%" PRId64); + char* pCont = (char*)taos_collector_registry_bridge_new(pMonitor->registry, ts, "%" PRId64, NULL); + uInfo("report cont:\n%s", pCont); if (send && strlen(pCont) != TSDB_CODE_SUCCESS) { if (sendReport(pMonitor, pCont) == 0) { - taos_collector_registry_clear_out(pMonitor->registry); + taos_collector_registry_clear_batch(pMonitor->registry); } } } diff --git a/source/libs/monitor/src/monFramework.c b/source/libs/monitor/src/monFramework.c new file mode 100644 index 0000000000..3cc2ab8c90 --- /dev/null +++ b/source/libs/monitor/src/monFramework.c @@ -0,0 +1,656 @@ +/* + * 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 "monitor.h" +#include "monInt.h" + +#include "thash.h" +#include "taos_monitor.h" +#include "thttp.h" +#include "ttime.h" +#include "tglobal.h" + +extern SMonitor tsMonitor; +extern char* tsMonUri; +extern char* tsMonFwUri; + +#define LEVEL_LEN 11 + +#define CLUSTER_TABLE "taosd_cluster_info" + +#define MASTER_UPTIME CLUSTER_TABLE":master_uptime" +#define DBS_TOTAL CLUSTER_TABLE":dbs_total" +#define TBS_TOTAL CLUSTER_TABLE":tbs_total" +#define STBS_TOTAL CLUSTER_TABLE":stbs_total" +#define VGROUPS_TOTAL CLUSTER_TABLE":vgroups_total" +#define VGROUPS_ALIVE CLUSTER_TABLE":vgroups_alive" +#define VNODES_TOTAL CLUSTER_TABLE":vnodes_total" +#define VNODES_ALIVE CLUSTER_TABLE":vnodes_alive" +#define DNODES_TOTAL CLUSTER_TABLE":dnodes_total" +#define DNODES_ALIVE CLUSTER_TABLE":dnodes_alive" +#define CONNECTIONS_TOTAL CLUSTER_TABLE":connections_total" +#define TOPICS_TOTAL CLUSTER_TABLE":topics_total" +#define STREAMS_TOTAL CLUSTER_TABLE":streams_total" +#define EXPIRE_TIME CLUSTER_TABLE":expire_time" +#define TIMESERIES_USED CLUSTER_TABLE":timeseries_used" +#define TIMESERIES_TOTAL CLUSTER_TABLE":timeseries_total" + +#define VGROUP_TABLE "taosd_cluster_vgroups_info" + +#define TABLES_NUM VGROUP_TABLE":tables_num" +#define STATUS VGROUP_TABLE":status" + +#define DNODE_TABLE "taosd_dnodes_info" + +#define UPTIME DNODE_TABLE":uptime" +#define CPU_ENGINE DNODE_TABLE":cpu_engine" +#define CPU_SYSTEM DNODE_TABLE":cpu_system" +#define CPU_CORE DNODE_TABLE":cpu_cores" +#define MEM_ENGINE DNODE_TABLE":mem_engine" +#define MEM_SYSTEM DNODE_TABLE":mem_system" +#define MEM_TOTAL DNODE_TABLE":mem_total" +#define DISK_ENGINE DNODE_TABLE":disk_engine" +#define DISK_USED DNODE_TABLE":disk_used" +#define DISK_TOTAL DNODE_TABLE":disk_total" +#define NET_IN DNODE_TABLE":net_in" +#define NET_OUT DNODE_TABLE":net_out" +#define IO_READ DNODE_TABLE":io_read" +#define IO_WRITE DNODE_TABLE":io_write" +#define IO_READ_DISK DNODE_TABLE":io_read_disk" +#define IO_WRITE_DISK DNODE_TABLE":io_write_disk" +#define ERRORS DNODE_TABLE":errors" +#define VNODES_NUM DNODE_TABLE":vnodes_num" +#define MASTERS DNODE_TABLE":masters" +#define HAS_MNODE DNODE_TABLE":has_mnode" +#define HAS_QNODE DNODE_TABLE":has_qnode" +#define HAS_SNODE DNODE_TABLE":has_snode" +#define DNODE_LOG_ERROR DNODE_TABLE":ERROR" +#define DNODE_LOG_INFO DNODE_TABLE":INFO" +#define DNODE_LOG_DEBUG DNODE_TABLE":DEBUG" +#define DNODE_LOG_TRACE DNODE_TABLE":TRACE" + +#define DNODE_STATUS "taosd_dnodes_status:status" + +#define DATADIR_TABLE "taosd_dnodes_data_dirs" + +#define DNODE_DATA_AVAIL DATADIR_TABLE":avail" +#define DNODE_DATA_USED DATADIR_TABLE":used" +#define DNODE_DATA_TOTAL DATADIR_TABLE":total" + +#define LOGDIR_TABLE "taosd_dnodes_log_dirs" + +#define DNODE_LOG_AVAIL LOGDIR_TABLE":avail" +#define DNODE_LOG_USED LOGDIR_TABLE":used" +#define DNODE_LOG_TOTAL LOGDIR_TABLE":total" + +#define MNODE_ROLE "taosd_mnodes_info:role" +#define VNODE_ROLE "taosd_vnodes_info:role" + +void monInitMonitorFW(){ + taos_collector_registry_default_init(); + + tsMonitor.metrics = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); + taos_gauge_t *gauge = NULL; + + int32_t label_count =1; + const char *sample_labels[] = {"cluster_id"}; + char *metric[] = {MASTER_UPTIME, DBS_TOTAL, TBS_TOTAL, STBS_TOTAL, VGROUPS_TOTAL, + VGROUPS_ALIVE, VNODES_TOTAL, VNODES_ALIVE, CONNECTIONS_TOTAL, TOPICS_TOTAL, STREAMS_TOTAL, + DNODES_TOTAL, DNODES_ALIVE, EXPIRE_TIME, TIMESERIES_USED, + TIMESERIES_TOTAL}; + for(int32_t i = 0; i < 16; i++){ + gauge= taos_gauge_new(metric[i], "", label_count, sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, metric[i], strlen(metric[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t vgroup_label_count = 3; + const char *vgroup_sample_labels[] = {"cluster_id", "vgroup_id", "database_name"}; + char *vgroup_metrics[] = {TABLES_NUM, STATUS}; + for(int32_t i = 0; i < 2; i++){ + gauge= taos_gauge_new(vgroup_metrics[i], "", vgroup_label_count, vgroup_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, vgroup_metrics[i], strlen(vgroup_metrics[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t dnodes_label_count = 3; + const char *dnodes_sample_labels[] = {"cluster_id", "dnode_id", "dnode_ep"}; + char *dnodes_gauges[] = {UPTIME, CPU_ENGINE, CPU_SYSTEM, CPU_CORE, MEM_ENGINE, MEM_SYSTEM, + MEM_TOTAL, DISK_ENGINE, DISK_USED, DISK_TOTAL, NET_IN, + NET_OUT, IO_READ, IO_WRITE, IO_READ_DISK, IO_WRITE_DISK, ERRORS, + VNODES_NUM, MASTERS, HAS_MNODE, HAS_QNODE, HAS_SNODE, DNODE_STATUS, + DNODE_LOG_ERROR, DNODE_LOG_INFO, DNODE_LOG_DEBUG, DNODE_LOG_TRACE}; + for(int32_t i = 0; i < 27; i++){ + gauge= taos_gauge_new(dnodes_gauges[i], "", dnodes_label_count, dnodes_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, dnodes_gauges[i], strlen(dnodes_gauges[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t dnodes_data_label_count = 5; + const char *dnodes_data_sample_labels[] = {"cluster_id", "dnode_id", "dnode_ep", "data_dir_name", "data_dir_level"}; + char *dnodes_data_gauges[] = {DNODE_DATA_AVAIL, DNODE_DATA_USED, DNODE_DATA_TOTAL}; + for(int32_t i = 0; i < 3; i++){ + gauge= taos_gauge_new(dnodes_data_gauges[i], "", dnodes_data_label_count, dnodes_data_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, dnodes_data_gauges[i], strlen(dnodes_data_gauges[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t dnodes_log_label_count = 4; + const char *dnodes_log_sample_labels[] = {"cluster_id", "dnode_id", "dnode_ep", "data_dir_name"}; + char *dnodes_log_gauges[] = {DNODE_LOG_AVAIL, DNODE_LOG_USED, DNODE_LOG_TOTAL}; + for(int32_t i = 0; i < 3; i++){ + gauge= taos_gauge_new(dnodes_log_gauges[i], "", dnodes_log_label_count, dnodes_log_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, dnodes_log_gauges[i], strlen(dnodes_log_gauges[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t mnodes_role_label_count = 3; + const char *mnodes_role_sample_labels[] = {"cluster_id", "mnode_id", "mnode_ep"}; + char *mnodes_role_gauges[] = {MNODE_ROLE}; + for(int32_t i = 0; i < 1; i++){ + gauge= taos_gauge_new(mnodes_role_gauges[i], "", mnodes_role_label_count, mnodes_role_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, mnodes_role_gauges[i], strlen(mnodes_role_gauges[i]), &gauge, sizeof(taos_gauge_t *)); + } + + int32_t vnodes_role_label_count = 4; + const char *vnodes_role_sample_labels[] = {"cluster_id", "vgroup_id", "database_name", "dnode_id"}; + char *vnodes_role_gauges[] = {VNODE_ROLE}; + for(int32_t i = 0; i < 1; i++){ + gauge= taos_gauge_new(vnodes_role_gauges[i], "", vnodes_role_label_count, vnodes_role_sample_labels); + if(taos_collector_registry_register_metric(gauge) == 1){ + taos_counter_destroy(gauge); + } + taosHashPut(tsMonitor.metrics, vnodes_role_gauges[i], strlen(vnodes_role_gauges[i]), &gauge, sizeof(taos_gauge_t *)); + } +} + +void monCleanupMonitorFW(){ + taosHashCleanup(tsMonitor.metrics); + taos_collector_registry_destroy(TAOS_COLLECTOR_REGISTRY_DEFAULT); + TAOS_COLLECTOR_REGISTRY_DEFAULT = NULL; +} + +void monGenClusterInfoTable(SMonInfo *pMonitor){ + SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster; + SMonBasicInfo *pBasicInfo = &pMonitor->dmInfo.basic; + SMonGrantInfo *pGrantInfo = &pMonitor->mmInfo.grant; + + if(pBasicInfo->cluster_id == 0) { + uError("failed to generate dnode info table since cluster_id is 0"); + return; + } + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + //cluster info + char buf[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(buf, TSDB_CLUSTER_ID_LEN, "%"PRId64, pBasicInfo->cluster_id); + const char *sample_labels[] = {buf}; + + taos_gauge_t **metric = NULL; + + metric = taosHashGet(tsMonitor.metrics, MASTER_UPTIME, strlen(MASTER_UPTIME)); + taos_gauge_set(*metric, pInfo->master_uptime, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DBS_TOTAL, strlen(DBS_TOTAL)); + taos_gauge_set(*metric, pInfo->dbs_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, TBS_TOTAL, strlen(TBS_TOTAL)); + taos_gauge_set(*metric, pInfo->tbs_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, STBS_TOTAL, strlen(STBS_TOTAL)); + taos_gauge_set(*metric, pInfo->stbs_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, VGROUPS_TOTAL, strlen(VGROUPS_TOTAL)); + taos_gauge_set(*metric, pInfo->vgroups_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, VGROUPS_ALIVE, strlen(VGROUPS_ALIVE)); + taos_gauge_set(*metric, pInfo->vgroups_alive, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, VNODES_TOTAL, strlen(VNODES_TOTAL)); + taos_gauge_set(*metric, pInfo->vnodes_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, VNODES_ALIVE, strlen(VNODES_ALIVE)); + taos_gauge_set(*metric, pInfo->vnodes_alive, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, CONNECTIONS_TOTAL, strlen(CONNECTIONS_TOTAL)); + taos_gauge_set(*metric, pInfo->connections_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, TOPICS_TOTAL, strlen(TOPICS_TOTAL)); + taos_gauge_set(*metric, pInfo->topics_toal, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, STREAMS_TOTAL, strlen(STREAMS_TOTAL)); + taos_gauge_set(*metric, pInfo->streams_total, sample_labels); + + //dnodes number + int32_t dnode_total = taosArrayGetSize(pInfo->dnodes); + int32_t dnode_alive = 0; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->dnodes); ++i) { + SMonDnodeDesc *pDnodeDesc = taosArrayGet(pInfo->dnodes, i); + + if(strcmp(pDnodeDesc->status, "ready") == 0){ + dnode_alive++; + } + } + + metric = taosHashGet(tsMonitor.metrics, DNODES_TOTAL, strlen(DNODES_TOTAL)); + taos_gauge_set(*metric, dnode_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODES_ALIVE, strlen(DNODES_ALIVE)); + taos_gauge_set(*metric, dnode_alive, sample_labels); + + //grant info + metric = taosHashGet(tsMonitor.metrics, EXPIRE_TIME, strlen(EXPIRE_TIME)); + taos_gauge_set(*metric, pGrantInfo->expire_time, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, TIMESERIES_USED, strlen(TIMESERIES_USED)); + taos_gauge_set(*metric, pGrantInfo->timeseries_used, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, TIMESERIES_TOTAL, strlen(TIMESERIES_TOTAL)); + taos_gauge_set(*metric, pGrantInfo->timeseries_total, sample_labels); +} + +void monGenVgroupInfoTable(SMonInfo *pMonitor){ + if(pMonitor->dmInfo.basic.cluster_id == 0) return; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + SMonVgroupInfo *pInfo = &pMonitor->mmInfo.vgroup; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + char cluster_id[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(cluster_id, TSDB_CLUSTER_ID_LEN, "%"PRId64, pMonitor->dmInfo.basic.cluster_id); + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->vgroups); ++i) { + SMonVgroupDesc *pVgroupDesc = taosArrayGet(pInfo->vgroups, i); + + char vgroup_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(vgroup_id, TSDB_NODE_ID_LEN, "%"PRId32, pVgroupDesc->vgroup_id); + + const char *sample_labels[] = {cluster_id, vgroup_id, pVgroupDesc->database_name}; + + taos_gauge_t **metric = NULL; + + metric = taosHashGet(tsMonitor.metrics, TABLES_NUM, strlen(TABLES_NUM)); + taos_gauge_set(*metric, pVgroupDesc->tables_num, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, STATUS, strlen(STATUS)); + int32_t status = 0; + if(strcmp(pVgroupDesc->status, "ready") == 0){ + status = 1; + } + taos_gauge_set(*metric, status, sample_labels); + } +} + +void monGenDnodeInfoTable(SMonInfo *pMonitor) { + if(pMonitor->dmInfo.basic.cluster_id == 0) { + uError("failed to generate dnode info table since cluster_id is 0"); + return; + } + + char cluster_id[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(cluster_id, TSDB_CLUSTER_ID_LEN, "%"PRId64, pMonitor->dmInfo.basic.cluster_id); + + char dnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(dnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pMonitor->dmInfo.basic.dnode_id); + + const char *sample_labels[] = {cluster_id, dnode_id, pMonitor->dmInfo.basic.dnode_ep}; + + taos_gauge_t **metric = NULL; + + //dnode info + SMonDnodeInfo *pInfo = &pMonitor->dmInfo.dnode; + SMonSysInfo *pSys = &pMonitor->dmInfo.sys; + SVnodesStat *pStat = &pMonitor->vmInfo.vstat; + SMonClusterInfo *pClusterInfo = &pMonitor->mmInfo.cluster; + + double interval = (pMonitor->curTime - pMonitor->lastTime) / 1000.0; + if (pMonitor->curTime - pMonitor->lastTime == 0) { + interval = 1; + } + + double cpu_engine = 0; + double mem_engine = 0; + double net_in = 0; + double net_out = 0; + double io_read = 0; + double io_write = 0; + double io_read_disk = 0; + double io_write_disk = 0; + + SMonSysInfo *sysArrays[6]; + sysArrays[0] = &pMonitor->dmInfo.sys; + sysArrays[1] = &pMonitor->mmInfo.sys; + sysArrays[2] = &pMonitor->vmInfo.sys; + sysArrays[3] = &pMonitor->qmInfo.sys; + sysArrays[4] = &pMonitor->smInfo.sys; + sysArrays[5] = &pMonitor->bmInfo.sys; + for (int32_t i = 0; i < 6; ++i) { + cpu_engine += sysArrays[i]->cpu_engine; + mem_engine += sysArrays[i]->mem_engine; + net_in += sysArrays[i]->net_in; + net_out += sysArrays[i]->net_out; + io_read += sysArrays[i]->io_read; + io_write += sysArrays[i]->io_write; + io_read_disk += sysArrays[i]->io_read_disk; + io_write_disk += sysArrays[i]->io_write_disk; + } + + double req_select_rate = pStat->numOfSelectReqs / interval; + double req_insert_rate = pStat->numOfInsertReqs / interval; + double req_insert_batch_rate = pStat->numOfBatchInsertReqs / interval; + double net_in_rate = net_in / interval; + double net_out_rate = net_out / interval; + double io_read_rate = io_read / interval; + double io_write_rate = io_write / interval; + double io_read_disk_rate = io_read_disk / interval; + double io_write_disk_rate = io_write_disk / interval; + + metric = taosHashGet(tsMonitor.metrics, UPTIME, strlen(UPTIME)); + taos_gauge_set(*metric, pInfo->uptime, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, CPU_ENGINE, strlen(CPU_ENGINE)); + taos_gauge_set(*metric, cpu_engine, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, CPU_SYSTEM, strlen(CPU_SYSTEM)); + taos_gauge_set(*metric, pSys->cpu_system, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, CPU_CORE, strlen(CPU_CORE)); + taos_gauge_set(*metric, pSys->cpu_cores, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, MEM_ENGINE, strlen(MEM_ENGINE)); + taos_gauge_set(*metric, mem_engine, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, MEM_SYSTEM, strlen(MEM_SYSTEM)); + taos_gauge_set(*metric, pSys->mem_system, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, MEM_TOTAL, strlen(MEM_TOTAL)); + taos_gauge_set(*metric, pSys->mem_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DISK_ENGINE, strlen(DISK_ENGINE)); + taos_gauge_set(*metric, pSys->disk_engine, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DISK_USED, strlen(DISK_USED)); + taos_gauge_set(*metric, pSys->disk_used, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DISK_TOTAL, strlen(DISK_TOTAL)); + taos_gauge_set(*metric, pSys->disk_total, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, NET_IN, strlen(NET_IN)); + taos_gauge_set(*metric, net_in_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, NET_OUT, strlen(NET_OUT)); + taos_gauge_set(*metric, net_out_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, IO_READ, strlen(IO_READ)); + taos_gauge_set(*metric, io_read_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, IO_WRITE, strlen(IO_WRITE)); + taos_gauge_set(*metric, io_write_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, IO_READ_DISK, strlen(IO_READ_DISK)); + taos_gauge_set(*metric, io_read_disk_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, IO_WRITE_DISK, strlen(IO_WRITE_DISK)); + taos_gauge_set(*metric, io_write_disk_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, ERRORS, strlen(ERRORS)); + taos_gauge_set(*metric, io_read_disk_rate, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, VNODES_NUM, strlen(VNODES_NUM)); + taos_gauge_set(*metric, pStat->errors, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, MASTERS, strlen(MASTERS)); + taos_gauge_set(*metric, pStat->masterNum, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, HAS_MNODE, strlen(HAS_MNODE)); + taos_gauge_set(*metric, pInfo->has_mnode, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, HAS_QNODE, strlen(HAS_QNODE)); + taos_gauge_set(*metric, pInfo->has_qnode, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, HAS_SNODE, strlen(HAS_SNODE)); + taos_gauge_set(*metric, pInfo->has_snode, sample_labels); + + //log number + SMonLogs *logs[6]; + logs[0] = &pMonitor->log; + logs[1] = &pMonitor->mmInfo.log; + logs[2] = &pMonitor->vmInfo.log; + logs[3] = &pMonitor->smInfo.log; + logs[4] = &pMonitor->qmInfo.log; + logs[5] = &pMonitor->bmInfo.log; + + int32_t numOfErrorLogs = 0; + int32_t numOfInfoLogs = 0; + int32_t numOfDebugLogs = 0; + int32_t numOfTraceLogs = 0; + + for (int32_t j = 0; j < 6; j++) { + SMonLogs *pLog = logs[j]; + numOfErrorLogs += pLog->numOfErrorLogs; + numOfInfoLogs += pLog->numOfInfoLogs; + numOfDebugLogs += pLog->numOfDebugLogs; + numOfTraceLogs += pLog->numOfTraceLogs; + } + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_ERROR, strlen(DNODE_LOG_ERROR)); + taos_gauge_set(*metric, numOfErrorLogs, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_INFO, strlen(DNODE_LOG_INFO)); + taos_gauge_set(*metric, numOfInfoLogs, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_DEBUG, strlen(DNODE_LOG_DEBUG)); + taos_gauge_set(*metric, numOfDebugLogs, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_TRACE, strlen(DNODE_LOG_TRACE)); + taos_gauge_set(*metric, numOfTraceLogs, sample_labels); +} + +void monGenDnodeStatusInfoTable(SMonInfo *pMonitor){ + if(pMonitor->dmInfo.basic.cluster_id == 0) { + uError("failed to generate dnode info table since cluster_id is 0"); + return; + } + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + char cluster_id[TSDB_CLUSTER_ID_LEN]; + snprintf(cluster_id, TSDB_CLUSTER_ID_LEN, "%"PRId64, pMonitor->dmInfo.basic.cluster_id); + + taos_gauge_t **metric = NULL; + //dnodes status + + SMonClusterInfo *pClusterInfo = &pMonitor->mmInfo.cluster; + + for (int32_t i = 0; i < taosArrayGetSize(pClusterInfo->dnodes); ++i) { + SMonDnodeDesc *pDnodeDesc = taosArrayGet(pClusterInfo->dnodes, i); + + char dnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(dnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pDnodeDesc->dnode_id); + + const char *sample_labels[] = {cluster_id, dnode_id, pDnodeDesc->dnode_ep}; + + metric = taosHashGet(tsMonitor.metrics, DNODE_STATUS, strlen(DNODE_STATUS)); + + int32_t status = 0; + if(strcmp(pDnodeDesc->status, "ready") == 0){ + status = 1; + } + taos_gauge_set(*metric, status, sample_labels); + } +} + +void monGenDataDiskTable(SMonInfo *pMonitor){ + if(pMonitor->dmInfo.basic.cluster_id == 0) return; + + SMonDiskInfo *pInfo = &pMonitor->vmInfo.tfs; + + char cluster_id[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(cluster_id, TSDB_CLUSTER_ID_LEN, "%" PRId64, pMonitor->dmInfo.basic.cluster_id); + + char dnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(dnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pMonitor->dmInfo.basic.dnode_id); + + taos_gauge_t **metric = NULL; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->datadirs); ++i) { + SMonDiskDesc *pDatadirDesc = taosArrayGet(pInfo->datadirs, i); + + char level[LEVEL_LEN] = {0}; + snprintf(level, LEVEL_LEN, "%"PRId32, pDatadirDesc->level); + + const char *sample_labels[] = {cluster_id, dnode_id, pMonitor->dmInfo.basic.dnode_ep, pDatadirDesc->name, level}; + + metric = taosHashGet(tsMonitor.metrics, DNODE_DATA_AVAIL, strlen(DNODE_DATA_AVAIL)); + taos_gauge_set(*metric, pDatadirDesc->size.avail, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_DATA_USED, strlen(DNODE_DATA_USED)); + taos_gauge_set(*metric, pDatadirDesc->size.used, sample_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_DATA_TOTAL, strlen(DNODE_DATA_TOTAL)); + taos_gauge_set(*metric, pDatadirDesc->size.total, sample_labels); + } +} + +void monGenLogDiskTable(SMonInfo *pMonitor){ + if(pMonitor->dmInfo.basic.cluster_id == 0) return; + + SMonDiskDesc *pLogDesc = &pMonitor->dmInfo.dnode.logdir; + SMonDiskDesc *pTempDesc = &pMonitor->dmInfo.dnode.tempdir; + + char cluster_id[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(cluster_id, TSDB_CLUSTER_ID_LEN, "%" PRId64, pMonitor->dmInfo.basic.cluster_id); + + char dnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(dnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pMonitor->dmInfo.basic.dnode_id); + + taos_gauge_t **metric = NULL; + + const char *sample_log_labels[] = {cluster_id, dnode_id, pMonitor->dmInfo.basic.dnode_ep, pLogDesc->name}; + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_AVAIL, strlen(DNODE_LOG_AVAIL)); + taos_gauge_set(*metric, pLogDesc->size.avail, sample_log_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_USED, strlen(DNODE_LOG_USED)); + taos_gauge_set(*metric, pLogDesc->size.used, sample_log_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_TOTAL, strlen(DNODE_LOG_TOTAL)); + taos_gauge_set(*metric, pLogDesc->size.total, sample_log_labels); + + const char *sample_temp_labels[] = {cluster_id, dnode_id, pMonitor->dmInfo.basic.dnode_ep, pTempDesc->name}; + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_AVAIL, strlen(DNODE_LOG_AVAIL)); + taos_gauge_set(*metric, pTempDesc->size.avail, sample_temp_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_USED, strlen(DNODE_LOG_USED)); + taos_gauge_set(*metric, pTempDesc->size.used, sample_temp_labels); + + metric = taosHashGet(tsMonitor.metrics, DNODE_LOG_TOTAL, strlen(DNODE_LOG_TOTAL)); + taos_gauge_set(*metric, pTempDesc->size.total, sample_temp_labels); +} + +void monGenMnodeRoleTable(SMonInfo *pMonitor){ + SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + SMonBasicInfo *pBasicInfo = &pMonitor->dmInfo.basic; + if(pBasicInfo->cluster_id == 0) return; + + char buf[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(buf, TSDB_CLUSTER_ID_LEN, "%" PRId64, pBasicInfo->cluster_id); + + taos_gauge_t **metric = NULL; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->mnodes); ++i) { + + SMonMnodeDesc *pMnodeDesc = taosArrayGet(pInfo->mnodes, i); + + char mnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(mnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pMnodeDesc->mnode_id); + + const char *sample_labels[] = {buf, mnode_id, pMnodeDesc->mnode_ep}; + + metric = taosHashGet(tsMonitor.metrics, MNODE_ROLE, strlen(MNODE_ROLE)); + taos_gauge_set(*metric, pMnodeDesc->syncState, sample_labels); + } +} + +void monGenVnodeRoleTable(SMonInfo *pMonitor){ + SMonVgroupInfo *pInfo = &pMonitor->mmInfo.vgroup; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + SMonBasicInfo *pBasicInfo = &pMonitor->dmInfo.basic; + if(pBasicInfo->cluster_id == 0) return; + + char buf[TSDB_CLUSTER_ID_LEN] = {0}; + snprintf(buf, TSDB_CLUSTER_ID_LEN, "%" PRId64, pBasicInfo->cluster_id); + + taos_gauge_t **metric = NULL; + + for (int32_t i = 0; i < taosArrayGetSize(pInfo->vgroups); ++i) { + SMonVgroupDesc *pVgroupDesc = taosArrayGet(pInfo->vgroups, i); + + char vgroup_id[TSDB_VGROUP_ID_LEN] = {0}; + snprintf(vgroup_id, TSDB_VGROUP_ID_LEN, "%"PRId32, pVgroupDesc->vgroup_id); + + for (int32_t j = 0; j < TSDB_MAX_REPLICA; ++j) { + SMonVnodeDesc *pVnodeDesc = &pVgroupDesc->vnodes[j]; + if (pVnodeDesc->dnode_id <= 0) continue; + + char dnode_id[TSDB_NODE_ID_LEN] = {0}; + snprintf(dnode_id, TSDB_NODE_ID_LEN, "%"PRId32, pVnodeDesc->dnode_id); + + const char *sample_labels[] = {buf, vgroup_id, pVgroupDesc->database_name, dnode_id}; + + metric = taosHashGet(tsMonitor.metrics, VNODE_ROLE, strlen(VNODE_ROLE)); + taos_gauge_set(*metric, pVnodeDesc->syncState, sample_labels); + } + } +} + +void monSendPromReport() { + char ts[50] = {0}; + sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); + + char* promStr = NULL; + char* pCont = (char *)taos_collector_registry_bridge_new(TAOS_COLLECTOR_REGISTRY_DEFAULT, ts, "%" PRId64, &promStr); + if(tsMonitorLogProtocol){ + uInfoL("report cont:\n%s\n", pCont); + uDebugL("report cont prom:\n%s\n", promStr); + } + if (pCont != NULL) { + EHttpCompFlag flag = tsMonitor.cfg.comp ? HTTP_GZIP : HTTP_FLAT; + if (taosSendHttpReport(tsMonitor.cfg.server, tsMonFwUri, tsMonitor.cfg.port, pCont, strlen(pCont), flag) != 0) { + uError("failed to send monitor msg"); + }else{ + taos_collector_registry_clear_batch(TAOS_COLLECTOR_REGISTRY_DEFAULT); + } + taosMemoryFreeClear(pCont); + } + if(promStr != NULL){ + taosMemoryFreeClear(promStr); + } +} \ No newline at end of file diff --git a/source/libs/monitor/src/monMain.c b/source/libs/monitor/src/monMain.c index d417a97db7..83cde72a91 100644 --- a/source/libs/monitor/src/monMain.c +++ b/source/libs/monitor/src/monMain.c @@ -21,9 +21,9 @@ #include "taos_monitor.h" #include "tglobal.h" -static SMonitor tsMonitor = {0}; -static char* tsMonUri = "/report"; -static char* tsMonFwUri = "/td_metric"; +SMonitor tsMonitor = {0}; +char* tsMonUri = "/report"; +char* tsMonFwUri = "/td_metric"; void monRecordLog(int64_t ts, ELogLevel level, const char *content) { taosThreadMutexLock(&tsMonitor.lock); @@ -112,7 +112,7 @@ int32_t monInit(const SMonCfg *pCfg) { tsMonitor.lastTime = taosGetTimestampMs(); taosThreadMutexInit(&tsMonitor.lock, NULL); - taos_collector_registry_default_init(); + monInitMonitorFW(); return 0; } @@ -128,8 +128,7 @@ void monCleanup() { tFreeSMonBmInfo(&tsMonitor.bmInfo); taosThreadMutexDestroy(&tsMonitor.lock); - taos_collector_registry_destroy(TAOS_COLLECTOR_REGISTRY_DEFAULT); - TAOS_COLLECTOR_REGISTRY_DEFAULT = NULL; + monCleanupMonitorFW(); } static void monCleanupMonitorInfo(SMonInfo *pMonitor) { @@ -195,6 +194,22 @@ static void monGenBasicJson(SMonInfo *pMonitor) { tjsonAddDoubleToObject(pJson, "protocol", pInfo->protocol); } +static void monGenBasicJsonBasic(SMonInfo *pMonitor) { + SMonBasicInfo *pInfo = &pMonitor->dmInfo.basic; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + SJson *pJson = pMonitor->pJson; + char buf[40] = {0}; + + sprintf(buf, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); + tjsonAddStringToObject(pJson, "ts", buf); + tjsonAddDoubleToObject(pJson, "dnode_id", pInfo->dnode_id); + tjsonAddStringToObject(pJson, "dnode_ep", pInfo->dnode_ep); + snprintf(buf, sizeof(buf), "%" PRId64, pInfo->cluster_id); + tjsonAddStringToObject(pJson, "cluster_id", buf); + tjsonAddDoubleToObject(pJson, "protocol", pInfo->protocol); +} + static void monGenClusterJson(SMonInfo *pMonitor) { SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster; if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; @@ -253,6 +268,16 @@ static void monGenClusterJson(SMonInfo *pMonitor) { } } +static void monGenClusterJsonBasic(SMonInfo *pMonitor) { + SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; + + tjsonAddStringToObject(pMonitor->pJson, "first_ep", pInfo->first_ep); + tjsonAddDoubleToObject(pMonitor->pJson, "first_ep_dnode_id", pInfo->first_ep_dnode_id); + tjsonAddStringToObject(pMonitor->pJson, "cluster_version", pInfo->version); + tjsonAddDoubleToObject(pMonitor->pJson, "monitor_interval", pInfo->monitor_interval); +} + static void monGenVgroupJson(SMonInfo *pMonitor) { SMonVgroupInfo *pInfo = &pMonitor->mmInfo.vgroup; if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; @@ -526,21 +551,11 @@ static void monGenLogJson(SMonInfo *pMonitor) { if (tjsonAddItemToArray(pSummaryJson, pLogTrace) != 0) tjsonDelete(pLogTrace); } -void monSendReport() { - SMonInfo *pMonitor = monCreateMonitorInfo(); - if (pMonitor == NULL) return; - - monGenBasicJson(pMonitor); - monGenClusterJson(pMonitor); - monGenVgroupJson(pMonitor); - monGenStbJson(pMonitor); - monGenGrantJson(pMonitor); - monGenDnodeJson(pMonitor); - monGenDiskJson(pMonitor); - monGenLogJson(pMonitor); - +void monSendReport(SMonInfo *pMonitor){ char *pCont = tjsonToString(pMonitor->pJson); - // uDebugL("report cont:%s\n", pCont); + if(tsMonitorLogProtocol){ + uInfoL("report cont basic:\n%s", pCont); + } if (pCont != NULL) { EHttpCompFlag flag = tsMonitor.cfg.comp ? HTTP_GZIP : HTTP_FLAT; if (taosSendHttpReport(tsMonitor.cfg.server, tsMonUri, tsMonitor.cfg.port, pCont, strlen(pCont), flag) != 0) { @@ -548,32 +563,49 @@ void monSendReport() { } taosMemoryFree(pCont); } +} + +void monGenAndSendReport() { + SMonInfo *pMonitor = monCreateMonitorInfo(); + if (pMonitor == NULL) return; + + if(!tsMonitorForceV2){ + monGenBasicJson(pMonitor); + monGenClusterJson(pMonitor); + monGenVgroupJson(pMonitor); + monGenStbJson(pMonitor); + monGenGrantJson(pMonitor); + monGenDnodeJson(pMonitor); + monGenDiskJson(pMonitor); + monGenLogJson(pMonitor); + + monSendReport(pMonitor); + } + else{ + monGenClusterInfoTable(pMonitor); + monGenVgroupInfoTable(pMonitor); + monGenDnodeInfoTable(pMonitor); + monGenDnodeStatusInfoTable(pMonitor); + monGenDataDiskTable(pMonitor); + monGenLogDiskTable(pMonitor); + monGenMnodeRoleTable(pMonitor); + monGenVnodeRoleTable(pMonitor); + + monSendPromReport(); + } monCleanupMonitorInfo(pMonitor); } -void monSendPromReport() { - char ts[50]; - sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); - char *pCont = (char *)taos_collector_registry_bridge(TAOS_COLLECTOR_REGISTRY_DEFAULT, ts, "%" PRId64); - //uInfoL("report cont:\n%s\n", pCont); - if (pCont != NULL) { - EHttpCompFlag flag = tsMonitor.cfg.comp ? HTTP_GZIP : HTTP_FLAT; - if (taosSendHttpReport(tsMonitor.cfg.server, tsMonFwUri, tsMonitor.cfg.port, pCont, strlen(pCont), flag) != 0) { - uError("failed to send monitor msg"); - }else{ - taos_collector_registry_clear_out(TAOS_COLLECTOR_REGISTRY_DEFAULT); - } - } -} +void monGenAndSendReportBasic() { + SMonInfo *pMonitor = monCreateMonitorInfo(); + if (pMonitor == NULL) return; + if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; -void monSendContent(char *pCont) { - if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return; - //uInfoL("report cont:\n%s\n", pCont); - if (pCont != NULL) { - EHttpCompFlag flag = tsMonitor.cfg.comp ? HTTP_GZIP : HTTP_FLAT; - if (taosSendHttpReport(tsMonitor.cfg.server, tsMonFwUri, tsMonitor.cfg.port, pCont, strlen(pCont), flag) != 0) { - uError("failed to send monitor msg"); - } - } + monGenBasicJsonBasic(pMonitor); + monGenClusterJsonBasic(pMonitor); + + monSendReport(pMonitor); + + monCleanupMonitorInfo(pMonitor); } \ No newline at end of file diff --git a/source/libs/monitor/test/monTest.cpp b/source/libs/monitor/test/monTest.cpp index 3f7b1b51da..2660cff216 100644 --- a/source/libs/monitor/test/monTest.cpp +++ b/source/libs/monitor/test/monTest.cpp @@ -283,10 +283,10 @@ TEST_F(MonitorTest, 01_Full) { tFreeSMonSmInfo(&smInfo); tFreeSMonQmInfo(&qmInfo); tFreeSMonBmInfo(&bmInfo); - monSendReport(); + monGenAndSendReport(); } TEST_F(MonitorTest, 02_Log) { AddLogInfo2(); - monSendReport(); + monGenAndSendReport(); } diff --git a/source/libs/monitorfw/inc/taos_collector_registry_t.h b/source/libs/monitorfw/inc/taos_collector_registry_t.h index 2264d18081..8e8a881fca 100644 --- a/source/libs/monitorfw/inc/taos_collector_registry_t.h +++ b/source/libs/monitorfw/inc/taos_collector_registry_t.h @@ -34,7 +34,7 @@ struct taos_collector_registry { taos_string_builder_t *string_builder; /**< Enables string building */ taos_metric_formatter_t *metric_formatter; /**< metric formatter for metric exposition on bridge call */ pthread_rwlock_t *lock; /**< mutex for safety against concurrent registration */ - taos_string_builder_t *out; + taos_string_builder_t *string_builder_batch; }; #endif // TAOS_REGISTRY_T_H diff --git a/source/libs/monitorfw/inc/taos_metric_formatter_custom_i.h b/source/libs/monitorfw/inc/taos_metric_formatter_custom_i.h new file mode 100644 index 0000000000..f3e4ebae75 --- /dev/null +++ b/source/libs/monitorfw/inc/taos_metric_formatter_custom_i.h @@ -0,0 +1,31 @@ +/* + * 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 TAOS_METRIC_FORMATTER_CUSTOMV2_I_H +#define TAOS_METRIC_FORMATTER_CUSTOMV2_I_H + +#define ALLOW_FORBID_FUNC + +#include +#include "tjson.h" + +int taos_metric_formatter_load_sample_new(taos_metric_formatter_t *self, taos_metric_sample_t *sample, + char *ts, char *format, char *metricName, int32_t metric_type, + SJson *arrayMetricGroups); +int taos_metric_formatter_load_metric_new(taos_metric_formatter_t *self, taos_metric_t *metric, char *ts, char *format, + SJson* tableArray); +int taos_metric_formatter_load_metrics_new(taos_metric_formatter_t *self, taos_map_t *collectors, char *ts, + char *format, SJson* tableArray); +#endif // TAOS_METRIC_FORMATTER_CUSTOMV2_I_H \ No newline at end of file diff --git a/source/libs/monitorfw/inc/taos_metric_sample_t.h b/source/libs/monitorfw/inc/taos_metric_sample_t.h index 92885acf4a..3aa9b7bb99 100644 --- a/source/libs/monitorfw/inc/taos_metric_sample_t.h +++ b/source/libs/monitorfw/inc/taos_metric_sample_t.h @@ -19,10 +19,16 @@ #include "taos_metric_sample.h" #include "taos_metric_t.h" +#define DOUBLE_ATOMIC + struct taos_metric_sample { taos_metric_type_t type; /**< type is the metric type for the sample */ char *l_value; /**< l_value is the full metric name and label set represeted as a string */ - /*_Atomic*/ int64_t r_value; /**< r_value is the value of the metric sample */ +#ifdef DOUBLE_ATOMIC + _Atomic double r_value; /**< r_value is the value of the metric sample */ +#else + int64_t r_value; /**< r_value is the value of the metric sample */ +#endif }; #endif // TAOS_METRIC_SAMPLE_T_H diff --git a/source/libs/monitorfw/inc/taos_metric_t.h b/source/libs/monitorfw/inc/taos_metric_t.h index 806466528d..da237aa814 100644 --- a/source/libs/monitorfw/inc/taos_metric_t.h +++ b/source/libs/monitorfw/inc/taos_metric_t.h @@ -42,7 +42,7 @@ extern char *taos_metric_type_map[4]; */ struct taos_metric { taos_metric_type_t type; /**< metric_type The type of metric */ - const char *name; /**< name The name of the metric */ + char *name; /**< name The name of the metric */ const char *help; /**< help The help output for the metric */ taos_map_t *samples; /**< samples Map comprised of samples for the given metric */ size_t label_key_count; /**< label_keys_count The count of labe_keys*/ diff --git a/source/libs/monitorfw/inc/taos_monitor_util_i.h b/source/libs/monitorfw/inc/taos_monitor_util_i.h new file mode 100644 index 0000000000..fe072204a3 --- /dev/null +++ b/source/libs/monitorfw/inc/taos_monitor_util_i.h @@ -0,0 +1,26 @@ +/* + * 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 TAOS_MONITOR_UTIL_I_H +#define TAOS_MONITOR_UTIL_I_H + +#include + +void taos_monitor_split_str(char** arr, char* str, const char* del); +int taos_monitor_count_occurrences(char *str, char *toSearch); +void taos_monitor_strip(char *s); +bool taos_monitor_is_match(const SJson* tags, char** pairs, int32_t count); + +#endif // TAOS_MONITOR_UTIL_I_H \ No newline at end of file diff --git a/source/libs/monitorfw/src/taos_collector.c b/source/libs/monitorfw/src/taos_collector.c index d3228facb6..997bf5587c 100644 --- a/source/libs/monitorfw/src/taos_collector.c +++ b/source/libs/monitorfw/src/taos_collector.c @@ -107,3 +107,9 @@ int taos_collector_add_metric(taos_collector_t *self, taos_metric_t *metric) { } return taos_map_set(self->metrics, metric->name, metric); } + +taos_metric_t* taos_collector_get_metric(taos_collector_t *self, char *metric_name){ + TAOS_ASSERT(self != NULL); + if (self == NULL) return NULL; + return taos_map_get(self->metrics, metric_name); +} \ No newline at end of file diff --git a/source/libs/monitorfw/src/taos_collector_registry.c b/source/libs/monitorfw/src/taos_collector_registry.c index 711e66ed84..c94675c95c 100644 --- a/source/libs/monitorfw/src/taos_collector_registry.c +++ b/source/libs/monitorfw/src/taos_collector_registry.c @@ -33,6 +33,10 @@ #include "taos_metric_i.h" #include "taos_metric_t.h" #include "taos_string_builder_i.h" +#include "taos_metric_formatter_custom_i.h" + +#define ALLOW_FORBID_FUNC +#include "tjson.h" taos_collector_registry_t *TAOS_COLLECTOR_REGISTRY_DEFAULT; @@ -50,7 +54,7 @@ taos_collector_registry_t *taos_collector_registry_new(const char *name) { self->metric_formatter = taos_metric_formatter_new(); self->string_builder = taos_string_builder_new(); - self->out = taos_string_builder_new(); + self->string_builder_batch = taos_string_builder_new(); self->lock = (pthread_rwlock_t *)taos_malloc(sizeof(pthread_rwlock_t)); r = pthread_rwlock_init(self->lock, NULL); if (r) { @@ -88,8 +92,8 @@ int taos_collector_registry_destroy(taos_collector_registry_t *self) { self->string_builder = NULL; if (r) ret = r; - r = taos_string_builder_destroy(self->out); - self->out = NULL; + r = taos_string_builder_destroy(self->string_builder_batch); + self->string_builder_batch = NULL; if (r) ret = r; r = pthread_rwlock_destroy(self->lock); @@ -119,6 +123,19 @@ int taos_collector_registry_register_metric(taos_metric_t *metric) { return taos_collector_add_metric(default_collector, metric); } +taos_metric_t *taos_collector_registry_get_metric(char* metric_name){ + TAOS_ASSERT(metric != NULL); + + taos_collector_t *default_collector = + (taos_collector_t *)taos_map_get(TAOS_COLLECTOR_REGISTRY_DEFAULT->collectors, "default"); + + if (default_collector == NULL) { + return NULL; + } + + return taos_collector_get_metric(default_collector, metric_name); +} + taos_metric_t *taos_collector_registry_must_register_metric(taos_metric_t *metric) { int err = taos_collector_registry_register_metric(metric); if (err != 0) { @@ -193,13 +210,71 @@ const char *taos_collector_registry_bridge(taos_collector_registry_t *self, char char *out = taos_metric_formatter_dump(self->metric_formatter); int r = 0; - r = taos_string_builder_add_str(self->out, out); + r = taos_string_builder_add_str(self->string_builder_batch, out); if (r) return NULL; taos_free(out); - return taos_string_builder_str(self->out); + return taos_string_builder_str(self->string_builder_batch); } -int taos_collector_registry_clear_out(taos_collector_registry_t *self){ - return taos_string_builder_clear(self->out); +int taos_collector_registry_clear_batch(taos_collector_registry_t *self){ + return taos_string_builder_clear(self->string_builder_batch); +} + +const char *taos_collector_registry_bridge_new(taos_collector_registry_t *self, char *ts, char *format, char** prom_str) { + taos_metric_formatter_clear(self->metric_formatter); + + SJson* pJson = tjsonCreateArray(); + SJson* item = tjsonCreateObject(); + tjsonAddItemToArray(pJson, item); + tjsonAddStringToObject(item, "ts", ts); + tjsonAddDoubleToObject(item, "protocol", 2); + SJson* array = tjsonCreateArray(); + tjsonAddItemToObject(item, "tables", array); + + taos_metric_formatter_load_metrics_new(self->metric_formatter, self->collectors, ts, format, array); + + //caller free this + //generate prom protocol for debug + if(prom_str != NULL){ + *prom_str = taos_metric_formatter_dump(self->metric_formatter); + } + + //add this result to batch cache, format in batch cache is {},{} + int r = 0; + char* old_str = taos_string_builder_str(self->string_builder_batch); + if(old_str[0] != '\0'){ + r = taos_string_builder_add_str(self->string_builder_batch, ","); + if (r) return NULL; + } + char * item_str = tjsonToString(item); + r = taos_string_builder_add_str(self->string_builder_batch, item_str); + taos_free(item_str); + if (r) return NULL; + + tjsonDelete(pJson); + + //generate final array format result, ie, add [] to str in batch cache + taos_string_builder_t* tmp_builder = taos_string_builder_new(); + + r = taos_string_builder_add_str(tmp_builder, "["); + if (r) return NULL; + + r = taos_string_builder_add_str(tmp_builder, taos_string_builder_str(self->string_builder_batch)); + if (r) return NULL; + + r = taos_string_builder_add_str(tmp_builder, "]"); + if (r) return NULL; + + //caller free this + char *data = taos_string_builder_dump(tmp_builder); + if (data == NULL) return NULL; + r = taos_string_builder_clear(tmp_builder); + if (r) return NULL; + + r = taos_string_builder_destroy(tmp_builder); + tmp_builder = NULL; + if (r) return NULL; + + return data; } diff --git a/source/libs/monitorfw/src/taos_gauge.c b/source/libs/monitorfw/src/taos_gauge.c new file mode 100644 index 0000000000..74d2665194 --- /dev/null +++ b/source/libs/monitorfw/src/taos_gauge.c @@ -0,0 +1,100 @@ +/* + * 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 . + */ + +// Public +#include "taos_gauge.h" + +#include "taos_alloc.h" + +// Private +#include "taos_assert.h" +#include "taos_errors.h" +#include "taos_log.h" +#include "taos_metric_i.h" +#include "taos_metric_sample_i.h" +#include "taos_metric_sample_t.h" +#include "taos_metric_t.h" + +taos_gauge_t *taos_gauge_new(const char *name, const char *help, size_t label_key_count, const char **label_keys) { + return (taos_gauge_t *)taos_metric_new(TAOS_GAUGE, name, help, label_key_count, label_keys); +} + +int taos_gauge_destroy(taos_gauge_t *self) { + TAOS_ASSERT(self != NULL); + int r = 0; + r = taos_metric_destroy(self); + self = NULL; + return r; +} + +int taos_gauge_inc(taos_gauge_t *self, const char **label_values) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + if (self->type != TAOS_GAUGE) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + taos_metric_sample_t *sample = taos_metric_sample_from_labels(self, label_values); + if (sample == NULL) return 1; + return taos_metric_sample_add(sample, 1.0); +} + +int taos_gauge_dec(taos_gauge_t *self, const char **label_values) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + if (self->type != TAOS_GAUGE) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + taos_metric_sample_t *sample = taos_metric_sample_from_labels(self, label_values); + if (sample == NULL) return 1; + return taos_metric_sample_sub(sample, 1.0); +} + +int taos_gauge_add(taos_gauge_t *self, double r_value, const char **label_values) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + if (self->type != TAOS_GAUGE) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + taos_metric_sample_t *sample = taos_metric_sample_from_labels(self, label_values); + if (sample == NULL) return 1; + return taos_metric_sample_add(sample, r_value); +} + +int taos_gauge_sub(taos_gauge_t *self, double r_value, const char **label_values) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + if (self->type != TAOS_GAUGE) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + taos_metric_sample_t *sample = taos_metric_sample_from_labels(self, label_values); + if (sample == NULL) return 1; + return taos_metric_sample_sub(sample, r_value); +} + +int taos_gauge_set(taos_gauge_t *self, double r_value, const char **label_values) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + if (self->type != TAOS_GAUGE) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + taos_metric_sample_t *sample = taos_metric_sample_from_labels(self, label_values); + if (sample == NULL) return 1; + return taos_metric_sample_set(sample, r_value); +} diff --git a/source/libs/monitorfw/src/taos_metric.c b/source/libs/monitorfw/src/taos_metric.c index a9bb6d45a4..4e9af35f34 100644 --- a/source/libs/monitorfw/src/taos_metric.c +++ b/source/libs/monitorfw/src/taos_metric.c @@ -34,7 +34,11 @@ taos_metric_t *taos_metric_new(taos_metric_type_t metric_type, const char *name, int r = 0; taos_metric_t *self = (taos_metric_t *)taos_malloc(sizeof(taos_metric_t)); self->type = metric_type; - self->name = name; + int len = strlen(name) + 1; + self->name = taos_malloc(len); + memset(self->name, 0, len); + strcpy(self->name, name); + //self->name = name; self->help = help; const char **k = (const char **)taos_malloc(sizeof(const char *) * label_key_count); @@ -111,6 +115,9 @@ int taos_metric_destroy(taos_metric_t *self) { taos_free(self->label_keys); self->label_keys = NULL; + taos_free(self->name); + self->name = NULL; + taos_free(self); self = NULL; diff --git a/source/libs/monitorfw/src/taos_metric_formatter.c b/source/libs/monitorfw/src/taos_metric_formatter.c index 6c6f1974eb..a9f35c3e8d 100644 --- a/source/libs/monitorfw/src/taos_metric_formatter.c +++ b/source/libs/monitorfw/src/taos_metric_formatter.c @@ -181,7 +181,7 @@ int taos_metric_formatter_load_sample(taos_metric_formatter_t *self, taos_metric r = taos_string_builder_add_str(self->string_builder, ts); if (r) return r; - taos_metric_sample_set(sample, 0); + //taos_metric_sample_set(sample, 0); return taos_string_builder_add_char(self->string_builder, '\n'); } diff --git a/source/libs/monitorfw/src/taos_metric_formatter_custom.c b/source/libs/monitorfw/src/taos_metric_formatter_custom.c new file mode 100644 index 0000000000..d7650098e0 --- /dev/null +++ b/source/libs/monitorfw/src/taos_metric_formatter_custom.c @@ -0,0 +1,234 @@ +/* + * 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 . + */ + +#define ALLOW_FORBID_FUNC + +#include +#include "taos_metric_formatter_i.h" +#include "taos_metric_sample_t.h" +#include "tjson.h" +#include "taos_monitor_util_i.h" +#include "taos_assert.h" +#include "tdef.h" +#include "taos_collector_t.h" + +int taos_metric_formatter_load_sample_new(taos_metric_formatter_t *self, taos_metric_sample_t *sample, + char *ts, char *format, char *metricName, int32_t metric_type, + SJson *arrayMetricGroups) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + + int r = 0; + + char* start = strstr(sample->l_value, "{"); + char* end = strstr(sample->l_value, "}"); + + int32_t len = end -start; + + char* keyvalues = taosMemoryMalloc(len); + memset(keyvalues, 0, len); + memcpy(keyvalues, start + 1, len - 1); + + int32_t count = taos_monitor_count_occurrences(keyvalues, ","); + + char** keyvalue = taosMemoryMalloc(sizeof(char*) * (count + 1)); + memset(keyvalue, 0, sizeof(char*) * (count + 1)); + taos_monitor_split_str(keyvalue, keyvalues, ","); + + char** arr = taosMemoryMalloc(sizeof(char*) * (count + 1) * 2); + memset(arr, 0, sizeof(char*) * (count + 1) * 2); + + bool isfound = true; + for(int32_t i = 0; i < count + 1; i++){ + char* str = *(keyvalue + i); + + char** pair = arr + i * 2; + taos_monitor_split_str(pair, str, "="); + + taos_monitor_strip(pair[1]); + } + + int32_t table_size = tjsonGetArraySize(arrayMetricGroups); + + SJson* item = NULL; + for(int32_t i = 0; i < table_size; i++){ + SJson *cur = tjsonGetArrayItem(arrayMetricGroups, i); + + SJson* tag = tjsonGetObjectItem(cur, "tags"); + + if(taos_monitor_is_match(tag, arr, count + 1)) { + item = cur; + break; + } + } + + SJson* metrics = NULL; + if(item == NULL) { + item = tjsonCreateObject(); + + SJson* arrayTag = tjsonCreateArray(); + for(int32_t i = 0; i < count + 1; i++){ + char** pair = arr + i * 2; + + char* key = *pair; + char* value = *(pair + 1); + + SJson* tag = tjsonCreateObject(); + tjsonAddStringToObject(tag, "name", key); + tjsonAddStringToObject(tag, "value", value); + + tjsonAddItemToArray(arrayTag, tag); + } + tjsonAddItemToObject(item, "tags", arrayTag); + + metrics = tjsonCreateArray(); + tjsonAddItemToObject(item, "metrics", metrics); + + tjsonAddItemToArray(arrayMetricGroups, item); + } + else{ + metrics = tjsonGetObjectItem(item, "metrics"); + } + + taosMemoryFreeClear(arr); + taosMemoryFreeClear(keyvalue); + taosMemoryFreeClear(keyvalues); + + SJson* metric = tjsonCreateObject(); + tjsonAddStringToObject(metric, "name", metricName); + + double old_value = 0; +#define USE_EXCHANGE +#ifdef USE_EXCHANGE + taos_metric_sample_exchange(sample, 0, &old_value); +#else + old_value = sample->r_value; + taos_metric_sample_set(sample, 0); +#endif + + tjsonAddDoubleToObject(metric, "value", old_value); + tjsonAddDoubleToObject(metric, "type", metric_type); + tjsonAddItemToArray(metrics, metric); + + return 0; +} + +int taos_metric_formatter_load_metric_new(taos_metric_formatter_t *self, taos_metric_t *metric, char *ts, char *format, + SJson* tableArray) { + TAOS_ASSERT(self != NULL); + if (self == NULL) return 1; + + int r = 0; + + int32_t size = strlen(metric->name); + char* name = taosMemoryMalloc(size + 1); + memset(name, 0, size + 1); + memcpy(name, metric->name, size); + char* arr[2] = {0}; //arr[0] is table name, arr[1] is metric name + taos_monitor_split_str((char**)&arr, name, ":"); + + bool isFound = false; + SJson* table = NULL; + SJson* arrayMetricGroups = NULL; + + int32_t table_count = tjsonGetArraySize(tableArray); + for(int32_t i = 0; i < table_count; i++){ + SJson* table = tjsonGetArrayItem(tableArray, i); + + char tableName[MONITOR_TABLENAME_LEN] = {0}; + tjsonGetStringValue(table, "name", tableName); + if(strcmp(tableName, arr[0]) == 0){ + isFound = true; + arrayMetricGroups = tjsonGetObjectItem(table, "metric_groups"); + break; + } + } + + if(!isFound){ + table = tjsonCreateObject(); + + tjsonAddStringToObject(table, "name", arr[0]); + + arrayMetricGroups = tjsonCreateArray(); + tjsonAddItemToObject(table, "metric_groups", arrayMetricGroups); + } + + int32_t sample_count = 0; + for (taos_linked_list_node_t *current_node = metric->samples->keys->head; current_node != NULL; + current_node = current_node->next) { + const char *key = (const char *)current_node->item; + if (metric->type == TAOS_HISTOGRAM) { + + } else { + taos_metric_sample_t *sample = (taos_metric_sample_t *)taos_map_get(metric->samples, key); + if (sample == NULL) return 1; + r = taos_metric_formatter_load_sample_new(self, sample, ts, format, arr[1], metric->type, arrayMetricGroups); + if (r) return r; + } + sample_count++; + } + + if(!isFound && sample_count > 0){ + tjsonAddItemToArray(tableArray, table); + } + else{ + if(table != NULL) tjsonDelete(table); + } + + taosMemoryFreeClear(name); + return r; +} + +int taos_metric_formatter_load_metrics_new(taos_metric_formatter_t *self, taos_map_t *collectors, char *ts, + char *format, SJson* tableArray) { + TAOS_ASSERT(self != NULL); + int r = 0; + + for (taos_linked_list_node_t *current_node = collectors->keys->head; current_node != NULL; + current_node = current_node->next) { + const char *collector_name = (const char *)current_node->item; + taos_collector_t *collector = (taos_collector_t *)taos_map_get(collectors, collector_name); + if (collector == NULL) return 1; + + taos_map_t *metrics = collector->collect_fn(collector); + if (metrics == NULL) return 1; + + //if(strcmp(collector->name, "custom") != 0 ){ + + for (taos_linked_list_node_t *current_node = metrics->keys->head; current_node != NULL; + current_node = current_node->next) { + const char *metric_name = (const char *)current_node->item; + taos_metric_t *metric = (taos_metric_t *)taos_map_get(metrics, metric_name); + if (metric == NULL) return 1; + r = taos_metric_formatter_load_metric_new(self, metric, ts, format, tableArray); + if (r) return r; + } + + //} + //else{ + + for (taos_linked_list_node_t *current_node = metrics->keys->head; current_node != NULL; + current_node = current_node->next) { + const char *metric_name = (const char *)current_node->item; + taos_metric_t *metric = (taos_metric_t *)taos_map_get(metrics, metric_name); + if (metric == NULL) return 1; + r = taos_metric_formatter_load_metric(self, metric, ts, format); + if (r) return r; + } + + //} + } + return r; +} \ No newline at end of file diff --git a/source/libs/monitorfw/src/taos_metric_sample.c b/source/libs/monitorfw/src/taos_metric_sample.c index 034657cb3d..04a6045e76 100644 --- a/source/libs/monitorfw/src/taos_metric_sample.c +++ b/source/libs/monitorfw/src/taos_metric_sample.c @@ -13,7 +13,7 @@ * along with this program. If not, see . */ -//#include + // Public #include "taos_alloc.h" @@ -25,10 +25,14 @@ #include "taos_metric_sample_i.h" #include "taos_metric_sample_t.h" -#define ALLOW_FORBID_FUNC +#define DOUBLE_ATOMIC -#include "tdef.h" +#ifdef DOUBLE_ATOMIC +#include +#else +#define ALLOW_FORBID_FUNC #include "osAtomic.h" +#endif taos_metric_sample_t *taos_metric_sample_new(taos_metric_type_t type, const char *l_value, double r_value) { taos_metric_sample_t *self = (taos_metric_sample_t *)taos_malloc(sizeof(taos_metric_sample_t)); @@ -67,7 +71,8 @@ int taos_metric_sample_add(taos_metric_sample_t *self, double r_value) { if (r_value < 0) { return 1; } - /* + +#ifdef DOUBLE_ATOMIC _Atomic double old = atomic_load(&self->r_value); for (;;) { @@ -76,8 +81,10 @@ int taos_metric_sample_add(taos_metric_sample_t *self, double r_value) { return 0; } } - */ +#else atomic_fetch_add_64(&self->r_value, r_value); +#endif + return 0; } @@ -87,7 +94,8 @@ int taos_metric_sample_sub(taos_metric_sample_t *self, double r_value) { TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); return 1; } - /* + +#ifdef DOUBLE_ATOMIC _Atomic double old = atomic_load(&self->r_value); for (;;) { _Atomic double new = ATOMIC_VAR_INIT(old - r_value); @@ -95,8 +103,10 @@ int taos_metric_sample_sub(taos_metric_sample_t *self, double r_value) { return 0; } } - */ +#else atomic_fetch_sub_64(&self->r_value, r_value); +#endif + return 0; } @@ -105,9 +115,34 @@ int taos_metric_sample_set(taos_metric_sample_t *self, double r_value) { TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); return 1; } - /* + +#ifdef DOUBLE_ATOMIC atomic_store(&self->r_value, r_value); - */ +#else atomic_store_64(&self->r_value, r_value); +#endif + return 0; } + +int taos_metric_sample_exchange(taos_metric_sample_t *self, double r_value, double* old_value) { + if (self->type != TAOS_GAUGE && self->type != TAOS_COUNTER) { + TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE); + return 1; + } + +#ifdef DOUBLE_ATOMIC + _Atomic double new = ATOMIC_VAR_INIT(r_value); + for (;;) { + _Atomic double old = atomic_load(&self->r_value); + *old_value = old; + if (atomic_compare_exchange_weak(&self->r_value, &old, new)) { + return 0; + } + } +#else + *old_value = atomic_exchange_64(&self->r_value, r_value); +#endif + + return 0; +} \ No newline at end of file diff --git a/source/libs/monitorfw/src/taos_monitor_util.c b/source/libs/monitorfw/src/taos_monitor_util.c new file mode 100644 index 0000000000..182402b3ff --- /dev/null +++ b/source/libs/monitorfw/src/taos_monitor_util.c @@ -0,0 +1,110 @@ +/* + * 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 . + */ + + +#define ALLOW_FORBID_FUNC +#include "tjson.h" +#include +#include +#include +#include "osMemory.h" +#include "tdef.h" + +#include "taos_metric_t.h" + +void taos_monitor_split_str(char** arr, char* str, const char* del) { + char *lasts; + char* s = strsep(&str, del); + while (s != NULL) { + *arr++ = s; + s = strsep(&str, del); + } +} + +void taos_monitor_split_str_metric(char** arr, taos_metric_t* metric, const char* del, char** buf) { + int32_t size = strlen(metric->name); + char* name = taosMemoryMalloc(size + 1); + memset(name, 0, size + 1); + memcpy(name, metric->name, size); + + char* s = strtok(name, del); + while (s != NULL) { + *arr++ = s; + s = strtok(NULL, del); + } + + *buf = name; +} + +const char* taos_monitor_get_metric_name(taos_metric_t* metric){ + return metric->name; +} + +int taos_monitor_count_occurrences(char *str, char *toSearch) { + int count = 0; + char *ptr = str; + while ((ptr = strstr(ptr, toSearch)) != NULL) { + count++; + ptr++; + } + return count; +} + +void taos_monitor_strip(char *s) +{ + size_t i; + size_t len = strlen(s); + size_t offset = 0; + for(i = 0; i < len; ++i){ + char c = s[i]; + if(c=='\"') ++offset; + else s[i-offset] = c; + } + s[len-offset] = '\0'; +} + +bool taos_monitor_is_match(const SJson* tags, char** pairs, int32_t count) { + int32_t size = tjsonGetArraySize(tags); + if(size != count) return false; + + for(int32_t i = 0; i < size; i++){ + SJson* item = tjsonGetArrayItem(tags, i); + + char item_name[MONITOR_TAG_NAME_LEN] = {0}; + tjsonGetStringValue(item, "name", item_name); + + char item_value[MONITOR_TAG_VALUE_LEN] = {0}; + tjsonGetStringValue(item, "value", item_value); + + bool isfound = false; + for(int32_t j = 0; j < count; j++){ + + char** pair = pairs + j * 2; + + char* key = *pair; + char* value = *(pair + 1); + + + if(strcmp(value, item_value) == 0 && strcmp(key, item_name) == 0){ + isfound = true; + break; + } + } + + if(!isfound) return false; + } + + return true; +}