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

Merge pull request #1036 from eeff/eeff/dev

Browse Source 
feat(monitor): handle group events
This commit is contained in:
eeff
2023-01-18 22:11:01 +08:00
committed by GitHub
parent 257341bca1 b7aecf74d3
commit 6f7163e2e7
9 changed files with 143 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+6
View File
@@ -98,6 +98,9 @@ typedef enum neu_reqresp_type {
NEU_REQ_DEL_NODE_EVENT,
NEU_REQ_NODE_CTL_EVENT,
NEU_REQ_NODE_SETTING_EVENT,
NEU_REQ_ADD_GROUP_EVENT,
NEU_REQ_DEL_GROUP_EVENT,
NEU_REQ_UPDATE_GROUP_EVENT,
} neu_reqresp_type_e;
static const char *neu_reqresp_type_string_t[] = {
@@ -162,6 +165,9 @@ static const char *neu_reqresp_type_string_t[] = {
[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",
[NEU_REQ_ADD_GROUP_EVENT] = "NEU_REQ_ADD_GROUP_EVENT",
[NEU_REQ_DEL_GROUP_EVENT] = "NEU_REQ_DEL_GROUP_EVENT",
[NEU_REQ_UPDATE_GROUP_EVENT] = "NEU_REQ_UPDATE_GROUP_EVENT",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
plugins/monitor/monitor.c
+4 -1
View File
@@ -397,7 +397,10 @@ static int monitor_plugin_request(neu_plugin_t * plugin,
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_SETTING_EVENT:
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT: {
handle_events(plugin, header->type, data);
break;
}
plugins/monitor/monitor_config.c
+31
View File
@@ -125,6 +125,9 @@ static int make_event_topics(monitor_config_t *config, const char *prefix)
char *node_del_topic = NULL;
char *node_ctl_topic = NULL;
char *node_setting_topic = NULL;
char *group_add_topic = NULL;
char *group_del_topic = NULL;
char *group_update_topic = NULL;
neu_asprintf(&node_add_topic, "%s/node/add", prefix);
if (NULL == node_add_topic) {
@@ -146,14 +149,35 @@ static int make_event_topics(monitor_config_t *config, const char *prefix)
goto error;
}
neu_asprintf(&group_add_topic, "%s/group/add", prefix);
if (NULL == group_add_topic) {
goto error;
}
neu_asprintf(&group_del_topic, "%s/group/delete", prefix);
if (NULL == group_del_topic) {
goto error;
}
neu_asprintf(&group_update_topic, "%s/group/update", prefix);
if (NULL == group_update_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;
config->group_add_topic = group_add_topic;
config->group_del_topic = group_del_topic;
config->group_update_topic = group_update_topic;
return 0;
error:
free(group_update_topic);
free(group_del_topic);
free(group_add_topic);
free(node_setting_topic);
free(node_ctl_topic);
free(node_del_topic);
@@ -299,6 +323,10 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
plog_info(plugin, "node ctl event topic : %s", config->node_ctl_topic);
plog_info(plugin, "node setting event topic : %s",
config->node_setting_topic);
plog_info(plugin, "group add topic : %s", config->group_add_topic);
plog_info(plugin, "group del topic : %s", config->group_del_topic);
plog_info(plugin, "group update topic : %s",
config->group_update_topic);
return 0;
@@ -337,6 +365,9 @@ void monitor_config_fini(monitor_config_t *config)
free(config->node_del_topic);
free(config->node_ctl_topic);
free(config->node_setting_topic);
free(config->group_add_topic);
free(config->group_del_topic);
free(config->group_update_topic);
memset(config, 0, sizeof(*config));
}
plugins/monitor/monitor_config.h
+3
View File
@@ -48,6 +48,9 @@ typedef struct {
char *node_del_topic;
char *node_ctl_topic;
char *node_setting_topic;
char *group_add_topic;
char *group_del_topic;
char *group_update_topic;
} monitor_config_t;
int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
plugins/monitor/mqtt_handle.c
+34 -4
View File
@@ -23,6 +23,7 @@
#include "json/neu_json_mqtt.h"
#include "json/neu_json_rw.h"
#include "parser/neu_json_group_config.h"
#include "parser/neu_json_node.h"
#include "monitor.h"
@@ -64,10 +65,13 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
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;
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;
neu_json_add_group_config_req_t add_grp;
neu_json_del_group_config_req_t del_grp;
neu_json_update_group_req_t update_grp;
} json_req;
switch (event) {
@@ -102,6 +106,32 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
*topic_p = plugin->config->node_setting_topic;
break;
}
case NEU_REQ_ADD_GROUP_EVENT: {
neu_req_add_group_t *add_grp = data;
json_req.add_grp.node = add_grp->driver;
json_req.add_grp.group = add_grp->group;
json_req.add_grp.interval = add_grp->interval;
encode_fn = neu_json_encode_add_group_config_req;
*topic_p = plugin->config->group_add_topic;
break;
}
case NEU_REQ_DEL_GROUP_EVENT: {
neu_req_del_group_t *del_grp = data;
json_req.del_grp.node = del_grp->driver;
json_req.del_grp.group = del_grp->group;
encode_fn = neu_json_encode_del_group_config_req;
*topic_p = plugin->config->group_del_topic;
break;
}
case NEU_REQ_UPDATE_GROUP_EVENT: {
neu_req_update_group_t *update_grp = data;
json_req.update_grp.node = update_grp->driver;
json_req.update_grp.group = update_grp->group;
json_req.update_grp.interval = update_grp->interval;
encode_fn = neu_json_encode_add_group_config_req;
*topic_p = plugin->config->group_update_topic;
break;
}
default: {
plog_error(plugin, "unsupported event:%s",
neu_reqresp_type_string(event));
src/adapter/adapter.c
+12 -1
View File
@@ -404,6 +404,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_DEL_NODE_EVENT:
case NEU_REQ_NODE_CTL_EVENT:
case NEU_REQ_NODE_SETTING_EVENT:
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT:
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
break;
@@ -549,6 +552,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_add_group(adapter->name, cmd->group, cmd->interval);
notify_monitor(adapter, NEU_REQ_ADD_GROUP_EVENT, cmd);
}
neu_msg_exchange(header);
@@ -575,6 +579,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_update_group(adapter->name, cmd->group,
cmd->interval);
notify_monitor(adapter, NEU_REQ_UPDATE_GROUP_EVENT, cmd);
}
neu_msg_exchange(header);
@@ -595,6 +600,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_del_group(cmd->driver, cmd->group);
notify_monitor(adapter, NEU_REQ_DEL_GROUP_EVENT, cmd);
}
neu_msg_exchange(header);
@@ -1123,10 +1129,15 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
data_size = sizeof(neu_resp_get_node_t);
break;
case NEU_REQ_ADD_GROUP:
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_ADD_GROUP_EVENT:
data_size = sizeof(neu_req_add_group_t);
break;
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_UPDATE_GROUP_EVENT:
data_size = sizeof(neu_req_update_group_t);
break;
case NEU_REQ_DEL_GROUP:
case NEU_REQ_DEL_GROUP_EVENT:
data_size = sizeof(neu_req_del_group_t);
break;
case NEU_REQ_GET_DRIVER_GROUP:
src/core/manager.c
+4 -1
View File
@@ -527,7 +527,10 @@ 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_CTL_EVENT:
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT: {
if (neu_node_manager_find(manager->node_manager, header->receiver)) {
forward_msg(manager, msg, header->receiver);
}
src/parser/neu_json_group_config.c
+47
View File
@@ -31,6 +31,32 @@
#include "neu_json_group_config.h"
int neu_json_encode_add_group_config_req(void *json_object, void *param)
{
int ret = 0;
neu_json_add_group_config_req_t *req = param;
neu_json_elem_t req_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
.v.val_str = req->node,
},
{
.name = "group",
.t = NEU_JSON_STR,
.v.val_str = req->group,
},
{
.name = "interval",
.t = NEU_JSON_INT,
.v.val_int = req->interval,
} };
ret = neu_json_encode_field(json_object, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
return ret;
}
int neu_json_decode_add_group_config_req(
char *buf, neu_json_add_group_config_req_t **result)
{
@@ -89,6 +115,27 @@ void neu_json_decode_add_group_config_req_free(
free(req);
}
int neu_json_encode_del_group_config_req(void *json_object, void *param)
{
int ret = 0;
neu_json_del_group_config_req_t *req = param;
neu_json_elem_t req_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
.v.val_str = req->node,
},
{
.name = "group",
.t = NEU_JSON_STR,
.v.val_str = req->group,
} };
ret = neu_json_encode_field(json_object, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
return ret;
}
int neu_json_decode_del_group_config_req(
char *buf, neu_json_del_group_config_req_t **result)
{
src/parser/neu_json_group_config.h
+2
View File
@@ -37,6 +37,7 @@ typedef struct {
int64_t interval;
} neu_json_add_group_config_req_t, neu_json_update_group_req_t;
int neu_json_encode_add_group_config_req(void *json_object, void *param);
int neu_json_decode_add_group_config_req(
char *buf, neu_json_add_group_config_req_t **result);
void neu_json_decode_add_group_config_req_free(
@@ -47,6 +48,7 @@ typedef struct {
char *node;
} neu_json_del_group_config_req_t;
int neu_json_encode_del_group_config_req(void *json_object, void *param);
int neu_json_decode_del_group_config_req(
char *buf, neu_json_del_group_config_req_t **result);
void neu_json_decode_del_group_config_req_free(