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

new table structure

This commit is contained in:
dmchen 2024-01-24 09:44:44 +00:00
parent 39f2e75a9e
commit d9af07de6f
43 changed files with 1878 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/tglobal.h
View File

@ -105,6 +105,9 @@ extern char tsMonitorFqdn[];
extern uint16_t tsMonitorPort; extern uint16_t tsMonitorPort;
extern int32_t tsMonitorMaxLogs; extern int32_t tsMonitorMaxLogs;
extern bool tsMonitorComp; extern bool tsMonitorComp;
extern bool tsMonitorLogProtocol;
extern int32_t tsMonitorIntervalForBasic;
extern bool tsMonitorForceV2;
// audit // audit
extern bool tsEnableAudit; extern bool tsEnableAudit;

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

@ -100,6 +100,7 @@ typedef struct {
int32_t mnode_id; int32_t mnode_id;
char mnode_ep[TSDB_EP_LEN]; char mnode_ep[TSDB_EP_LEN];
char role[MON_ROLE_LEN]; char role[MON_ROLE_LEN];
int32_t syncState;
} SMonMnodeDesc; } SMonMnodeDesc;
typedef struct { typedef struct {
@ -125,6 +126,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t dnode_id; int32_t dnode_id;
char vnode_role[MON_ROLE_LEN]; char vnode_role[MON_ROLE_LEN];
int32_t syncState;
} SMonVnodeDesc; } SMonVnodeDesc;
typedef struct { typedef struct {
@ -221,9 +223,8 @@ void monSetVmInfo(SMonVmInfo *pInfo);
void monSetQmInfo(SMonQmInfo *pInfo); void monSetQmInfo(SMonQmInfo *pInfo);
void monSetSmInfo(SMonSmInfo *pInfo); void monSetSmInfo(SMonSmInfo *pInfo);
void monSetBmInfo(SMonBmInfo *pInfo); void monSetBmInfo(SMonBmInfo *pInfo);
void monSendReport(); void monGenAndSendReport();
void monSendPromReport(); void monGenAndSendReportBasic();
void monSendContent(char *pCont);
void tFreeSMonMmInfo(SMonMmInfo *pInfo); void tFreeSMonMmInfo(SMonMmInfo *pInfo);
void tFreeSMonVmInfo(SMonVmInfo *pInfo); void tFreeSMonVmInfo(SMonVmInfo *pInfo);

2
include/libs/monitorfw/taos_collector.h
View File

@ -76,6 +76,8 @@ int taos_collector_destroy_generic(void *gen);
*/ */
int taos_collector_add_metric(taos_collector_t *self, taos_metric_t *metric); 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 * @brief The collect function is responsible for doing any work involving a set of metrics and then returning them
* for metric exposition. * for metric exposition.

6
include/libs/monitorfw/taos_collector_registry.h
View File

@ -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); 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. * @brief Register a collector with the given registry. Returns a non-zero integer value on failure.
* @param self The target taos_collector_registry_t* * @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); 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: *@brief Validates that the given metric name complies with the specification:

154
include/libs/monitorfw/taos_gauge.h Normal file
View File

@ -0,0 +1,154 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
/**
* @file taos_gauge.h
* @brief https://prometheus.io/docs/concepts/metric_types/#gauge
*/
#ifndef TAOS_GAUGE_H
#define TAOS_GAUGE_H
#include <stdlib.h>
#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

2
include/libs/monitorfw/taos_metric_sample.h
View File

@ -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_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 #endif // TAOS_METRIC_SAMPLE_H

2
include/libs/monitorfw/taos_monitor.h
View File

@ -126,5 +126,7 @@ limitations under the License.
#include "taos_map.h" #include "taos_map.h"
#include "taos_metric.h" #include "taos_metric.h"
#include "taos_metric_sample.h" #include "taos_metric_sample.h"
#include "taos_monitor_util.h"
#include "taos_gauge.h"
#endif // TAOS_INCLUDED #endif // TAOS_INCLUDED

31
include/libs/monitorfw/taos_monitor_util.h Normal file
View File

