diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index a41df027..e93359c4 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -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 { diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c index d11b612d..89d4ce36 100644 --- a/plugins/ekuiper/read_write.c +++ b/plugins/ekuiper/read_write.c @@ -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; diff --git a/plugins/modbus/modbus_req.c b/plugins/modbus/modbus_req.c index 7832ed64..56ddac8b 100644 --- a/plugins/modbus/modbus_req.c +++ b/plugins/modbus/modbus_req.c @@ -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); diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index fb52a515..bb4a0276 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -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(); diff --git a/plugins/restful/adapter_handle.h b/plugins/restful/adapter_handle.h index 693263d5..206cf10e 100644 --- a/plugins/restful/adapter_handle.h +++ b/plugins/restful/adapter_handle.h @@ -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); diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 426c4654..a2b17341 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -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, diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 984573c8..7f79a3e7 100644 --- a/plugins/restful/rest.c +++ b/plugins/restful/rest.c @@ -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); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 5e483b11..6c8cb28b 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 68595183..831c500a 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -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, diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 0a5fd1a9..74ca6d9e 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -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, diff --git a/src/core/manager.c b/src/core/manager.c index 2478d4f7..dd56843d 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -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: diff --git a/src/parser/neu_json_node.c b/src/parser/neu_json_node.c index c2b2f486..14def345 100644 --- a/src/parser/neu_json_node.c +++ b/src/parser/neu_json_node.c @@ -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; diff --git a/src/parser/neu_json_node.h b/src/parser/neu_json_node.h index 9a549726..c6d60d49 100644 --- a/src/parser/neu_json_node.h +++ b/src/parser/neu_json_node.h @@ -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;