diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index d0fce842..de973ef4 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -93,6 +93,11 @@ typedef enum neu_reqresp_type { NEU_REQRESP_NODE_DELETED, NEU_REQ_UPDATE_LOG_LEVEL, + + NEU_REQ_ADD_NODE_EVENT, + NEU_REQ_DEL_NODE_EVENT, + NEU_REQ_NODE_CTL_EVENT, + NEU_REQ_NODE_SETTING_EVENT, } neu_reqresp_type_e; static const char *neu_reqresp_type_string_t[] = { @@ -152,6 +157,11 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQRESP_NODE_DELETED] = "NEU_REQRESP_NODE_DELETED", [NEU_REQ_UPDATE_LOG_LEVEL] = "NEU_REQ_UPDATE_LOG_LEVEL", + + [NEU_REQ_ADD_NODE_EVENT] = "NEU_REQ_ADD_NODE_EVENT", + [NEU_REQ_DEL_NODE_EVENT] = "NEU_REQ_DEL_NODE_EVENT", + [NEU_REQ_NODE_CTL_EVENT] = "NEU_REQ_NODE_CTL_EVENT", + [NEU_REQ_NODE_SETTING_EVENT] = "NEU_REQ_NODE_SETTING_EVENT", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 8f7e8a13..0ca60247 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -311,7 +311,7 @@ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting) rv = monitor_config_parse(plugin, setting, &config); if (0 != rv) { - plog_error(plugin, "neu_mqtt_config_parse fail"); + plog_error(plugin, "monitor_config_parse fail"); return NEU_ERR_NODE_SETTING_INVALID; } @@ -394,10 +394,13 @@ 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_NODE_SETTING_EVENT: { + free(((neu_req_node_setting_t *) data)->setting); + break; + } default: - nlog_fatal("recv unhandle msg: %s", - neu_reqresp_type_string(header->type)); - assert(false); + nlog_warn("recv unsupported msg: %s", + neu_reqresp_type_string(header->type)); break; } return 0; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 51e751ab..951ecd02 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -54,6 +54,8 @@ static int adapter_update_metric(neu_adapter_t *adapter, const char *group); inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); +inline static void notify_monitor(neu_adapter_t * adapter, + neu_reqresp_type_e event, void *data); static int level_check(void *usr_data); static const adapter_callbacks_t callback_funs = { @@ -398,6 +400,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_ERROR: case NEU_REQRESP_TRANS_DATA: case NEU_REQRESP_NODES_STATE: + case NEU_REQ_ADD_NODE_EVENT: + case NEU_REQ_DEL_NODE_EVENT: + case NEU_REQ_NODE_CTL_EVENT: + case NEU_REQ_NODE_SETTING_EVENT: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; @@ -440,8 +446,11 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) error.error = neu_adapter_set_setting(adapter, cmd->setting); if (error.error == NEU_ERR_SUCCESS) { adapter_storage_setting(adapter->name, cmd->setting); + // ownership of `cmd->setting` transfer + notify_monitor(adapter, NEU_REQ_NODE_SETTING_EVENT, cmd); + } else { + free(cmd->setting); } - free(cmd->setting); header->type = NEU_RESP_ERROR; neu_msg_exchange(header); @@ -605,6 +614,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } + if (0 == error.error) { + notify_monitor(adapter, NEU_REQ_NODE_CTL_EVENT, cmd); + } + neu_msg_exchange(header); header->type = NEU_RESP_ERROR; reply(adapter, header, &error); @@ -1024,6 +1037,26 @@ inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, nng_sendmsg(adapter->sock, msg, 0); } +inline static void notify_monitor(neu_adapter_t * adapter, + neu_reqresp_type_e event, void *data) +{ + neu_reqresp_head_t header = { + .receiver = "monitor", + .type = event, + }; + + strncpy(header.sender, adapter->name, NEU_NODE_NAME_LEN); + + nng_msg *msg = neu_msg_gen(&header, data); + + int ret = nng_sendmsg(adapter->sock, msg, 0); + if (ret != 0) { + nng_msg_free(msg); + nlog_warn("notify %s of %s, error: %s", header.receiver, + neu_reqresp_type_string(header.type), nng_strerror(ret)); + } +} + static int level_check(void *usr_data) { neu_adapter_t *adapter = (neu_adapter_t *) usr_data; @@ -1075,9 +1108,11 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_plugin_t); break; case NEU_REQ_ADD_NODE: + case NEU_REQ_ADD_NODE_EVENT: data_size = sizeof(neu_req_add_node_t); break; case NEU_REQ_DEL_NODE: + case NEU_REQ_DEL_NODE_EVENT: data_size = sizeof(neu_req_del_node_t); break; case NEU_REQ_GET_NODE: @@ -1138,6 +1173,7 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_subscribe_group_t); break; case NEU_REQ_NODE_SETTING: + case NEU_REQ_NODE_SETTING_EVENT: data_size = sizeof(neu_req_node_setting_t); break; case NEU_REQ_GET_NODE_SETTING: @@ -1147,6 +1183,7 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_node_setting_t); break; case NEU_REQ_NODE_CTL: + case NEU_REQ_NODE_CTL_EVENT: data_size = sizeof(neu_req_node_ctl_t); break; case NEU_REQ_GET_NODE_STATE: diff --git a/src/core/manager.c b/src/core/manager.c index 95219b64..298ddb18 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -55,6 +55,8 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, nng_pipe pipe); inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *ndoe); +inline static void notify_monitor(neu_manager_t * manager, + neu_reqresp_type_e event, void *data); static void start_static_adapter(neu_manager_t *manager, const char *name); static int update_timestamp(void *usr_data); static void start_single_adapter(neu_manager_t *manager, const char *name, @@ -315,6 +317,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error == NEU_ERR_SUCCESS) { manager_storage_add_node(manager, cmd->node); + notify_monitor(manager, NEU_REQ_ADD_NODE_EVENT, cmd); } header->type = NEU_RESP_ERROR; @@ -354,6 +357,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } header->type = NEU_REQ_NODE_UNINIT; forward_msg(manager, msg, header->receiver); + notify_monitor(manager, NEU_REQ_DEL_NODE_EVENT, cmd); if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) { UT_array *apps = neu_subscribe_manager_find_by_driver( manager->subscribe_manager, cmd->node); @@ -521,6 +525,17 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_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: { + if (neu_node_manager_find(manager->node_manager, header->receiver)) { + forward_msg(manager, msg, header->receiver); + } + + break; + } + case NEU_REQ_DEL_GROUP: { neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1]; @@ -658,6 +673,32 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, } } +inline static void notify_monitor(neu_manager_t * manager, + neu_reqresp_type_e event, void *data) +{ + nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, "monitor"); + if (0 == pipe.id) { + nlog_error("no monitor node"); + return; + } + + neu_reqresp_head_t header = { + .sender = "manager", + .receiver = "monitor", + .type = event, + }; + + nng_msg *msg = neu_msg_gen(&header, data); + nng_msg_set_pipe(msg, pipe); + + int ret = nng_sendmsg(manager->socket, msg, 0); + if (ret != 0) { + nng_msg_free(msg); + nlog_warn("notify %s of %s, error: %s", header.receiver, + neu_reqresp_type_string(header.type), nng_strerror(ret)); + } +} + static void start_static_adapter(neu_manager_t *manager, const char *name) { neu_adapter_t * adapter = NULL;