@ -0,0 +1,31 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
/**
* @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

6
include/util/tdef.h
View File

@ -196,6 +196,8 @@ typedef enum ELogicConditionType {
// ACCOUNT is a 32 bit positive integer // ACCOUNT is a 32 bit positive integer
// this is the length of its string representation, including the terminator zero // this is the length of its string representation, including the terminator zero
#define TSDB_ACCT_ID_LEN 11 #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_MAX_COLUMNS 4096
#define TSDB_MIN_COLUMNS 2 // PRIMARY COLUMN(timestamp) + other columns #define TSDB_MIN_COLUMNS 2 // PRIMARY COLUMN(timestamp) + other columns
@ -543,6 +545,10 @@ enum {
#define VNODE_TIMEOUT_SEC 60 #define VNODE_TIMEOUT_SEC 60
#define MNODE_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 #ifdef __cplusplus
} }
#endif #endif

5
source/client/src/clientEnv.c
View File

@ -18,6 +18,7 @@
#include "clientLog.h" #include "clientLog.h"
#include "functionMgt.h" #include "functionMgt.h"
#include "os.h" #include "os.h"
#include "osSleep.h"
#include "query.h" #include "query.h"
#include "qworker.h" #include "qworker.h"
#include "scheduler.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 num = atomic_sub_fetch_32(&pTscObj->numOfReqs, 1);
int32_t reqType = SLOW_LOG_TYPE_OTHERS; int32_t reqType = SLOW_LOG_TYPE_OTHERS;
//taosSsleep(3);
int64_t duration = taosGetTimestampUs() - pRequest->metric.start; int64_t duration = taosGetTimestampUs() - pRequest->metric.start;
tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64
" elapsed:%.2f ms, " " elapsed:%.2f ms, "
@ -108,6 +111,8 @@ static void deregisterRequest(SRequestObj *pRequest) {
} }
} }
duration = 7000000;
if (duration >= (tsSlowLogThreshold * 1000000UL)) { if (duration >= (tsSlowLogThreshold * 1000000UL)) {
atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1); atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
if (tsSlowLogScope & reqType) { if (tsSlowLogScope & reqType) {

2
source/client/src/selectMonitor.c
View File

@ -16,7 +16,7 @@
#include "clientMonitor.h" #include "clientMonitor.h"
#include "clientLog.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 char* selectMonitorHelp = "slow query log when cost > 3s";
const int selectMonitorLabelCount = 1; const int selectMonitorLabelCount = 1;
const char* selectMonitorLabels[] = {"default"}; const char* selectMonitorLabels[] = {"default"};

2
source/client/src/slowQueryMonitor.c
View File

@ -17,7 +17,7 @@
#include "clientLog.h" #include "clientLog.h"
#include "tglobal.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 char* slowQueryHelp = "slow query log when cost > 3s";
const int slowQueryLabelCount = 1; const int slowQueryLabelCount = 1;
const char* slowQueryLabels[] = {"cost"}; const char* slowQueryLabels[] = {"cost"};

10
source/common/src/tglobal.c
View File

@ -93,6 +93,9 @@ char tsMonitorFqdn[TSDB_FQDN_LEN] = {0};
uint16_t tsMonitorPort = 6043; uint16_t tsMonitorPort = 6043;
int32_t tsMonitorMaxLogs = 100; int32_t tsMonitorMaxLogs = 100;
bool tsMonitorComp = false; bool tsMonitorComp = false;
bool tsMonitorLogProtocol = false;
int32_t tsMonitorIntervalForBasic = 30;
bool tsMonitorForceV2 = false;
// audit // audit
bool tsEnableAudit = true; bool tsEnableAudit = true;
@ -690,6 +693,10 @@ 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, "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 (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, "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, "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 (cfgAddBool(pCfg, "auditCreateTable", tsEnableAuditCreateTable, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
@ -1152,6 +1159,9 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsMonitorMaxLogs = cfgGetItem(pCfg, "monitorMaxLogs")->i32; tsMonitorMaxLogs = cfgGetItem(pCfg, "monitorMaxLogs")->i32;
tsMonitorComp = cfgGetItem(pCfg, "monitorComp")->bval; tsMonitorComp = cfgGetItem(pCfg, "monitorComp")->bval;
tsQueryRspPolicy = cfgGetItem(pCfg, "queryRspPolicy")->i32; 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; tsEnableAudit = cfgGetItem(pCfg, "audit")->bval;
tsEnableAuditCreateTable = cfgGetItem(pCfg, "auditCreateTable")->bval; tsEnableAuditCreateTable = cfgGetItem(pCfg, "auditCreateTable")->bval;

1
source/dnode/mgmt/mgmt_dnode/inc/dmInt.h
View File

@ -43,6 +43,7 @@ typedef struct SDnodeMgmt {
GetMnodeLoadsFp getMnodeLoadsFp; GetMnodeLoadsFp getMnodeLoadsFp;
GetQnodeLoadsFp getQnodeLoadsFp; GetQnodeLoadsFp getQnodeLoadsFp;
int32_t statusSeq; int32_t statusSeq;
SendMonitorReportFp sendMonitorReportFpBasic;
} SDnodeMgmt; } SDnodeMgmt;
// dmHandle.c // dmHandle.c

1
source/dnode/mgmt/mgmt_dnode/src/dmInt.c
View File

@ -65,6 +65,7 @@ static int32_t dmOpenMgmt(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
pMgmt->processDropNodeFp = pInput->processDropNodeFp; pMgmt->processDropNodeFp = pInput->processDropNodeFp;
pMgmt->sendMonitorReportFp = pInput->sendMonitorReportFp; pMgmt->sendMonitorReportFp = pInput->sendMonitorReportFp;
pMgmt->sendAuditRecordsFp = pInput->sendAuditRecordFp; pMgmt->sendAuditRecordsFp = pInput->sendAuditRecordFp;
pMgmt->sendMonitorReportFpBasic = pInput->sendMonitorReportFpBasic;
pMgmt->getVnodeLoadsFp = pInput->getVnodeLoadsFp; pMgmt->getVnodeLoadsFp = pInput->getVnodeLoadsFp;
pMgmt->getVnodeLoadsLiteFp = pInput->getVnodeLoadsLiteFp; pMgmt->getVnodeLoadsLiteFp = pInput->getVnodeLoadsLiteFp;
pMgmt->getMnodeLoadsFp = pInput->getMnodeLoadsFp; pMgmt->getMnodeLoadsFp = pInput->getMnodeLoadsFp;

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

@ -81,6 +81,7 @@ static void *dmNotifyThreadFp(void *param) {
static void *dmMonitorThreadFp(void *param) { static void *dmMonitorThreadFp(void *param) {
SDnodeMgmt *pMgmt = param; SDnodeMgmt *pMgmt = param;
int64_t lastTime = taosGetTimestampMs(); int64_t lastTime = taosGetTimestampMs();
int64_t lastTimeForBasic = taosGetTimestampMs();
setThreadName("dnode-monitor"); setThreadName("dnode-monitor");
while (1) { while (1) {
@ -88,12 +89,22 @@ static void *dmMonitorThreadFp(void *param) {
if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; if (pMgmt->pData->dropped || pMgmt->pData->stopped) break;
int64_t curTime = taosGetTimestampMs(); int64_t curTime = taosGetTimestampMs();
if (curTime < lastTime) lastTime = curTime; if (curTime < lastTime) lastTime = curTime;
float interval = (curTime - lastTime) / 1000.0f; float interval = (curTime - lastTime) / 1000.0f;
if (interval >= tsMonitorInterval) { if (interval >= tsMonitorInterval) {
(*pMgmt->sendMonitorReportFp)(); (*pMgmt->sendMonitorReportFp)();
lastTime = curTime; lastTime = curTime;
} }
if(tsMonitorForceV2){
if (curTime < lastTimeForBasic) lastTimeForBasic = curTime;
float intervalForBasic = (curTime - lastTimeForBasic) / 1000.0f;
if (intervalForBasic >= tsMonitorIntervalForBasic) {
(*pMgmt->sendMonitorReportFpBasic)();
lastTimeForBasic = curTime;
}
}
} }
return NULL; return NULL;

1
source/dnode/mgmt/node_mgmt/inc/dmMgmt.h
View File

@ -149,6 +149,7 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
// dmMonitor.c // dmMonitor.c
void dmSendMonitorReport(); void dmSendMonitorReport();
void dmSendAuditRecords(); void dmSendAuditRecords();
void dmSendMonitorReportBasic();
void dmGetVnodeLoads(SMonVloadInfo *pInfo); void dmGetVnodeLoads(SMonVloadInfo *pInfo);
void dmGetVnodeLoadsLite(SMonVloadInfo *pInfo); void dmGetVnodeLoadsLite(SMonVloadInfo *pInfo);
void dmGetMnodeLoads(SMonMloadInfo *pInfo); void dmGetMnodeLoads(SMonMloadInfo *pInfo);

1
source/dnode/mgmt/node_mgmt/src/dmEnv.c
View File

@ -398,6 +398,7 @@ SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) {
.processDropNodeFp = dmProcessDropNodeReq, .processDropNodeFp = dmProcessDropNodeReq,
.sendMonitorReportFp = dmSendMonitorReport, .sendMonitorReportFp = dmSendMonitorReport,
.sendAuditRecordFp = auditSendRecordsInBatch, .sendAuditRecordFp = auditSendRecordsInBatch,
.sendMonitorReportFpBasic = dmSendMonitorReportBasic,
.getVnodeLoadsFp = dmGetVnodeLoads, .getVnodeLoadsFp = dmGetVnodeLoads,
.getVnodeLoadsLiteFp = dmGetVnodeLoadsLite, .getVnodeLoadsLiteFp = dmGetVnodeLoadsLite,
.getMnodeLoadsFp = dmGetMnodeLoads, .getMnodeLoadsFp = dmGetMnodeLoads,

18
source/dnode/mgmt/node_mgmt/src/dmMonitor.c
View File

@ -19,7 +19,8 @@
#include "audit.h" #include "audit.h"
static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) { static void dmGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->protocol = 1; //pInfo->protocol = 1;
pInfo->protocol = 2;
pInfo->dnode_id = pDnode->data.dnodeId; pInfo->dnode_id = pDnode->data.dnodeId;
pInfo->cluster_id = pDnode->data.clusterId; pInfo->cluster_id = pDnode->data.clusterId;
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN); tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
@ -106,9 +107,20 @@ void dmSendMonitorReport() {
dmGetVmMonitorInfo(pDnode); dmGetVmMonitorInfo(pDnode);
dmGetQmMonitorInfo(pDnode); dmGetQmMonitorInfo(pDnode);
dmGetSmMonitorInfo(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 //Todo: put this in seperate file in the future

1
source/dnode/mgmt/node_util/inc/dmUtil.h
View File

@ -122,6 +122,7 @@ typedef struct {
ProcessDropNodeFp processDropNodeFp; ProcessDropNodeFp processDropNodeFp;
SendMonitorReportFp sendMonitorReportFp; SendMonitorReportFp sendMonitorReportFp;
SendAuditRecordsFp sendAuditRecordFp; SendAuditRecordsFp sendAuditRecordFp;
SendMonitorReportFp sendMonitorReportFpBasic;
GetVnodeLoadsFp getVnodeLoadsFp; GetVnodeLoadsFp getVnodeLoadsFp;
GetVnodeLoadsFp getVnodeLoadsLiteFp; GetVnodeLoadsFp getVnodeLoadsLiteFp;
GetMnodeLoadsFp getMnodeLoadsFp; GetMnodeLoadsFp getMnodeLoadsFp;

124
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -14,6 +14,7 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tjson.h"
#include "mndDnode.h" #include "mndDnode.h"
#include "audit.h" #include "audit.h"
#include "mndCluster.h" #include "mndCluster.h"
@ -28,6 +29,7 @@
#include "mndVgroup.h" #include "mndVgroup.h"
#include "tmisce.h" #include "tmisce.h"
#include "tunit.h" #include "tunit.h"
#include "taos_monitor.h"
#define TSDB_DNODE_VER_NUMBER 2 #define TSDB_DNODE_VER_NUMBER 2
#define TSDB_DNODE_RESERVE_SIZE 64 #define TSDB_DNODE_RESERVE_SIZE 64
@ -503,9 +505,129 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) {
goto _OVER; 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: _OVER:
if(pJson != NULL){
tjsonDelete(pJson);
pJson = NULL;
}
tFreeSStatisReq(&statisReq); tFreeSStatisReq(&statisReq);
return code; return code;
} }

3
source/dnode/mnode/impl/src/mndMain.c
View File

@ -859,8 +859,10 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
pClusterInfo->master_uptime = (float)mndGetClusterUpTime(pMnode) / 86400.0f; pClusterInfo->master_uptime = (float)mndGetClusterUpTime(pMnode) / 86400.0f;
// pClusterInfo->master_uptime = (ms - pObj->stateStartTime) / (86400000.0f); // pClusterInfo->master_uptime = (ms - pObj->stateStartTime) / (86400000.0f);
tstrncpy(desc.role, syncStr(TAOS_SYNC_STATE_LEADER), sizeof(desc.role)); tstrncpy(desc.role, syncStr(TAOS_SYNC_STATE_LEADER), sizeof(desc.role));
desc.syncState = TAOS_SYNC_STATE_LEADER;
} else { } else {
tstrncpy(desc.role, syncStr(pObj->syncState), sizeof(desc.role)); tstrncpy(desc.role, syncStr(pObj->syncState), sizeof(desc.role));
desc.syncState = pObj->syncState;
} }
taosArrayPush(pClusterInfo->mnodes, &desc); taosArrayPush(pClusterInfo->mnodes, &desc);
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
@ -891,6 +893,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
SMonVnodeDesc *pVnDesc = &desc.vnodes[i]; SMonVnodeDesc *pVnDesc = &desc.vnodes[i];
pVnDesc->dnode_id = pVgid->dnodeId; pVnDesc->dnode_id = pVgid->dnodeId;
tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->syncState), sizeof(pVnDesc->vnode_role)); tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->syncState), sizeof(pVnDesc->vnode_role));
pVnDesc->syncState = pVgid->syncState;
if (pVgid->syncState == TAOS_SYNC_STATE_LEADER) { if (pVgid->syncState == TAOS_SYNC_STATE_LEADER) {
tstrncpy(desc.status, "ready", sizeof(desc.status)); tstrncpy(desc.status, "ready", sizeof(desc.status));
pClusterInfo->vgroups_alive++; pClusterInfo->vgroups_alive++;

24
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -47,6 +47,8 @@
#include "vnode.h" #include "vnode.h"
#include "taos_monitor.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
@ -104,6 +106,20 @@ typedef struct SQueryNode SQueryNode;
#define VND_INFO_FNAME "vnode.json" #define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME_TMP "vnode_tmp.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 // vnd.h
typedef int32_t (*_query_reseek_func_t)(void* pQHandle); typedef int32_t (*_query_reseek_func_t)(void* pQHandle);
struct SQueryNode { struct SQueryNode {
@ -438,6 +454,13 @@ typedef struct SVCommitSched {
int64_t maxWaitMs; int64_t maxWaitMs;
} SVCommitSched; } 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 { struct SVnode {
char* path; char* path;
SVnodeCfg config; SVnodeCfg config;
@ -476,6 +499,7 @@ struct SVnode {
int32_t blockSec; int32_t blockSec;
int64_t blockSeq; int64_t blockSeq;
SQHandle* pQuery; SQHandle* pQuery;
SVMonitorObj monitor;
}; };
#define TD_VID(PVNODE) ((PVNODE)->config.vgId) #define TD_VID(PVNODE) ((PVNODE)->config.vgId)

22
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -480,6 +480,28 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC
vnodeRollback(pVnode); 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; return pVnode;
_err: _err:

65
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -21,8 +21,6 @@
#include "vnd.h" #include "vnd.h"
#include "vnode.h" #include "vnode.h"
#include "vnodeInt.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 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); 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 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, static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc); 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 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 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 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 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 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 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 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 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 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); 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 vnodeProcessKillCompactReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
extern int32_t vnodeQueryCompactProgress(SVnode *pVnode, SRpcMsg *pMsg); extern int32_t vnodeQueryCompactProgress(SVnode *pVnode, SRpcMsg *pMsg);
@ -535,10 +534,10 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg
break; break;
/* TSDB */ /* TSDB */
case TDMT_VND_SUBMIT: 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; break;
case TDMT_VND_DELETE: 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; break;
case TDMT_VND_BATCH_DEL: case TDMT_VND_BATCH_DEL:
if (vnodeProcessBatchDeleteReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err; if (vnodeProcessBatchDeleteReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
@ -1482,7 +1481,8 @@ static int32_t vnodeRebuildSubmitReqMsg(SSubmitReq2 *pSubmitReq, void **ppMsg) {
return code; 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; int32_t code = 0;
terrno = 0; terrno = 0;
@ -1700,28 +1700,33 @@ _exit:
atomic_add_fetch_64(&pVnode->statis.nInsert, pSubmitRsp->affectedRows); atomic_add_fetch_64(&pVnode->statis.nInsert, pSubmitRsp->affectedRows);
atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, pSubmitRsp->affectedRows); atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, pSubmitRsp->affectedRows);
atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1); 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) { if (code == 0) {
atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1); atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1);
code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len); 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){ const char *batch_sample_labels[] = {VNODE_METRIC_TAG_VALUE_INSERT, pVnode->monitor.strClusterId,
int32_t label_count =2; pVnode->monitor.strDnodeId, tsLocalEp, pVnode->monitor.strVgId,
const char *sample_labels[] = {"vgid", "endpoint"}; pOriginalMsg->info.conn.user, "Success"};
taos_counter_t *counter = taos_counter_new("insert_counter", "counter for insert sql", label_count, sample_labels); taos_counter_inc(pVnode->monitor.insertCounter, batch_sample_labels);
if(taos_collector_registry_register_metric(counter) == 1){
taos_counter_destroy(counter);
} }
else{ 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, "Failed"};
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);
// clear // clear
taosArrayDestroy(newTbUids); taosArrayDestroy(newTbUids);
@ -1977,7 +1982,8 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pRe
return 0; 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; int32_t code = 0;
SDecoder *pCoder = &(SDecoder){0}; SDecoder *pCoder = &(SDecoder){0};
SDeleteRes *pRes = &(SDeleteRes){0}; SDeleteRes *pRes = &(SDeleteRes){0};
@ -2020,6 +2026,19 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, in
return code; return code;
_err: _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; return code;
} }
static int32_t vnodeProcessCreateIndexReq(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) {

15
source/libs/monitor/inc/monInt.h
View File

@ -20,6 +20,7 @@
#include "query.h" #include "query.h"
#include "tjson.h" #include "tjson.h"
#include "thash.h"
typedef struct { typedef struct {
int64_t curTime; int64_t curTime;
@ -45,8 +46,22 @@ typedef struct {
SMonSmInfo smInfo; SMonSmInfo smInfo;
SMonQmInfo qmInfo; SMonQmInfo qmInfo;
SMonBmInfo bmInfo; SMonBmInfo bmInfo;
SHashObj *metrics;
} SMonitor; } 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 #ifdef __cplusplus
} }
#endif #endif

