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

Merge pull request #1040 from eeff/eeff/dev

Browse Source 
feat(monitor): handle tag add/del/update events
This commit is contained in:
eeff
2023-01-31 14:41:32 +08:00
committed by GitHub
parent 90d50c7988 b81eafa0ed
commit 459463d618
9 changed files with 283 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+6
View File
@@ -101,6 +101,9 @@ typedef enum neu_reqresp_type {
NEU_REQ_ADD_GROUP_EVENT,
NEU_REQ_DEL_GROUP_EVENT,
NEU_REQ_UPDATE_GROUP_EVENT,
NEU_REQ_ADD_TAG_EVENT,
NEU_REQ_DEL_TAG_EVENT,
NEU_REQ_UPDATE_TAG_EVENT,
} neu_reqresp_type_e;
static const char *neu_reqresp_type_string_t[] = {
@@ -168,6 +171,9 @@ static const char *neu_reqresp_type_string_t[] = {
[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",
[NEU_REQ_ADD_TAG_EVENT] = "NEU_REQ_ADD_TAG_EVENT",
[NEU_REQ_DEL_TAG_EVENT] = "NEU_REQ_DEL_TAG_EVENT",
[NEU_REQ_UPDATE_TAG_EVENT] = "NEU_REQ_UPDATE_TAG_EVENT",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
plugins/monitor/monitor.c
+4 -1
View File
@@ -400,7 +400,10 @@ static int monitor_plugin_request(neu_plugin_t * plugin,
case NEU_REQ_NODE_SETTING_EVENT:
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT: {
case NEU_REQ_UPDATE_GROUP_EVENT:
case NEU_REQ_ADD_TAG_EVENT:
case NEU_REQ_DEL_TAG_EVENT:
case NEU_REQ_UPDATE_TAG_EVENT: {
handle_events(plugin, header->type, data);
break;
}
plugins/monitor/monitor_config.c
+31
View File
@@ -128,6 +128,9 @@ static int make_event_topics(monitor_config_t *config, const char *prefix)
char *group_add_topic = NULL;
char *group_del_topic = NULL;
char *group_update_topic = NULL;
char *tag_add_topic = NULL;
char *tag_del_topic = NULL;
char *tag_update_topic = NULL;
neu_asprintf(&node_add_topic, "%s/node/add", prefix);
if (NULL == node_add_topic) {
@@ -164,6 +167,21 @@ static int make_event_topics(monitor_config_t *config, const char *prefix)
goto error;
}
neu_asprintf(&tag_add_topic, "%s/tag/add", prefix);
if (NULL == tag_add_topic) {
goto error;
}
neu_asprintf(&tag_del_topic, "%s/tag/delete", prefix);
if (NULL == tag_del_topic) {
goto error;
}
neu_asprintf(&tag_update_topic, "%s/tag/update", prefix);
if (NULL == tag_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;
@@ -171,10 +189,16 @@ static int make_event_topics(monitor_config_t *config, const char *prefix)
config->group_add_topic = group_add_topic;
config->group_del_topic = group_del_topic;
config->group_update_topic = group_update_topic;
config->tag_add_topic = tag_add_topic;
config->tag_del_topic = tag_del_topic;
config->tag_update_topic = tag_update_topic;
return 0;
error:
free(tag_update_topic);
free(tag_del_topic);
free(tag_add_topic);
free(group_update_topic);
free(group_del_topic);
free(group_add_topic);
@@ -327,6 +351,10 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
plog_info(plugin, "group del topic : %s", config->group_del_topic);
plog_info(plugin, "group update topic : %s",
config->group_update_topic);
plog_info(plugin, "tag add topic : %s", config->tag_add_topic);
plog_info(plugin, "tag del topic : %s", config->tag_del_topic);
plog_info(plugin, "tag update topic : %s",
config->tag_update_topic);
return 0;
@@ -368,6 +396,9 @@ void monitor_config_fini(monitor_config_t *config)
free(config->group_add_topic);
free(config->group_del_topic);
free(config->group_update_topic);
free(config->tag_add_topic);
free(config->tag_del_topic);
free(config->tag_update_topic);
memset(config, 0, sizeof(*config));
}
plugins/monitor/monitor_config.h
+3
View File
@@ -51,6 +51,9 @@ typedef struct {
char *group_add_topic;
char *group_del_topic;
char *group_update_topic;
char *tag_add_topic;
char *tag_del_topic;
char *tag_update_topic;
} monitor_config_t;
int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
plugins/monitor/mqtt_handle.c
+87 -12
View File
@@ -25,6 +25,7 @@
#include "parser/neu_json_group_config.h"
#include "parser/neu_json_node.h"
#include "parser/neu_json_tag.h"
#include "monitor.h"
#include "mqtt_handle.h"
@@ -62,8 +63,7 @@ static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states)
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;
char *json_str = NULL;
union {
neu_json_add_node_req_t add_node;
neu_json_del_node_req_t del_node;
@@ -72,38 +72,44 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
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;
neu_json_add_tags_req_t add_tags;
neu_json_del_tags_req_t del_tags;
} 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;
neu_json_encode_by_fn(&json_req, neu_json_encode_add_node_req,
&json_str);
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;
neu_json_encode_by_fn(&json_req, neu_json_encode_del_node_req,
&json_str);
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;
neu_json_encode_by_fn(&json_req, neu_json_encode_node_ctl_req,
&json_str);
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;
neu_json_encode_by_fn(&json_req, neu_json_encode_node_setting_req,
&json_str);
break;
}
case NEU_REQ_ADD_GROUP_EVENT: {
@@ -111,16 +117,18 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
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;
neu_json_encode_by_fn(&json_req, neu_json_encode_add_group_config_req,
&json_str);
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;
neu_json_encode_by_fn(&json_req, neu_json_encode_del_group_config_req,
&json_str);
break;
}
case NEU_REQ_UPDATE_GROUP_EVENT: {
@@ -128,8 +136,61 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
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;
*topic_p = plugin->config->group_update_topic;
neu_json_encode_by_fn(&json_req, neu_json_encode_add_group_config_req,
&json_str);
break;
}
case NEU_REQ_ADD_TAG_EVENT:
case NEU_REQ_UPDATE_TAG_EVENT: {
neu_req_add_tag_t *add_tag = data;
json_req.add_tags.node = add_tag->driver;
json_req.add_tags.group = add_tag->group;
json_req.add_tags.n_tag = add_tag->n_tag;
json_req.add_tags.tags =
calloc(add_tag->n_tag, sizeof(*json_req.add_tags.tags));
if (NULL == json_req.add_tags.tags) {
plog_error(plugin, "calloc fail");
break;
}
for (uint16_t i = 0; i < add_tag->n_tag; ++i) {
json_req.add_tags.tags[i].type = add_tag->tags[i].type;
json_req.add_tags.tags[i].name = add_tag->tags[i].name;
json_req.add_tags.tags[i].attribute = add_tag->tags[i].attribute;
json_req.add_tags.tags[i].address = add_tag->tags[i].address;
json_req.add_tags.tags[i].decimal = add_tag->tags[i].decimal;
json_req.add_tags.tags[i].precision = add_tag->tags[i].precision;
json_req.add_tags.tags[i].description =
add_tag->tags[i].description;
}
*topic_p = NEU_REQ_ADD_TAG_EVENT == event
? plugin->config->tag_add_topic
: plugin->config->tag_update_topic;
neu_json_encode_by_fn(&json_req, neu_json_encode_add_tags_req,
&json_str);
free(json_req.add_tags.tags);
break;
}
case NEU_REQ_DEL_TAG_EVENT: {
neu_req_del_tag_t *del_tag = data;
json_req.del_tags.node = del_tag->driver;
json_req.del_tags.group = del_tag->group;
json_req.del_tags.n_tags = del_tag->n_tag;
json_req.del_tags.tags =
calloc(del_tag->n_tag, sizeof(*json_req.del_tags.tags));
if (NULL == json_req.del_tags.tags) {
plog_error(plugin, "calloc fail");
break;
}
for (uint16_t i = 0; i < del_tag->n_tag; ++i) {
json_req.del_tags.tags[i] = del_tag->tags[i];
}
*topic_p = plugin->config->tag_del_topic;
neu_json_encode_by_fn(&json_req, neu_json_encode_del_tags_req,
&json_str);
free(json_req.del_tags.tags);
break;
}
default: {
@@ -139,7 +200,6 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event,
}
}
neu_json_encode_by_fn(&json_req, encode_fn, &json_str);
return json_str;
}
@@ -252,6 +312,21 @@ int handle_events(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data)
end:
if (NEU_REQ_NODE_SETTING_EVENT == event) {
free(((neu_req_node_setting_t *) data)->setting);
} else if (NEU_REQ_ADD_TAG_EVENT == event ||
NEU_REQ_UPDATE_TAG_EVENT == event) {
neu_req_add_tag_t *add_tag = data;
for (uint16_t i = 0; i < add_tag->n_tag; i++) {
free(add_tag->tags[i].address);
free(add_tag->tags[i].name);
free(add_tag->tags[i].description);
}
free(add_tag->tags);
} else if (NEU_REQ_DEL_TAG_EVENT == event) {
neu_req_del_tag_t *del_tag = data;
for (uint16_t i = 0; i < del_tag->n_tag; i++) {
free(del_tag->tags[i]);
}
free(del_tag->tags);
}
return rv;
}
src/adapter/adapter.c
+32 -13
View File
@@ -407,6 +407,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT:
case NEU_REQ_ADD_TAG_EVENT:
case NEU_REQ_DEL_TAG_EVENT:
case NEU_REQ_UPDATE_TAG_EVENT:
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
break;
@@ -647,10 +650,14 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i]);
if (0 == error.error) {
notify_monitor(adapter, NEU_REQ_DEL_TAG_EVENT, cmd);
} else {
for (uint16_t i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i]);
}
free(cmd->tags);
}
free(cmd->tags);
neu_msg_exchange(header);
header->type = NEU_RESP_ERROR;
@@ -677,18 +684,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
}
if (resp.index) {
// we have added some tags, try to persist
adapter_storage_add_tags(cmd->driver, cmd->group, cmd->tags,
resp.index);
}
for (int i = 0; i < cmd->n_tag; i++) {
for (uint16_t i = resp.index; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
free(cmd->tags[i].description);
}
free(cmd->tags);
if (resp.index) {
// we have added some tags, try to persist
adapter_storage_add_tags(cmd->driver, cmd->group, cmd->tags,
resp.index);
cmd->n_tag = resp.index;
notify_monitor(adapter, NEU_REQ_ADD_TAG_EVENT, cmd);
} else {
free(cmd->tags);
}
neu_msg_exchange(header);
header->type = NEU_RESP_ADD_TAG;
@@ -717,12 +727,18 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
}
}
for (int i = 0; i < cmd->n_tag; i++) {
for (uint16_t i = resp.index; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
free(cmd->tags[i].description);
}
free(cmd->tags);
if (resp.index > 0) {
cmd->n_tag = resp.index;
notify_monitor(adapter, NEU_REQ_UPDATE_TAG_EVENT, cmd);
} else {
free(cmd->tags);
}
neu_msg_exchange(header);
header->type = NEU_RESP_UPDATE_TAG;
@@ -1148,6 +1164,7 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
data_size = sizeof(neu_resp_get_group_t);
break;
case NEU_REQ_ADD_TAG:
case NEU_REQ_ADD_TAG_EVENT:
data_size = sizeof(neu_req_add_tag_t);
break;
case NEU_RESP_ADD_TAG:
@@ -1157,9 +1174,11 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
data_size = sizeof(neu_resp_update_tag_t);
break;
case NEU_REQ_UPDATE_TAG:
case NEU_REQ_UPDATE_TAG_EVENT:
data_size = sizeof(neu_req_update_tag_t);
break;
case NEU_REQ_DEL_TAG:
case NEU_REQ_DEL_TAG_EVENT:
data_size = sizeof(neu_req_del_tag_t);
break;
case NEU_REQ_GET_TAG:
src/core/manager.c
+4 -1
View File
@@ -530,7 +530,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_NODE_CTL_EVENT:
case NEU_REQ_ADD_GROUP_EVENT:
case NEU_REQ_DEL_GROUP_EVENT:
case NEU_REQ_UPDATE_GROUP_EVENT: {
case NEU_REQ_UPDATE_GROUP_EVENT:
case NEU_REQ_ADD_TAG_EVENT:
case NEU_REQ_DEL_TAG_EVENT:
case NEU_REQ_UPDATE_TAG_EVENT: {
if (neu_node_manager_find(manager->node_manager, header->receiver)) {
forward_msg(manager, msg, header->receiver);
}
src/parser/neu_json_tag.c
+114
View File
@@ -29,6 +29,78 @@
#include "neu_json_tag.h"
int neu_json_encode_add_tags_req(void *json_object, void *param)
{
int ret = 0;
neu_json_add_tags_req_t *req = param;
void * tag_array = neu_json_array();
neu_json_add_tags_req_tag_t *p_tag = req->tags;
for (int i = 0; i < req->n_tag; i++) {
neu_json_elem_t tag_elems[] = {
{
.name = "type",
.t = NEU_JSON_INT,
.v.val_int = p_tag->type,
},
{
.name = "name",
.t = NEU_JSON_STR,
.v.val_str = p_tag->name,
},
{
.name = "attribute",
.t = NEU_JSON_INT,
.v.val_int = p_tag->attribute,
},
{
.name = "precision",
.t = NEU_JSON_INT,
.v.val_int = p_tag->precision,
},
{
.name = "decimal",
.t = NEU_JSON_DOUBLE,
.v.val_double = p_tag->decimal,
},
{
.name = "address",
.t = NEU_JSON_STR,
.v.val_str = p_tag->address,
},
{
.name = "description",
.t = NEU_JSON_STR,
.v.val_str = p_tag->description,
},
};
tag_array = neu_json_encode_array(tag_array, tag_elems,
NEU_JSON_ELEM_SIZE(tag_elems));
p_tag++;
}
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 = "tags",
.t = NEU_JSON_OBJECT,
.v.val_object = tag_array,
} };
ret = neu_json_encode_field(json_object, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
return ret;
}
int neu_json_decode_add_tags_req(char *buf, neu_json_add_tags_req_t **result)
{
int ret = 0;
@@ -176,6 +248,48 @@ int neu_json_encode_au_tags_resp(void *json_object, void *param)
return ret;
}
int neu_json_encode_del_tags_req(void *json_object, void *param)
{
int ret = 0;
neu_json_del_tags_req_t *req = param;
void * tag_array = neu_json_array();
neu_json_del_tags_req_name_t *p_name = req->tags;
for (int i = 0; i < req->n_tags; i++) {
neu_json_elem_t tag_elems[] = {
{
.name = NULL,
.t = NEU_JSON_STR,
.v.val_str = *p_name,
},
};
tag_array = neu_json_encode_array_value(tag_array, tag_elems,
NEU_JSON_ELEM_SIZE(tag_elems));
p_name++;
}
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 = "tags",
.t = NEU_JSON_OBJECT,
.v.val_object = tag_array,
} };
ret = neu_json_encode_field(json_object, req_elems,
NEU_JSON_ELEM_SIZE(req_elems));
return ret;
}
int neu_json_decode_del_tags_req(char *buf, neu_json_del_tags_req_t **result)
{
int ret = 0;
src/parser/neu_json_tag.h
+2
View File
@@ -48,6 +48,7 @@ typedef struct {
neu_json_add_tags_req_tag_t *tags;
} neu_json_add_tags_req_t;
int neu_json_encode_add_tags_req(void *json_object, void *param);
int neu_json_decode_add_tags_req(char *buf, neu_json_add_tags_req_t **result);
void neu_json_decode_add_tags_req_free(neu_json_add_tags_req_t *req);
@@ -67,6 +68,7 @@ typedef struct {
neu_json_del_tags_req_name_t *tags;
} neu_json_del_tags_req_t;
int neu_json_encode_del_tags_req(void *json_object, void *param);
int neu_json_decode_del_tags_req(char *buf, neu_json_del_tags_req_t **result);
void neu_json_decode_del_tags_req_free(neu_json_del_tags_req_t *req);