GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 04:19:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #846 from eeff/eeff/refactor

Browse Source 
refactor: node metrics data structures
This commit is contained in:
fengzeroz
2022-10-20 17:50:02 +08:00
committed by GitHub
parent 9665ad1aeb 27d7b2d4d5
commit f893947e74
10 changed files with 314 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+6 -57
View File
@@ -27,65 +27,10 @@ extern "C" {
#include <stdint.h>
#include "define.h"
#include "metrics.h"
#include "tag.h"
#include "type.h"
#define NEU_NODE_STAT_RTT_MAX 9999
/** Kinds of node statistics counters.
*
* Some counters are maintained by the neuron core, while others should be
* maintained by the plugin code using the adapter callback `stat_acc`.
*/
typedef enum {
// kinds of counters maintained by the plugins
NEU_NODE_STAT_BYTES_SENT, // number of bytes sent
NEU_NODE_STAT_BYTES_RECV, // number of bytes received
NEU_NODE_STAT_MSGS_SENT, // number of messages sent
NEU_NODE_STAT_MSGS_RECV, // number of messages received
NEU_NODE_STAT_AVG_RTT, // average round trip time (ms)
// kinds of counters maintained by the driver core
NEU_NODE_STAT_TAG_TOT_CNT, // number of tag read including errors
NEU_NODE_STAT_TAG_ERR_CNT, // number of tag read errors
NEU_NODE_STAT_MAX,
} neu_node_stat_e;
typedef enum {
NEU_METRICS_CATEGORY_GLOBAL,
NEU_METRICS_CATEGORY_DRIVER,
NEU_METRICS_CATEGORY_APP,
NEU_METRICS_CATEGORY_ALL,
} neu_metrics_category_e;
typedef struct {
bool core_dumped;
uint64_t uptime_seconds;
size_t north_nodes;
size_t north_running_nodes;
size_t north_disconnected_nodes;
size_t south_nodes;
size_t south_running_nodes;
size_t south_disconnected_nodes;
} neu_metrics_t;
static inline const char *neu_node_stat_string(neu_node_stat_e s)
{
static const char *map[] = {
[NEU_NODE_STAT_BYTES_SENT] = "bytes_sent",
[NEU_NODE_STAT_BYTES_RECV] = "bytes_received",
[NEU_NODE_STAT_MSGS_SENT] = "messages_sent",
[NEU_NODE_STAT_MSGS_RECV] = "messages_received",
[NEU_NODE_STAT_AVG_RTT] = "average_rtt",
[NEU_NODE_STAT_TAG_TOT_CNT] = "tag_total_count",
[NEU_NODE_STAT_TAG_ERR_CNT] = "tag_error_count",
[NEU_NODE_STAT_MAX] = NULL,
};
return map[s];
}
typedef struct {
neu_node_running_state_e running;
neu_node_link_state_e link;
@@ -494,11 +439,15 @@ typedef struct neu_adapter neu_adapter_t;
typedef struct neu_adapter_driver neu_adapter_driver_t;
typedef struct neu_adapter_app neu_adapter_app_t;
typedef int (*neu_adapter_update_metric_cb_t)(neu_adapter_t *adapter,
const char * metric_name,
uint64_t n);
typedef struct adapter_callbacks {
int (*command)(neu_adapter_t *adapter, neu_reqresp_head_t head, void *data);
int (*response)(neu_adapter_t *adapter, neu_reqresp_head_t *head,
void *data);
void (*stat_acc)(neu_adapter_t *adapter, neu_node_stat_e s, uint64_t n);
neu_adapter_update_metric_cb_t update_metric;
union {
struct {
include/neuron/metrics.h
+157
View File
@@ -0,0 +1,157 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* 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. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef NEU_METRICS_H
#define NEU_METRICS_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
#include "define.h"
#include "type.h"
#include "utils/uthash.h"
typedef enum {
NEU_METRIC_TYPE_COUNTER,
NEU_METRIC_TYPE_GAUAGE,
} neu_metric_type_e;
// last round trip time in millisends
#define NEU_METRIC_LAST_RTT_MS "last_rtt_ms"
#define NEU_METRIC_LAST_RTT_MS_TYPE NEU_METRIC_TYPE_GAUAGE
#define NEU_METRIC_LAST_RTT_MS_HELP \
"Last request round trip time in milliseconds"
// number of bytes sent
#define NEU_METRIC_SEND_BYTES "send_bytes"
#define NEU_METRIC_SEND_BYTES_TYPE NEU_METRIC_TYPE_COUNTER
#define NEU_METRIC_SEND_BYTES_HELP "Total number of bytes sent"
// number of bytes received
#define NEU_METRIC_RECV_BYTES "recv_bytes"
#define NEU_METRIC_RECV_BYTES_TYPE NEU_METRIC_TYPE_COUNTER
#define NEU_METRIC_RECV_BYTES_HELP "Total number of bytes received"
// number of tag read including errors
#define NEU_METRIC_TAG_READS_TOTAL "tag_reads_total"
#define NEU_METRIC_TAG_READS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER
#define NEU_METRIC_TAG_READS_TOTAL_HELP \
"Total number of tag reads including errors"
// number of tag read errors
#define NEU_METRIC_TAG_READ_ERRORS_TOTAL "tag_read_errors_total"
#define NEU_METRIC_TAG_READ_ERRORS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER
#define NEU_METRIC_TAG_READ_ERRORS_TOTAL_HELP "Total number of tag read errors"
typedef enum {
NEU_METRICS_CATEGORY_GLOBAL,
NEU_METRICS_CATEGORY_DRIVER,
NEU_METRICS_CATEGORY_APP,
NEU_METRICS_CATEGORY_ALL,
} neu_metrics_category_e;
// metric entry
typedef struct {
const char * name; // NOTE: should points to string literal
const char * help; // NOTE: should points to string literal
neu_metric_type_e type; //
uint64_t value; //
UT_hash_handle hh; // ordered by name
} neu_metric_entry_t;
// group metrics
typedef struct {
char * name; // group name
neu_metric_entry_t *entries; // group metric entries
UT_hash_handle hh; // ordered by name
} neu_group_metrics_t;
// node metrics
typedef struct {
neu_node_type_e type; // node type
char * name; // node name
neu_metric_entry_t * entries; // node metric entries
neu_group_metrics_t *group_metrics; // group metrics
UT_hash_handle hh; // ordered by name
} neu_node_metrics_t;
typedef struct {
bool core_dumped;
uint64_t uptime_seconds;
size_t north_nodes;
size_t north_running_nodes;
size_t north_disconnected_nodes;
size_t south_nodes;
size_t south_running_nodes;
size_t south_disconnected_nodes;
neu_node_metrics_t *node_metrics;
} neu_metrics_t;
static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
{
free(entry);
}
static inline void neu_group_metrics_free(neu_group_metrics_t *group_metrics)
{
if (NULL == group_metrics) {
return;
}
neu_metric_entry_t *e = NULL, *tmp = NULL;
HASH_ITER(hh, group_metrics->entries, e, tmp)
{
HASH_DEL(group_metrics->entries, e);
neu_metric_entry_free(e);
}
free(group_metrics->name);
free(group_metrics);
}
static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
{
if (NULL == node_metrics) {
return;
}
neu_metric_entry_t *e = NULL, *etmp = NULL;
HASH_ITER(hh, node_metrics->entries, e, etmp)
{
HASH_DEL(node_metrics->entries, e);
neu_metric_entry_free(e);
}
neu_group_metrics_t *g = NULL, *gtmp = NULL;
HASH_ITER(hh, node_metrics->group_metrics, g, gtmp)
{
HASH_DEL(node_metrics->group_metrics, g);
neu_group_metrics_free(g);
}
free(node_metrics);
}
#ifdef __cplusplus
}
#endif
#endif
plugins/ekuiper/plugin_ekuiper.c
-4
View File
@@ -54,8 +54,6 @@ static void pipe_add_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
neu_plugin_t *plugin = arg;
nng_mtx_lock(plugin->mtx);
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
plugin->common.adapter_callbacks->stat_acc(plugin->common.adapter,
NEU_NODE_STAT_AVG_RTT, 0);
nng_mtx_unlock(plugin->mtx);
}
@@ -66,8 +64,6 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
neu_plugin_t *plugin = arg;
nng_mtx_lock(plugin->mtx);
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
plugin->common.adapter_callbacks->stat_acc(
plugin->common.adapter, NEU_NODE_STAT_AVG_RTT, NEU_NODE_STAT_RTT_MAX);
nng_mtx_unlock(plugin->mtx);
}
plugins/ekuiper/read_write.c
+8 -8
View File
@@ -29,8 +29,9 @@
void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
{
int rv = 0;
void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) =
plugin->common.adapter_callbacks->stat_acc;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
char * json_str = NULL;
json_read_resp_t resp = {
@@ -58,8 +59,7 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
rv = nng_sendmsg(plugin->sock, msg,
NNG_FLAG_NONBLOCK); // TODO: use aio to send message
if (0 == rv) {
stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_SENT, 1);
stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_SENT, json_len);
update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, json_len);
} else {
plog_error(plugin, "nng cannot send msg");
nng_msg_free(msg);
@@ -74,8 +74,9 @@ void recv_data_callback(void *arg)
size_t json_len = 0;
char * json_str = NULL;
json_write_req_t *req = NULL;
void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) =
plugin->common.adapter_callbacks->stat_acc;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
rv = nng_aio_result(plugin->recv_aio);
if (0 != rv) {
@@ -87,8 +88,7 @@ void recv_data_callback(void *arg)
json_str = nng_msg_body(msg);
json_len = nng_msg_len(msg);
plog_debug(plugin, "<< %.*s", (int) json_len, json_str);
stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_RECV, 1);
stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_RECV, json_len);
update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, json_len);
if (json_decode_write_req(json_str, json_len, &req) < 0) {
plog_error(plugin, "fail decode write request json: %.*s",
(int) nng_msg_len(msg), json_str);
plugins/modbus/modbus_req.c
+8 -10
View File
@@ -49,14 +49,14 @@ void modbus_conn_disconnected(void *data, int fd)
int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes)
{
neu_plugin_t *plugin = (neu_plugin_t *) ctx;
void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) =
plugin->common.adapter_callbacks->stat_acc;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
plog_send_protocol(plugin, bytes, n_byte);
int ret = neu_conn_send(plugin->conn, bytes, n_byte);
if (ret > 0) {
stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_SENT, 1);
stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_SENT, ret);
update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, ret);
}
return ret;
}
@@ -303,19 +303,17 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
uint8_t * recv_buf = calloc(response_size, 1);
neu_protocol_unpack_buf_t pbuf = { 0 };
ssize_t ret = 0;
void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) =
plugin->common.adapter_callbacks->stat_acc;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
if (ret > 0) {
stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_RECV, ret);
update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, ret);
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
plog_recv_protocol(plugin, recv_buf, ret);
ret = modbus_stack_recv(plugin->stack, &pbuf);
if (ret > 0) {
stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_RECV, 1);
}
}
free(recv_buf);
plugins/monitor/metric_handle.c
+27 -25
View File
@@ -25,31 +25,33 @@
#include "restful/handle.h"
#include "restful/http.h"
#define METRIC_GLOBAL_TMPL \
"# HELP core_dumped Whether there is any core dump\n" \
"# TYPE core_dumped counter\n" \
"core_dumped %d\n" \
"# HELP uptime_seconds Uptime in seconds\n" \
"# TYPE uptime_seconds counter\n" \
"uptime_seconds %" PRIu64 "\n" \
"# HELP north_nodes Number of north nodes\n" \
"# TYPE north_nodes counter\n" \
"north_nodes %zu\n" \
"# HELP north_running_nodes Number of north nodes in running state\n" \
"# TYPE north_running_nodes counter\n" \
"north_running_nodes %zu\n" \
"# HELP north_disconnected_nodes Number of north nodes disconnected\n" \
"# TYPE north_disconnected_nodes counter\n" \
"north_disconnected_nodes %zu\n" \
"# HELP south_nodes Number of south nodes\n" \
"# TYPE south_nodes counter\n" \
"south_nodes %zu\n" \
"# HELP south_running_nodes Number of south nodes in running state\n" \
"# TYPE south_running_nodes counter\n" \
"south_running_nodes %zu\n" \
"# HELP south_disconnected_nodes Number of south nodes disconnected\n" \
"# TYPE south_disconnected_nodes counter\n" \
"south_disconnected_nodes %zu\n"
// clang-format off
#define METRIC_GLOBAL_TMPL \
"# HELP core_dumped Whether there is any core dump\n" \
"# TYPE core_dumped gauge\n" \
"core_dumped %d\n" \
"# HELP uptime_seconds Uptime in seconds\n" \
"# TYPE uptime_seconds counter\n" \
"uptime_seconds %" PRIu64 "\n" \
"# HELP north_nodes_total Number of north nodes\n" \
"# TYPE north_nodes_total gauge\n" \
"north_nodes_total %zu\n" \
"# HELP north_running_nodes_total Number of north nodes in running state\n" \
"# TYPE north_running_nodes_total gauge\n" \
"north_running_nodes_total %zu\n" \
"# HELP north_disconnected_nodes_total Number of north nodes disconnected\n" \
"# TYPE north_disconnected_nodes_total gauge\n" \
"north_disconnected_nodes_total %zu\n" \
"# HELP south_nodes_total Number of south nodes\n" \
"# TYPE south_nodes_total gauge\n" \
"south_nodes_total %zu\n" \
"# HELP south_running_nodes_total Number of south nodes in running state\n" \
"# TYPE south_running_nodes_total gauge\n" \
"south_running_nodes_total %zu\n" \
"# HELP south_disconnected_nodes_total Number of south nodes disconnected\n" \
"# TYPE south_disconnected_nodes_total gauge\n" \
"south_disconnected_nodes_total %zu\n"
// clang-format on
static int response(nng_aio *aio, char *content, enum nng_http_status status)
{
src/adapter/adapter.c
+89 -50
View File
@@ -20,6 +20,7 @@
#include <assert.h>
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -40,23 +41,39 @@
#include "plugin.h"
#include "storage.h"
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data);
static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static void adapter_stat_acc(neu_adapter_t *adapter, neu_node_stat_e s,
uint64_t n);
static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
neu_metric_type_e type, const char *help,
uint64_t init);
static int adapter_update_metric(neu_adapter_t *adapter,
const char *metric_name, uint64_t n);
inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int update_timestamp(void *usr_data);
static pthread_rwlock_t g_node_metrics_mtx_ = PTHREAD_RWLOCK_INITIALIZER;
static neu_metrics_t g_node_metrics_;
static const adapter_callbacks_t callback_funs = {
.command = adapter_command,
.response = adapter_response,
.stat_acc = adapter_stat_acc,
.command = adapter_command,
.response = adapter_response,
.update_metric = adapter_update_metric,
};
#define REGISTER_METRIC(adapter, name, init) \
adapter_register_metric(adapter, name, name##_TYPE, name##_HELP, init);
#define REGISTER_DRIVER_METRICS(adapter) \
REGISTER_METRIC(adapter, NEU_METRIC_LAST_RTT_MS, 9999); \
REGISTER_METRIC(adapter, NEU_METRIC_SEND_BYTES, 0); \
REGISTER_METRIC(adapter, NEU_METRIC_RECV_BYTES, 0); \
REGISTER_METRIC(adapter, NEU_METRIC_TAG_READS_TOTAL, 0); \
REGISTER_METRIC(adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, 0)
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
{
int rv = 0;
@@ -77,16 +94,21 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
break;
}
adapter->name = strdup(info->name);
adapter->events = neu_event_new();
adapter->state = NEU_NODE_RUNNING_STATE_INIT;
adapter->handle = info->handle;
adapter->cb_funs.command = callback_funs.command;
adapter->cb_funs.response = callback_funs.response;
adapter->cb_funs.stat_acc = callback_funs.stat_acc;
adapter->module = info->module;
adapter->name = strdup(info->name);
adapter->events = neu_event_new();
adapter->state = NEU_NODE_RUNNING_STATE_INIT;
adapter->handle = info->handle;
adapter->cb_funs.command = callback_funs.command;
adapter->cb_funs.response = callback_funs.response;
adapter->cb_funs.update_metric = callback_funs.update_metric;
adapter->module = info->module;
adapter->stat.avg_rtt = NEU_NODE_STAT_RTT_MAX;
adapter->metrics = calloc(1, sizeof(*adapter->metrics));
if (NULL != adapter->metrics) {
adapter->metrics->type = info->module->type;
adapter->metrics->name = adapter->name;
HASH_ADD_STR(g_node_metrics_.node_metrics, name, adapter->metrics);
}
rv = nng_pair1_open(&adapter->sock);
assert(rv == 0);
@@ -97,6 +119,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
switch (info->module->type) {
case NEU_NA_TYPE_DRIVER:
REGISTER_DRIVER_METRICS(adapter);
neu_adapter_driver_init((neu_adapter_driver_t *) adapter);
break;
case NEU_NA_TYPE_APP:
@@ -166,41 +189,46 @@ neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter)
return adapter->module->type;
}
static void adapter_stat_acc(neu_adapter_t *adapter, neu_node_stat_e s,
uint64_t n)
static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
neu_metric_type_e type, const char *help,
uint64_t init)
{
switch (s) {
case NEU_NODE_STAT_BYTES_SENT:
adapter->stat.bytes_sent += n;
break;
case NEU_NODE_STAT_BYTES_RECV:
adapter->stat.bytes_recv += n;
break;
case NEU_NODE_STAT_MSGS_SENT:
adapter->stat.msgs_sent += n;
break;
case NEU_NODE_STAT_MSGS_RECV:
adapter->stat.msgs_recv += n;
break;
case NEU_NODE_STAT_AVG_RTT: {
if (NEU_NODE_STAT_RTT_MAX <= n ||
adapter->stat.avg_rtt == NEU_NODE_STAT_RTT_MAX) {
adapter->stat.avg_rtt = n;
break;
}
// exponential moving average with alpha = 0.3
const double a = 0.3;
adapter->stat.avg_rtt = adapter->stat.avg_rtt * (1 - a) + n * a;
break;
if (NULL == adapter->metrics) {
return -1;
}
// these are maintained by neuron core
case NEU_NODE_STAT_TAG_TOT_CNT:
case NEU_NODE_STAT_TAG_ERR_CNT:
default:
assert(!"please supply a valid statistics counter kind");
neu_metric_entry_t *entry = calloc(1, sizeof(*entry));
if (NULL == entry) {
return -1;
}
entry->name = name;
entry->type = type;
entry->help = help;
entry->value = init;
HASH_ADD_STR(adapter->metrics->entries, name, entry);
return 0;
}
static int adapter_update_metric(neu_adapter_t *adapter,
const char *metric_name, uint64_t n)
{
neu_metric_entry_t *entry = NULL;
if (NULL != adapter->metrics) {
HASH_FIND_STR(adapter->metrics->entries, metric_name, entry);
}
if (NULL == entry) {
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else {
entry->value = n;
}
return 0;
}
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
@@ -402,8 +430,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_NODE_STATE: {
neu_resp_get_node_state_t resp = { 0 };
neu_metric_entry_t *e = NULL;
if (NULL != adapter->metrics) {
HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e);
}
resp.avg_rtt = NULL != e ? e->value : 0;
resp.state = neu_adapter_get_state(adapter);
resp.avg_rtt = adapter->stat.avg_rtt;
header->type = NEU_RESP_GET_NODE_STATE;
neu_msg_exchange(header);
reply(adapter, header, &resp);
@@ -680,6 +712,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
adapter->module->intf_funs->close(adapter->plugin);
if (NULL != adapter->metrics) {
pthread_rwlock_wrlock(&g_node_metrics_mtx_);
HASH_DEL(g_node_metrics_.node_metrics, adapter->metrics);
neu_node_metrics_free(adapter->metrics);
pthread_rwlock_unlock(&g_node_metrics_mtx_);
}
if (adapter->name != NULL) {
free(adapter->name);
}
src/adapter/adapter_internal.h
+2 -13
View File
@@ -52,19 +52,8 @@ struct neu_adapter {
neu_event_timer_t *timer;
int recv_fd;
// statistics counters
union {
struct {
uint64_t bytes_sent; // number of bytes sent over network
uint64_t bytes_recv; // number of bytes received over network
uint64_t msgs_sent; // number of messages sent
uint64_t msgs_recv; // number of messages received
uint64_t avg_rtt; // average round trip time in milliseconds
uint64_t tag_tot_cnt; // number of tag read including errors
uint64_t tag_err_cnt; // number of tag read errors
};
uint64_t data[NEU_NODE_STAT_MAX];
} stat;
// metrics
neu_node_metrics_t *metrics;
};
typedef void (*adapter_handler)(neu_adapter_t * adapter,
src/adapter/driver/driver.c
+11 -4
View File
@@ -99,15 +99,22 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_driver_cache_update(driver->cache, group, t->name,
driver->adapter.timestamp, value);
}
driver->adapter.stat.tag_tot_cnt += utarray_len(tags);
driver->adapter.stat.tag_err_cnt += utarray_len(tags);
driver->adapter.cb_funs.update_metric(&driver->adapter,
NEU_METRIC_TAG_READS_TOTAL,
utarray_len(tags));
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
utarray_len(tags));
utarray_free(tags);
}
} else {
neu_driver_cache_update(driver->cache, group, tag,
driver->adapter.timestamp, value);
driver->adapter.stat.tag_tot_cnt++;
driver->adapter.stat.tag_err_cnt += (NEU_TYPE_ERROR == value.type);
driver->adapter.cb_funs.update_metric(&driver->adapter,
NEU_METRIC_TAG_READS_TOTAL, 1);
driver->adapter.cb_funs.update_metric(&driver->adapter,
NEU_METRIC_TAG_READ_ERRORS_TOTAL,
NEU_TYPE_ERROR == value.type);
}
nlog_debug(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
src/core/node_manager.c
+6 -1
View File
@@ -303,7 +303,12 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr)
state.state.running = el->adapter->state;
state.state.link =
neu_plugin_to_plugin_common(el->adapter->plugin)->link_state;
state.avg_rtt = el->adapter->stat.avg_rtt;
neu_metric_entry_t *e = NULL;
if (NULL != el->adapter->metrics) {
HASH_FIND_STR(el->adapter->metrics->entries,
NEU_METRIC_LAST_RTT_MS, e);
}
state.avg_rtt = NULL != e ? e->value : 0;
utarray_push_back(states, &state);
}