7
source/libs/monitor/src/clientMonitor.c
View File

@ -10,16 +10,17 @@ tmr_h tmrStartHandle;
SHashObj* clusterMonitorInfoTable; SHashObj* clusterMonitorInfoTable;
static const int interval = 1000; // ms static const int interval = 1000; // ms
static const int sendBathchSize = 10; static const int sendBathchSize = 1;
int32_t sendReport(ClientMonitor* pMonitor, char* pCont); int32_t sendReport(ClientMonitor* pMonitor, char* pCont);
void generateClusterReport(ClientMonitor* pMonitor, bool send) { void generateClusterReport(ClientMonitor* pMonitor, bool send) {
char ts[50]; char ts[50];
sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); 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 (send && strlen(pCont) != TSDB_CODE_SUCCESS) {
if (sendReport(pMonitor, pCont) == 0) { if (sendReport(pMonitor, pCont) == 0) {
taos_collector_registry_clear_out(pMonitor->registry); taos_collector_registry_clear_batch(pMonitor->registry);
} }
} }
} }

656
source/libs/monitor/src/monFramework.c Normal file
View File

@ -0,0 +1,656 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#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);
}
}

106
source/libs/monitor/src/monMain.c
View File

@ -21,9 +21,9 @@
#include "taos_monitor.h" #include "taos_monitor.h"
#include "tglobal.h" #include "tglobal.h"
static SMonitor tsMonitor = {0}; SMonitor tsMonitor = {0};
static char* tsMonUri = "/report"; char* tsMonUri = "/report";
static char* tsMonFwUri = "/td_metric"; char* tsMonFwUri = "/td_metric";
void monRecordLog(int64_t ts, ELogLevel level, const char *content) { void monRecordLog(int64_t ts, ELogLevel level, const char *content) {
taosThreadMutexLock(&tsMonitor.lock); taosThreadMutexLock(&tsMonitor.lock);
@ -112,7 +112,7 @@ int32_t monInit(const SMonCfg *pCfg) {
tsMonitor.lastTime = taosGetTimestampMs(); tsMonitor.lastTime = taosGetTimestampMs();
taosThreadMutexInit(&tsMonitor.lock, NULL); taosThreadMutexInit(&tsMonitor.lock, NULL);
taos_collector_registry_default_init(); monInitMonitorFW();
return 0; return 0;
} }
@ -128,8 +128,7 @@ void monCleanup() {
tFreeSMonBmInfo(&tsMonitor.bmInfo); tFreeSMonBmInfo(&tsMonitor.bmInfo);
taosThreadMutexDestroy(&tsMonitor.lock); taosThreadMutexDestroy(&tsMonitor.lock);
taos_collector_registry_destroy(TAOS_COLLECTOR_REGISTRY_DEFAULT); monCleanupMonitorFW();
TAOS_COLLECTOR_REGISTRY_DEFAULT = NULL;
} }
static void monCleanupMonitorInfo(SMonInfo *pMonitor) { static void monCleanupMonitorInfo(SMonInfo *pMonitor) {
@ -195,6 +194,22 @@ static void monGenBasicJson(SMonInfo *pMonitor) {
tjsonAddDoubleToObject(pJson, "protocol", pInfo->protocol); 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) { static void monGenClusterJson(SMonInfo *pMonitor) {
SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster; SMonClusterInfo *pInfo = &pMonitor->mmInfo.cluster;
if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; 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) { static void monGenVgroupJson(SMonInfo *pMonitor) {
SMonVgroupInfo *pInfo = &pMonitor->mmInfo.vgroup; SMonVgroupInfo *pInfo = &pMonitor->mmInfo.vgroup;
if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return; if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 0) return;
@ -526,10 +551,25 @@ static void monGenLogJson(SMonInfo *pMonitor) {
if (tjsonAddItemToArray(pSummaryJson, pLogTrace) != 0) tjsonDelete(pLogTrace); if (tjsonAddItemToArray(pSummaryJson, pLogTrace) != 0) tjsonDelete(pLogTrace);
} }
void monSendReport() { void monSendReport(SMonInfo *pMonitor){
char *pCont = tjsonToString(pMonitor->pJson);
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) {
uError("failed to send monitor msg");
}
taosMemoryFree(pCont);
}
}
void monGenAndSendReport() {
SMonInfo *pMonitor = monCreateMonitorInfo(); SMonInfo *pMonitor = monCreateMonitorInfo();
if (pMonitor == NULL) return; if (pMonitor == NULL) return;
if(!tsMonitorForceV2){
monGenBasicJson(pMonitor); monGenBasicJson(pMonitor);
monGenClusterJson(pMonitor); monGenClusterJson(pMonitor);
monGenVgroupJson(pMonitor); monGenVgroupJson(pMonitor);
@ -539,41 +579,33 @@ void monSendReport() {
monGenDiskJson(pMonitor); monGenDiskJson(pMonitor);
monGenLogJson(pMonitor); monGenLogJson(pMonitor);
char *pCont = tjsonToString(pMonitor->pJson); monSendReport(pMonitor);
// uDebugL("report cont:%s\n", 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) {
uError("failed to send monitor msg");
} }
taosMemoryFree(pCont); else{
monGenClusterInfoTable(pMonitor);
monGenVgroupInfoTable(pMonitor);
monGenDnodeInfoTable(pMonitor);
monGenDnodeStatusInfoTable(pMonitor);
monGenDataDiskTable(pMonitor);
monGenLogDiskTable(pMonitor);
monGenMnodeRoleTable(pMonitor);
monGenVnodeRoleTable(pMonitor);
monSendPromReport();
} }
monCleanupMonitorInfo(pMonitor); monCleanupMonitorInfo(pMonitor);
} }
void monSendPromReport() { void monGenAndSendReportBasic() {
char ts[50]; SMonInfo *pMonitor = monCreateMonitorInfo();
sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI)); if (pMonitor == NULL) return;
char *pCont = (char *)taos_collector_registry_bridge(TAOS_COLLECTOR_REGISTRY_DEFAULT, ts, "%" PRId64); if (pMonitor->mmInfo.cluster.first_ep_dnode_id == 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");
}else{
taos_collector_registry_clear_out(TAOS_COLLECTOR_REGISTRY_DEFAULT);
}
}
}
void monSendContent(char *pCont) { monGenBasicJsonBasic(pMonitor);
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return; monGenClusterJsonBasic(pMonitor);
//uInfoL("report cont:\n%s\n", pCont);
if (pCont != NULL) { monSendReport(pMonitor);
EHttpCompFlag flag = tsMonitor.cfg.comp ? HTTP_GZIP : HTTP_FLAT;
if (taosSendHttpReport(tsMonitor.cfg.server, tsMonFwUri, tsMonitor.cfg.port, pCont, strlen(pCont), flag) != 0) { monCleanupMonitorInfo(pMonitor);
uError("failed to send monitor msg");
}
}
} }

