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

Merge pull request #1030 from eeff/eeff/dev

Browse Source 
feat(core): notify monitor of node add/del/ctl/setting events
This commit is contained in:
eeff
2023-01-16 10:05:37 +08:00
committed by GitHub
parent fd27701ef7 c046235a1b
commit f3e350dbae
4 changed files with 96 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+10
View File
@@ -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)
plugins/monitor/monitor.c
+7 -4
View File
@@ -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;
src/adapter/adapter.c
+38 -1
View File
@@ -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:
src/core/manager.c
+41
View File
@@ -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;