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

Merge pull request #1035 from eeff/eeff/dev

Browse Source 
feat(monitor): handle node events
This commit is contained in:
fengzeroz
2023-01-17 14:45:55 +08:00
committed by GitHub
parent 899b080171 d358dca495
commit 257341bca1
14 changed files with 349 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+1
View File
@@ -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;
include/neuron/json/json.h
+2
View File
@@ -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
}
plugins/monitor/CMakeLists.txt
+1
View File
@@ -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)
plugins/monitor/monitor.c
+4 -1
View File
@@ -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:
plugins/monitor/monitor_config.c
+60
View File
@@ -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));
}
plugins/monitor/monitor_config.h
+6
View File
@@ -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,
plugins/monitor/mqtt_handle.c
+93
View File
@@ -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;
}
plugins/monitor/mqtt_handle.h
+2
View File
@@ -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
plugins/restful/adapter_handle.c
+21 -6
View File
@@ -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)
src/adapter/adapter.c
+1
View File
@@ -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);
}
src/core/manager.c
+3 -3
View File
@@ -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) ==
src/parser/neu_json_node.c
+111
View File
@@ -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)
{
@@ -356,12 +438,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 +464,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;
}
src/parser/neu_json_node.h
+12
View File
@@ -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,17 +96,27 @@ 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);
typedef struct {
char *node;
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);
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
src/utils/json.c
+32
View File
@@ -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;
}