4
source/libs/monitor/test/monTest.cpp
View File

@ -283,10 +283,10 @@ TEST_F(MonitorTest, 01_Full) {
tFreeSMonSmInfo(&smInfo); tFreeSMonSmInfo(&smInfo);
tFreeSMonQmInfo(&qmInfo); tFreeSMonQmInfo(&qmInfo);
tFreeSMonBmInfo(&bmInfo); tFreeSMonBmInfo(&bmInfo);
monSendReport(); monGenAndSendReport();
} }
TEST_F(MonitorTest, 02_Log) { TEST_F(MonitorTest, 02_Log) {
AddLogInfo2(); AddLogInfo2();
monSendReport(); monGenAndSendReport();
} }

2
source/libs/monitorfw/inc/taos_collector_registry_t.h
View File

@ -34,7 +34,7 @@ struct taos_collector_registry {
taos_string_builder_t *string_builder; /**< Enables string building */ taos_string_builder_t *string_builder; /**< Enables string building */
taos_metric_formatter_t *metric_formatter; /**< metric formatter for metric exposition on bridge call */ taos_metric_formatter_t *metric_formatter; /**< metric formatter for metric exposition on bridge call */
pthread_rwlock_t *lock; /**< mutex for safety against concurrent registration */ 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 #endif // TAOS_REGISTRY_T_H

31
source/libs/monitorfw/inc/taos_metric_formatter_custom_i.h Normal file
View File

@ -0,0 +1,31 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef TAOS_METRIC_FORMATTER_CUSTOMV2_I_H
#define TAOS_METRIC_FORMATTER_CUSTOMV2_I_H
#define ALLOW_FORBID_FUNC
#include <stdint.h>
#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

8
source/libs/monitorfw/inc/taos_metric_sample_t.h
View File

@ -19,10 +19,16 @@
#include "taos_metric_sample.h" #include "taos_metric_sample.h"
#include "taos_metric_t.h" #include "taos_metric_t.h"
#define DOUBLE_ATOMIC
struct taos_metric_sample { struct taos_metric_sample {
taos_metric_type_t type; /**< type is the metric type for the 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 */ 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 #endif // TAOS_METRIC_SAMPLE_T_H

2
source/libs/monitorfw/inc/taos_metric_t.h
View File

@ -42,7 +42,7 @@ extern char *taos_metric_type_map[4];
*/ */
struct taos_metric { struct taos_metric {
taos_metric_type_t type; /**< metric_type The type of 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 */ const char *help; /**< help The help output for the metric */
taos_map_t *samples; /**< samples Map comprised of samples for the given 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*/ size_t label_key_count; /**< label_keys_count The count of labe_keys*/

26
source/libs/monitorfw/inc/taos_monitor_util_i.h Normal file
View File

@ -0,0 +1,26 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef TAOS_MONITOR_UTIL_I_H
#define TAOS_MONITOR_UTIL_I_H
#include <stdint.h>
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

6
source/libs/monitorfw/src/taos_collector.c
View File

@ -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); 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);
}

89
source/libs/monitorfw/src/taos_collector_registry.c
View File

@ -33,6 +33,10 @@
#include "taos_metric_i.h" #include "taos_metric_i.h"
#include "taos_metric_t.h" #include "taos_metric_t.h"
#include "taos_string_builder_i.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; 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->metric_formatter = taos_metric_formatter_new();
self->string_builder = taos_string_builder_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)); self->lock = (pthread_rwlock_t *)taos_malloc(sizeof(pthread_rwlock_t));
r = pthread_rwlock_init(self->lock, NULL); r = pthread_rwlock_init(self->lock, NULL);
if (r) { if (r) {
@ -88,8 +92,8 @@ int taos_collector_registry_destroy(taos_collector_registry_t *self) {
self->string_builder = NULL; self->string_builder = NULL;
if (r) ret = r; if (r) ret = r;
r = taos_string_builder_destroy(self->out); r = taos_string_builder_destroy(self->string_builder_batch);
self->out = NULL; self->string_builder_batch = NULL;
if (r) ret = r; if (r) ret = r;
r = pthread_rwlock_destroy(self->lock); 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); 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) { taos_metric_t *taos_collector_registry_must_register_metric(taos_metric_t *metric) {
int err = taos_collector_registry_register_metric(metric); int err = taos_collector_registry_register_metric(metric);
if (err != 0) { 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); char *out = taos_metric_formatter_dump(self->metric_formatter);
int r = 0; 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; if (r) return NULL;
taos_free(out); 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){ int taos_collector_registry_clear_batch(taos_collector_registry_t *self){
return taos_string_builder_clear(self->out); 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;
} }

100
source/libs/monitorfw/src/taos_gauge.c Normal file
View File

@ -0,0 +1,100 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
// 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);
}

9
source/libs/monitorfw/src/taos_metric.c
View File

@ -34,7 +34,11 @@ taos_metric_t *taos_metric_new(taos_metric_type_t metric_type, const char *name,
int r = 0; int r = 0;
taos_metric_t *self = (taos_metric_t *)taos_malloc(sizeof(taos_metric_t)); taos_metric_t *self = (taos_metric_t *)taos_malloc(sizeof(taos_metric_t));
self->type = metric_type; 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; self->help = help;
const char **k = (const char **)taos_malloc(sizeof(const char *) * label_key_count); 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); taos_free(self->label_keys);
self->label_keys = NULL; self->label_keys = NULL;
taos_free(self->name);
self->name = NULL;
taos_free(self); taos_free(self);
self = NULL; self = NULL;

2
source/libs/monitorfw/src/taos_metric_formatter.c
View File

@ -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); r = taos_string_builder_add_str(self->string_builder, ts);
if (r) return r; 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'); return taos_string_builder_add_char(self->string_builder, '\n');
} }

