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

Merge pull request #774 from eeff/eeff/dev

Browse Source 
Add node statistics counters
This commit is contained in:
fengzeroz
2022-09-13 09:38:27 +08:00
committed by GitHub
parent 18af6dfd3c 00a934453e
commit a1cbbf3a84
13 changed files with 237 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+48
View File
@@ -30,6 +30,40 @@ extern "C" {
#include "tag.h"
#include "type.h"
/** 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
// 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;
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_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;
@@ -65,6 +99,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_GET_NODES_STATE,
NEU_RESP_GET_NODES_STATE,
NEU_REQ_NODE_CTL,
NEU_REQ_GET_NODE_STAT,
NEU_RESP_GET_NODE_STAT,
NEU_REQ_ADD_GROUP,
NEU_REQ_DEL_GROUP,
@@ -126,6 +162,8 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_GET_NODES_STATE] = "NEU_REQ_GET_NODES_STATE",
[NEU_RESP_GET_NODES_STATE] = "NEU_RESP_GET_NODES_STATE",
[NEU_REQ_NODE_CTL] = "NEU_REQ_NODE_CTL",
[NEU_REQ_GET_NODE_STAT] = "NEU_REQ_GET_NODE_STAT",
[NEU_RESP_GET_NODE_STAT] = "NEU_RESP_GET_NODE_STAT",
[NEU_REQ_ADD_GROUP] = "NEU_REQ_ADD_GROUP",
[NEU_REQ_DEL_GROUP] = "NEU_REQ_DEL_GROUP",
@@ -373,6 +411,15 @@ typedef struct {
UT_array *states; // array of neu_nodes_state_t
} neu_resp_get_nodes_state_t, neu_reqresp_nodes_state_t;
typedef struct neu_req_get_node_stat {
char node[NEU_NODE_NAME_LEN];
} neu_req_get_node_stat_t;
typedef struct neu_resp_get_node_stat {
neu_node_type_e type;
uint64_t data[NEU_NODE_STAT_MAX];
} neu_resp_get_node_stat_t;
typedef struct neu_req_update_license {
} neu_req_update_license_t;
@@ -434,6 +481,7 @@ 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);
union {
struct {
plugins/ekuiper/read_write.c
+14 -3
View File
@@ -29,6 +29,8 @@
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;
char * json_str = NULL;
json_read_resp_t resp = {
@@ -55,7 +57,10 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
free(json_str);
rv = nng_sendmsg(plugin->sock, msg,
NNG_FLAG_NONBLOCK); // TODO: use aio to send message
if (0 != rv) {
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);
} else {
plog_error(plugin, "nng cannot send msg");
nng_msg_free(msg);
}
@@ -66,8 +71,11 @@ void recv_data_callback(void *arg)
int rv = 0;
neu_plugin_t * plugin = arg;
nng_msg * msg = NULL;
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;
rv = nng_aio_result(plugin->recv_aio);
if (0 != rv) {
@@ -77,8 +85,11 @@ void recv_data_callback(void *arg)
msg = nng_aio_get_msg(plugin->recv_aio);
json_str = nng_msg_body(msg);
plog_debug(plugin, "<< %.*s", (int) nng_msg_len(msg), json_str);
if (json_decode_write_req(json_str, nng_msg_len(msg), &req) < 0) {
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);
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);
goto recv_data_callback_end;
plugins/modbus/modbus_req.c
+14 -1
View File
@@ -49,9 +49,16 @@ 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;
plog_send_protocol(plugin, bytes, n_byte);
return neu_conn_send(plugin->conn, 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);
}
return ret;
}
int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
@@ -296,13 +303,19 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
uint8_t * recv_buf = calloc(256, 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;
ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
if (ret > 0) {
stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_RECV, 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/restful/adapter_handle.c
+44
View File
@@ -224,6 +224,50 @@ void handle_node_ctl(nng_aio *aio)
})
}
void handle_get_node_stat(nng_aio *aio)
{
neu_plugin_t * plugin = neu_rest_get_plugin();
char node_name[NEU_NODE_NAME_LEN] = { 0 };
int ret = 0;
neu_reqresp_head_t header = { 0 };
neu_req_get_node_state_t cmd = { 0 };
VALIDATE_JWT(aio);
header.ctx = aio;
if (http_get_param_str(aio, "node", node_name, sizeof(node_name)) <= 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
http_response(aio, NEU_ERR_PARAM_IS_WRONG, result_error);
});
return;
} else {
header.type = NEU_REQ_GET_NODE_STAT;
strcpy(cmd.node, node_name);
}
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
}
void handle_get_node_stat_resp(nng_aio *aio, neu_resp_get_node_stat_t *stat)
{
neu_json_get_node_stat_resp_t res = { 0 };
char * result = NULL;
res.type = stat->type;
res.data = stat->data;
neu_json_encode_by_fn(&res, neu_json_encode_get_node_stat_resp, &result);
http_ok(aio, result);
free(result);
}
void handle_get_node_state(nng_aio *aio)
{
neu_plugin_t * plugin = neu_rest_get_plugin();
plugins/restful/adapter_handle.h
+2
View File
@@ -34,7 +34,9 @@ void handle_get_node_setting_resp(nng_aio * aio,
neu_resp_get_node_setting_t *setting);
void handle_node_ctl(nng_aio *aio);
void handle_get_node_stat(nng_aio *aio);
void handle_get_node_state(nng_aio *aio);
void handle_get_node_stat_resp(nng_aio *aio, neu_resp_get_node_stat_t *stat);
void handle_get_node_state_resp(nng_aio *aio, neu_resp_get_node_state_t *state);
void handle_get_nodes_state_resp(nng_aio * aio,
neu_resp_get_nodes_state_t *states);
plugins/restful/handle.c
+6
View File
@@ -271,6 +271,12 @@ struct neu_rest_handler api_handlers[] = {
.url = "/api/v2/node/ctl",
.value.handler = handle_node_ctl,
},
{
.method = NEU_REST_METHOD_GET,
.type = NEU_REST_HANDLER_FUNCTION,
.url = "/api/v2/node/stat",
.value.handler = handle_get_node_stat,
},
{
.method = NEU_REST_METHOD_GET,
.type = NEU_REST_HANDLER_FUNCTION,
plugins/restful/rest.c
+4
View File
@@ -294,6 +294,10 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
handle_get_nodes_state_resp(header->ctx,
(neu_resp_get_nodes_state_t *) data);
break;
case NEU_RESP_GET_NODE_STAT:
handle_get_node_stat_resp(header->ctx,
(neu_resp_get_node_stat_t *) data);
break;
case NEU_RESP_GET_NODE_STATE:
handle_get_node_state_resp(header->ctx,
(neu_resp_get_node_state_t *) data);
src/adapter/adapter.c
+51 -4
View File
@@ -40,11 +40,13 @@
#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,
void *data);
static int adapter_response(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 void adapter_stat_acc(neu_adapter_t *adapter, neu_node_stat_e s,
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);
@@ -52,6 +54,7 @@ static int update_timestamp(void *usr_data);
static const adapter_callbacks_t callback_funs = {
.command = adapter_command,
.response = adapter_response,
.stat_acc = adapter_stat_acc,
};
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
@@ -80,6 +83,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
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->persister = neu_persister_create("persistence");
assert(adapter->persister != NULL);
@@ -164,6 +168,31 @@ 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)
{
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;
// 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");
}
}
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data)
{
@@ -200,6 +229,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
break;
}
case NEU_REQ_NODE_CTL:
case NEU_REQ_GET_NODE_STAT:
case NEU_REQ_GET_NODE_STATE:
case NEU_REQ_GET_NODE_SETTING:
case NEU_REQ_NODE_SETTING: {
@@ -279,6 +309,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQRESP_NODE_DELETED:
case NEU_RESP_GET_SUB_DRIVER_TAGS:
case NEU_REQ_UPDATE_LICENSE:
case NEU_RESP_GET_NODE_STAT:
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_GET_NODES_STATE:
case NEU_RESP_GET_NODE_SETTING:
@@ -360,6 +391,16 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_GET_NODE_STAT: {
neu_resp_get_node_stat_t resp = { 0 };
resp.type = neu_adapter_get_type(adapter);
memcpy(resp.data, adapter->stat.data, sizeof(resp.data));
header->type = NEU_RESP_GET_NODE_STAT;
neu_msg_exchange(header);
reply(adapter, header, &resp);
break;
}
case NEU_REQ_GET_NODE_STATE: {
neu_resp_get_node_state_t resp = { 0 };
@@ -930,9 +971,15 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_NODE_CTL:
data_size = sizeof(neu_req_node_ctl_t);
break;
case NEU_REQ_GET_NODE_STAT:
data_size = sizeof(neu_req_get_node_stat_t);
break;
case NEU_REQ_GET_NODE_STATE:
data_size = sizeof(neu_req_get_node_state_t);
break;
case NEU_RESP_GET_NODE_STAT:
data_size = sizeof(neu_resp_get_node_stat_t);
break;
case NEU_RESP_GET_NODE_STATE:
data_size = sizeof(neu_resp_get_node_state_t);
break;
src/adapter/adapter_internal.h
+13
View File
@@ -52,6 +52,19 @@ struct neu_adapter {
neu_event_io_t * nng_io;
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 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;
};
typedef void (*adapter_handler)(neu_adapter_t * adapter,
src/adapter/driver/driver.c
+4
View File
@@ -99,11 +99,15 @@ 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);
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);
}
nlog_debug(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
src/core/manager.c
+2
View File
@@ -458,6 +458,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_NODE_SETTING:
case NEU_REQ_READ_GROUP:
case NEU_REQ_WRITE_TAG:
case NEU_REQ_GET_NODE_STAT:
case NEU_REQ_GET_NODE_STATE:
case NEU_REQ_GET_TAG:
case NEU_REQ_NODE_CTL:
@@ -554,6 +555,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_TAG:
case NEU_RESP_GET_GROUP:
case NEU_RESP_GET_NODE_SETTING:
case NEU_RESP_GET_NODE_STAT:
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_ERROR:
case NEU_RESP_READ_GROUP:
src/parser/neu_json_node.c
+28
View File
@@ -27,6 +27,7 @@
#include "json/json.h"
#include "adapter.h"
#include "neu_json_node.h"
int neu_json_decode_add_node_req(char *buf, neu_json_add_node_req_t **result)
@@ -123,6 +124,33 @@ void neu_json_decode_del_node_req_free(neu_json_del_node_req_t *req)
free(req);
}
int neu_json_encode_get_node_stat_resp(void *json_object, void *param)
{
int ret = 0;
neu_json_get_node_stat_resp_t *resp =
(neu_json_get_node_stat_resp_t *) param;
for (int i = 0; i < NEU_NODE_STAT_MAX; ++i) {
if (i >= NEU_NODE_STAT_TAG_TOT_CNT &&
NEU_NA_TYPE_DRIVER != resp->type) {
// not a driver node, filter out driver specific counters
break;
}
neu_json_elem_t resp_elems[] = {
{
.name = (char *) neu_node_stat_string(i),
.t = NEU_JSON_INT,
.v.val_int = resp->data[i],
},
};
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
}
return ret;
}
int neu_json_encode_get_node_state_resp(void *json_object, void *param)
{
int ret = 0;
src/parser/neu_json_node.h
+7
View File
@@ -46,6 +46,13 @@ typedef struct {
int neu_json_decode_del_node_req(char *buf, neu_json_del_node_req_t **result);
void neu_json_decode_del_node_req_free(neu_json_del_node_req_t *req);
typedef struct {
uint64_t type;
uint64_t *data;
} neu_json_get_node_stat_resp_t;
int neu_json_encode_get_node_stat_resp(void *json_object, void *param);
typedef struct {
int64_t running;
int64_t link;