From 544579897f9d90f13b0cc32ef179ffe5cfba604f Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 16 Jan 2023 15:04:46 +0800 Subject: [PATCH 1/3] fix: node setting api --- include/neuron/adapter.h | 1 + include/neuron/json/json.h | 2 ++ plugins/restful/adapter_handle.c | 27 +++++++++++++++++++++------ src/adapter/adapter.c | 1 + src/core/manager.c | 6 +++--- src/parser/neu_json_node.c | 29 +++++++++++++++++++++++++++++ src/parser/neu_json_node.h | 8 ++++++++ src/utils/json.c | 32 ++++++++++++++++++++++++++++++++ 8 files changed, 97 insertions(+), 9 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index de973ef4..883b92cd 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -363,6 +363,7 @@ typedef struct neu_req_get_node_setting { } neu_req_get_node_setting_t; typedef struct neu_resp_get_node_setting { + char node[NEU_NODE_NAME_LEN]; char *setting; } neu_resp_get_node_setting_t; diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index af2a22d5..bfa3e452 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -90,6 +90,8 @@ int neu_json_encode(void *json_object, char **str); int neu_json_dump_key(void *object, const char *key, char **const result, bool must_exist); +int neu_json_load_key(void *object, const char *key, const char *input, + bool must_exist); #ifdef __cplusplus } diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index f4b0c1d0..9f77536d 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -157,17 +157,16 @@ void handle_set_node_setting(nng_aio *aio) NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT( aio, neu_json_node_setting_req_t, neu_json_decode_node_setting_req, { - char *config_buf = calloc(req_data_size + 1, sizeof(char)); - int ret = 0; + int ret = 0; neu_reqresp_head_t header = { 0 }; neu_req_node_setting_t cmd = { 0 }; - memcpy(config_buf, req_data, req_data_size); header.ctx = aio; header.type = NEU_REQ_NODE_SETTING; strcpy(cmd.node, req->node); - cmd.setting = config_buf; - ret = neu_plugin_op(plugin, header, &cmd); + cmd.setting = req->setting; + req->setting = NULL; // ownership moved + ret = neu_plugin_op(plugin, header, &cmd); if (ret != 0) { NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); @@ -209,8 +208,24 @@ void handle_get_node_setting(nng_aio *aio) void handle_get_node_setting_resp(nng_aio * aio, neu_resp_get_node_setting_t *setting) { - neu_http_ok(aio, setting->setting); + char * json_str = NULL; + neu_json_get_node_setting_resp_t resp = { + .node = setting->node, + .setting = setting->setting, + }; + + if (0 != + neu_json_encode_by_fn(&resp, neu_json_encode_get_node_setting_resp, + &json_str)) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, { + neu_http_response(aio, error_code.error, result_error); + }) + } else { + neu_http_ok(aio, json_str); + } + free(setting->setting); + free(json_str); } void handle_node_ctl(nng_aio *aio) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 951ecd02..8001563c 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -468,6 +468,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(adapter, header, &error); } else { header->type = NEU_RESP_GET_NODE_SETTING; + strcpy(resp.node, adapter->name); reply(adapter, header, &resp); } diff --git a/src/core/manager.c b/src/core/manager.c index 298ddb18..5c2c8a7c 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -527,8 +527,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_ADD_NODE_EVENT: case NEU_REQ_DEL_NODE_EVENT: - case NEU_REQ_NODE_CTL_EVENT: - case NEU_REQ_NODE_SETTING_EVENT: { + case NEU_REQ_NODE_CTL_EVENT: { if (neu_node_manager_find(manager->node_manager, header->receiver)) { forward_msg(manager, msg, header->receiver); } @@ -593,7 +592,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - case NEU_REQ_NODE_SETTING: { + case NEU_REQ_NODE_SETTING: + case NEU_REQ_NODE_SETTING_EVENT: { neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; if (neu_node_manager_find(manager->node_manager, header->receiver) == diff --git a/src/parser/neu_json_node.c b/src/parser/neu_json_node.c index 96bad5d9..86e69820 100644 --- a/src/parser/neu_json_node.c +++ b/src/parser/neu_json_node.c @@ -356,12 +356,19 @@ int neu_json_decode_node_setting_req(char * buf, goto decode_fail; } + ret = neu_json_dump_key(json_obj, "params", &req->setting, true); + if (0 != ret) { + goto decode_fail; + } + req->node = req_elems[0].v.val_str; *result = req; goto decode_exit; decode_fail: + free(req->node); + free(req->setting); free(req); ret = -1; @@ -375,5 +382,27 @@ decode_exit: void neu_json_decode_node_setting_req_free(neu_json_node_setting_req_t *req) { free(req->node); + free(req->setting); free(req); } + +int neu_json_encode_get_node_setting_resp(void *json_object, void *param) +{ + int ret = 0; + neu_json_get_node_setting_resp_t *resp = param; + + neu_json_elem_t resp_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = resp->node, + } }; + + ret = neu_json_encode_field(json_object, resp_elems, + NEU_JSON_ELEM_SIZE(resp_elems)); + + if (0 == ret) { + ret = neu_json_load_key(json_object, "params", resp->setting, true); + } + + return ret; +} diff --git a/src/parser/neu_json_node.h b/src/parser/neu_json_node.h index 8a95b865..ed9529b3 100644 --- a/src/parser/neu_json_node.h +++ b/src/parser/neu_json_node.h @@ -99,12 +99,20 @@ void neu_json_decode_node_ctl_req_free(neu_json_node_ctl_req_t *req); typedef struct { char *node; + char *setting; } neu_json_node_setting_req_t; int neu_json_decode_node_setting_req(char * buf, neu_json_node_setting_req_t **result); void neu_json_decode_node_setting_req_free(neu_json_node_setting_req_t *req); +typedef struct { + char *node; + char *setting; +} neu_json_get_node_setting_resp_t; + +int neu_json_encode_get_node_setting_resp(void *json_object, void *param); + #ifdef __cplusplus } #endif diff --git a/src/utils/json.c b/src/utils/json.c index 36ede662..52c748ce 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -417,6 +417,7 @@ int neu_json_dump_key(void *object, const char *key, char **const result, value = json_object_get(object, key); if (NULL == value) { if (must_exist) { + zlog_error(neuron, "json has no key `%s`", key); return -1; } else { *result = NULL; @@ -432,3 +433,34 @@ int neu_json_dump_key(void *object, const char *key, char **const result, json_decref(new); return rv; } + +int neu_json_load_key(void *object, const char *key, const char *input, + bool must_exist) +{ + int rv = 0; + json_t * input_object = NULL; + json_t * value = NULL; + json_error_t error = {}; + + input_object = json_loads(input, 0, &error); + if (NULL == input_object) { + zlog_error(neuron, + "json load error, line: %d, column: %d, position: %d, info: " + "%s", + error.line, error.column, error.position, error.text); + return -1; + } + + value = json_object_get(input_object, key); + if (NULL == value) { + if (must_exist) { + zlog_error(neuron, "json has no key `%s`", key); + rv = -1; + } + } else if (json_object_set(object, key, value) < 0) { + rv = -1; + } + + json_decref(input_object); + return rv; +} From e689a4d7c0003cc8d4942674002c916e7e4c8cdd Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 16 Jan 2023 17:32:28 +0800 Subject: [PATCH 2/3] feat: add json encoder for node add/del/ctl/setting --- src/parser/neu_json_node.c | 82 ++++++++++++++++++++++++++++++++++++++ src/parser/neu_json_node.h | 4 ++ 2 files changed, 86 insertions(+) diff --git a/src/parser/neu_json_node.c b/src/parser/neu_json_node.c index 86e69820..e57ed2ee 100644 --- a/src/parser/neu_json_node.c +++ b/src/parser/neu_json_node.c @@ -29,6 +29,28 @@ #include "neu_json_node.h" +int neu_json_encode_add_node_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_add_node_req_t *req = param; + + neu_json_elem_t req_elems[] = { { + .name = "plugin", + .t = NEU_JSON_STR, + .v.val_str = req->plugin, + }, + { + .name = "name", + .t = NEU_JSON_STR, + .v.val_str = req->name, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_add_node_req(char *buf, neu_json_add_node_req_t **result) { int ret = 0; @@ -80,6 +102,23 @@ void neu_json_decode_add_node_req_free(neu_json_add_node_req_t *req) free(req); } +int neu_json_encode_del_node_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_del_node_req_t *req = param; + + neu_json_elem_t req_elems[] = { { + .name = "name", + .t = NEU_JSON_STR, + .v.val_str = req->name, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_del_node_req(char *buf, neu_json_del_node_req_t **result) { int ret = 0; @@ -285,6 +324,28 @@ void neu_json_decode_update_node_req_free(neu_json_update_node_req_t *req) free(req); } +int neu_json_encode_node_ctl_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_node_ctl_req_t *req = param; + + neu_json_elem_t req_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = req->node, + }, + { + .name = "cmd", + .t = NEU_JSON_INT, + .v.val_int = req->cmd, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_node_ctl_req(char *buf, neu_json_node_ctl_req_t **result) { int ret = 0; @@ -333,6 +394,27 @@ void neu_json_decode_node_ctl_req_free(neu_json_node_ctl_req_t *req) free(req); } +int neu_json_encode_node_setting_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_node_setting_req_t *req = param; + + neu_json_elem_t req_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = req->node, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + if (0 == ret) { + ret = neu_json_load_key(json_object, "params", req->setting, true); + } + + return ret; +} + int neu_json_decode_node_setting_req(char * buf, neu_json_node_setting_req_t **result) { diff --git a/src/parser/neu_json_node.h b/src/parser/neu_json_node.h index ed9529b3..717f6f56 100644 --- a/src/parser/neu_json_node.h +++ b/src/parser/neu_json_node.h @@ -36,6 +36,7 @@ typedef struct { char *plugin; } neu_json_add_node_req_t; +int neu_json_encode_add_node_req(void *json_object, void *param); int neu_json_decode_add_node_req(char *buf, neu_json_add_node_req_t **result); void neu_json_decode_add_node_req_free(neu_json_add_node_req_t *req); @@ -43,6 +44,7 @@ typedef struct { char *name; } neu_json_del_node_req_t; +int neu_json_encode_del_node_req(void *json_object, void *param); 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); @@ -94,6 +96,7 @@ typedef struct { int64_t cmd; } neu_json_node_ctl_req_t; +int neu_json_encode_node_ctl_req(void *json_object, void *param); int neu_json_decode_node_ctl_req(char *buf, neu_json_node_ctl_req_t **result); void neu_json_decode_node_ctl_req_free(neu_json_node_ctl_req_t *req); @@ -102,6 +105,7 @@ typedef struct { char *setting; } neu_json_node_setting_req_t; +int neu_json_encode_node_setting_req(void *json_object, void *param); int neu_json_decode_node_setting_req(char * buf, neu_json_node_setting_req_t **result); void neu_json_decode_node_setting_req_free(neu_json_node_setting_req_t *req); From d358dca495125472f072d87383a5d47988e0f187 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 16 Jan 2023 16:25:24 +0800 Subject: [PATCH 3/3] feat(monitor): handle node events --- plugins/monitor/CMakeLists.txt | 1 + plugins/monitor/monitor.c | 5 +- plugins/monitor/monitor_config.c | 60 +++++++++++++++++++++ plugins/monitor/monitor_config.h | 6 +++ plugins/monitor/mqtt_handle.c | 93 ++++++++++++++++++++++++++++++++ plugins/monitor/mqtt_handle.h | 2 + 6 files changed, 166 insertions(+), 1 deletion(-) diff --git a/plugins/monitor/CMakeLists.txt b/plugins/monitor/CMakeLists.txt index e87239da..10bc1e4e 100644 --- a/plugins/monitor/CMakeLists.txt +++ b/plugins/monitor/CMakeLists.txt @@ -14,6 +14,7 @@ add_library(plugin-monitor SHARED ${src}) target_include_directories(plugin-monitor PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron + ${CMAKE_SOURCE_DIR}/src ) target_link_libraries(plugin-monitor neuron-base nng) diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 0ca60247..fd82d88b 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -394,8 +394,11 @@ static int monitor_plugin_request(neu_plugin_t * plugin, handle_nodes_state(header->ctx, (neu_resp_get_nodes_state_t *) data); break; } + case NEU_REQ_ADD_NODE_EVENT: + case NEU_REQ_DEL_NODE_EVENT: + case NEU_REQ_NODE_CTL_EVENT: case NEU_REQ_NODE_SETTING_EVENT: { - free(((neu_req_node_setting_t *) data)->setting); + handle_events(plugin, header->type, data); break; } default: diff --git a/plugins/monitor/monitor_config.c b/plugins/monitor/monitor_config.c index 3e8c3d4d..cc2ea715 100644 --- a/plugins/monitor/monitor_config.c +++ b/plugins/monitor/monitor_config.c @@ -20,6 +20,7 @@ #include "json/json.h" #include "json/neu_json_param.h" +#include "utils/asprintf.h" #include "utils/base64.h" #include "utils/log.h" @@ -118,6 +119,48 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting, return 0; } +static int make_event_topics(monitor_config_t *config, const char *prefix) +{ + char *node_add_topic = NULL; + char *node_del_topic = NULL; + char *node_ctl_topic = NULL; + char *node_setting_topic = NULL; + + neu_asprintf(&node_add_topic, "%s/node/add", prefix); + if (NULL == node_add_topic) { + goto error; + } + + neu_asprintf(&node_del_topic, "%s/node/delete", prefix); + if (NULL == node_del_topic) { + goto error; + } + + neu_asprintf(&node_ctl_topic, "%s/node/ctl", prefix); + if (NULL == node_ctl_topic) { + goto error; + } + + neu_asprintf(&node_setting_topic, "%s/node/setting", prefix); + if (NULL == node_setting_topic) { + goto error; + } + + config->node_add_topic = node_add_topic; + config->node_del_topic = node_del_topic; + config->node_ctl_topic = node_ctl_topic; + config->node_setting_topic = node_setting_topic; + + return 0; + +error: + free(node_setting_topic); + free(node_ctl_topic); + free(node_del_topic); + free(node_add_topic); + return -1; +} + int monitor_config_parse(neu_plugin_t *plugin, const char *setting, monitor_config_t *config) { @@ -203,6 +246,12 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting, goto error; } + ret = make_event_topics(config, event_topic_prefix.v.val_str); + if (0 != ret) { + plog_error(plugin, "make event topics fail"); + goto error; + } + config->client_id = client_id.v.val_str; config->event_topic_prefix = event_topic_prefix.v.val_str; config->heartbeat_interval = heartbeat_interval.v.val_int; @@ -245,6 +294,12 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting, plog_info(plugin, "config keypass : %s", placeholder); } + plog_info(plugin, "node add event topic : %s", config->node_add_topic); + plog_info(plugin, "node del event topic : %s", config->node_del_topic); + plog_info(plugin, "node ctl event topic : %s", config->node_ctl_topic); + plog_info(plugin, "node setting event topic : %s", + config->node_setting_topic); + return 0; error: @@ -278,5 +333,10 @@ void monitor_config_fini(monitor_config_t *config) free(config->key); free(config->keypass); + free(config->node_add_topic); + free(config->node_del_topic); + free(config->node_ctl_topic); + free(config->node_setting_topic); + memset(config, 0, sizeof(*config)); } diff --git a/plugins/monitor/monitor_config.h b/plugins/monitor/monitor_config.h index d923b3e4..46675f33 100644 --- a/plugins/monitor/monitor_config.h +++ b/plugins/monitor/monitor_config.h @@ -42,6 +42,12 @@ typedef struct { char * cert; // client cert char * key; // client key char * keypass; // client key password + + // event topics generated according to event_topic_prefix + char *node_add_topic; + char *node_del_topic; + char *node_ctl_topic; + char *node_setting_topic; } monitor_config_t; int monitor_config_parse(neu_plugin_t *plugin, const char *setting, diff --git a/plugins/monitor/mqtt_handle.c b/plugins/monitor/mqtt_handle.c index 2fda0501..4ae0d00e 100644 --- a/plugins/monitor/mqtt_handle.c +++ b/plugins/monitor/mqtt_handle.c @@ -23,6 +23,8 @@ #include "json/neu_json_mqtt.h" #include "json/neu_json_rw.h" +#include "parser/neu_json_node.h" + #include "monitor.h" #include "mqtt_handle.h" @@ -56,6 +58,61 @@ static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states) return json_str; } +static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, + void *data, char **topic_p) +{ + char *json_str = NULL; + int (*encode_fn)(void *json_object, void *param) = NULL; + union { + neu_json_add_node_req_t add_node; + neu_json_del_node_req_t del_node; + neu_json_node_ctl_req_t node_ctl; + neu_json_node_setting_req_t node_setting; + } json_req; + + switch (event) { + case NEU_REQ_ADD_NODE_EVENT: { + neu_req_add_node_t *add_node = data; + json_req.add_node.name = add_node->node; + json_req.add_node.plugin = add_node->plugin; + encode_fn = neu_json_encode_add_node_req; + *topic_p = plugin->config->node_add_topic; + break; + } + case NEU_REQ_DEL_NODE_EVENT: { + neu_req_del_node_t *del_node = data; + json_req.del_node.name = del_node->node; + encode_fn = neu_json_encode_del_node_req; + *topic_p = plugin->config->node_del_topic; + break; + } + case NEU_REQ_NODE_CTL_EVENT: { + neu_req_node_ctl_t *node_ctl = data; + json_req.node_ctl.node = node_ctl->node; + json_req.node_ctl.cmd = node_ctl->ctl; + encode_fn = neu_json_encode_node_ctl_req; + *topic_p = plugin->config->node_ctl_topic; + break; + } + case NEU_REQ_NODE_SETTING_EVENT: { + neu_req_node_setting_t *node_setting = data; + json_req.node_setting.node = node_setting->node; + json_req.node_setting.setting = node_setting->setting; + encode_fn = neu_json_encode_node_setting_req; + *topic_p = plugin->config->node_setting_topic; + break; + } + default: { + plog_error(plugin, "unsupported event:%s", + neu_reqresp_type_string(event)); + return NULL; + } + } + + neu_json_encode_by_fn(&json_req, encode_fn, &json_str); + return json_str; +} + static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic, uint8_t *payload, uint32_t len, void *data) { @@ -132,3 +189,39 @@ end: return rv; } + +int handle_events(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data) +{ + int rv = 0; + char *json_str = NULL; + char *topic = NULL; + + if (NULL == plugin->mqtt_client) { + rv = NEU_ERR_MQTT_IS_NULL; + goto end; + } + + if (!neu_mqtt_client_is_connected(plugin->mqtt_client)) { + // cache disable and we are disconnected + rv = NEU_ERR_MQTT_FAILURE; + goto end; + } + + json_str = generate_event_json(plugin, event, data, &topic); + if (NULL == json_str) { + plog_error(plugin, "generate event:%s json fail", + neu_reqresp_type_string(event)); + rv = NEU_ERR_EINTERNAL; + goto end; + } + + neu_mqtt_qos_e qos = NEU_MQTT_QOS0; + rv = publish(plugin, qos, topic, json_str, strlen(json_str)); + json_str = NULL; + +end: + if (NEU_REQ_NODE_SETTING_EVENT == event) { + free(((neu_req_node_setting_t *) data)->setting); + } + return rv; +} diff --git a/plugins/monitor/mqtt_handle.h b/plugins/monitor/mqtt_handle.h index faae1d03..208f7817 100644 --- a/plugins/monitor/mqtt_handle.h +++ b/plugins/monitor/mqtt_handle.h @@ -28,6 +28,8 @@ extern "C" { int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states); +int handle_events(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data); + #ifdef __cplusplus } #endif