234
source/libs/monitorfw/src/taos_metric_formatter_custom.c Normal file
View File

@ -0,0 +1,234 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#include <stdio.h>
#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;
}

53
source/libs/monitorfw/src/taos_metric_sample.c
View File

@ -13,7 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
//#include <stdatomic.h>
// Public // Public
#include "taos_alloc.h" #include "taos_alloc.h"
@ -25,10 +25,14 @@
#include "taos_metric_sample_i.h" #include "taos_metric_sample_i.h"
#include "taos_metric_sample_t.h" #include "taos_metric_sample_t.h"
#define ALLOW_FORBID_FUNC #define DOUBLE_ATOMIC
#include "tdef.h" #ifdef DOUBLE_ATOMIC
#include <stdatomic.h>
#else
#define ALLOW_FORBID_FUNC
#include "osAtomic.h" #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 *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)); 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) { if (r_value < 0) {
return 1; return 1;
} }
/*
#ifdef DOUBLE_ATOMIC
_Atomic double old = atomic_load(&self->r_value); _Atomic double old = atomic_load(&self->r_value);
for (;;) { for (;;) {
@ -76,8 +81,10 @@ int taos_metric_sample_add(taos_metric_sample_t *self, double r_value) {
return 0; return 0;
} }
} }
*/ #else
atomic_fetch_add_64(&self->r_value, r_value); atomic_fetch_add_64(&self->r_value, r_value);
#endif
return 0; 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); TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE);
return 1; return 1;
} }
/*
#ifdef DOUBLE_ATOMIC
_Atomic double old = atomic_load(&self->r_value); _Atomic double old = atomic_load(&self->r_value);
for (;;) { for (;;) {
_Atomic double new = ATOMIC_VAR_INIT(old - r_value); _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; return 0;
} }
} }
*/ #else
atomic_fetch_sub_64(&self->r_value, r_value); atomic_fetch_sub_64(&self->r_value, r_value);
#endif
return 0; 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); TAOS_LOG(TAOS_METRIC_INCORRECT_TYPE);
return 1; return 1;
} }
/*
#ifdef DOUBLE_ATOMIC
atomic_store(&self->r_value, r_value); atomic_store(&self->r_value, r_value);
*/ #else
atomic_store_64(&self->r_value, r_value); 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; return 0;
} }

110
source/libs/monitorfw/src/taos_monitor_util.c Normal file
View File

@ -0,0 +1,110 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#define ALLOW_FORBID_FUNC
#include "tjson.h"
#include <stdio.h>
#include <string.h>
#include <stdint.h>
#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;
}