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

Merge pull request #1555 from eeff/v2.6

Browse Source 
Pick to v2.6 feat(monitor): move metrics api to restful plugin and remove singleton property
This commit is contained in:
fengzeroz
2023-09-22 17:54:25 +08:00
committed by GitHub
parent ce747b57d4 78cd5f5105
commit 041b7423c8
20 changed files with 266 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/check.yml
+15
View File
@@ -114,6 +114,7 @@ jobs:
- uses: actions/checkout@v2
with:
submodules: recursive
fetch-depth: 0
- name: compile
run: |
@@ -122,11 +123,25 @@ jobs:
cmake .. -DCMAKE_TOOLCHAIN_FILE=../cmake/x86_64-linux-gnu.cmake
make -j4
- name: get neuron current branch
shell: bash
run: |
if [ '${{ github.event_name }}' == 'push' ]; then
branch=${{ github.ref_name }}
elif [ '${{ github.event_name }}' == 'pull_request' ]; then
branch=${{ github.event.pull_request.base.ref }}
elif [ '${{ github.event_name }}' == 'release' ]; then
branch=${{ github.event.release.target_commitish }}
fi
echo "BRANCH=$branch" >> $GITHUB_ENV
echo "neuron current branch is $branch"
- name: fetch neu-ft
uses: actions/checkout@v2
with:
repository: neugates/neu-ft
path: neu-ft
ref: ${{ env.BRANCH }}
- name: function test
run: |
CMakeLists.txt
+1
View File
@@ -129,6 +129,7 @@ set(NEURON_SOURCES
plugins/restful/handle.c
plugins/restful/log_handle.c
plugins/restful/file_handle.c
plugins/restful/metric_handle.c
plugins/restful/normal_handle.c
plugins/restful/rw_handle.c
plugins/restful/adapter_handle.c
ft/metrics_test.robot
+13 -3
View File
@@ -5,8 +5,8 @@ Resource resource/api.resource
Resource resource/common.resource
Resource resource/error.resource
Suite Setup Start Neuronx
Suite Teardown Stop Neuronx
Suite Setup Metrics Test Setup
Suite Teardown Metrics Test Teardown
*** Test Cases ***
@@ -30,4 +30,14 @@ Check node state, it should return success
${nlines}= Get metrics lines
${parameter_lines}= Get parameter lines ${nlines}
Check node state ${parameter_lines}
Check node state ${parameter_lines}
*** Keywords ***
Metrics Test Setup
Start Neuronx
${res}= Add Node monitor ${PLUGIN_MONITOR}
Check Response Status ${res} 200
Metrics Test Teardown
Del Node monitor
Stop Neuronx
ft/node_test.robot
-12
View File
@@ -93,18 +93,6 @@ Update node name to the same, it should fail as name conflicts
Check Response Status ${res} 409
Check Error Code ${res} ${NEU_ERR_NODE_EXIST}
Update monitor node, it should fail
${res} = Update Node monitor monitor-new
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_NODE_NOT_ALLOW_UPDATE}
Update node name to monitor, it should fail
${res} = Update Node modbus-node monitor
Check Response Status ${res} 409
Check Error Code ${res} ${NEU_ERR_NODE_EXIST}
Update driver node name, it should success
${res} = Update Node modbus-node modbus-node-new
Check Response Status ${res} 200
ft/resource/common.resource
+1
View File
@@ -18,6 +18,7 @@ ${test_data_dir} ft/data
${PLUGIN_MODBUS_TCP} Modbus TCP
${PLUGIN_MONITOR} Monitor
${PLUGIN_MQTT} MQTT
${MQTT_CMD_NONE}
include/neuron/adapter.h
+67
View File
@@ -391,6 +391,30 @@ typedef struct {
neu_datatag_t *tags;
} neu_req_add_tag_t, neu_req_update_tag_t;
static inline void neu_req_add_tag_fini(neu_req_add_tag_t *req)
{
for (uint16_t i = 0; i < req->n_tag; i++) {
neu_tag_fini(&req->tags[i]);
}
free(req->tags);
}
static inline int neu_req_add_tag_copy(neu_req_add_tag_t *dst,
neu_req_add_tag_t *src)
{
strcpy(dst->driver, src->driver);
strcpy(dst->group, src->group);
dst->tags = (neu_datatag_t *) calloc(src->n_tag, sizeof(src->tags[0]));
if (NULL == dst->tags) {
return -1;
}
dst->n_tag = src->n_tag;
for (uint16_t i = 0; i < src->n_tag; i++) {
neu_tag_copy(&dst->tags[i], &src->tags[i]);
}
return 0;
}
typedef struct {
uint16_t index;
int error;
@@ -403,6 +427,36 @@ typedef struct neu_req_del_tag {
char ** tags;
} neu_req_del_tag_t;
static inline void neu_req_del_tag_fini(neu_req_del_tag_t *req)
{
for (uint16_t i = 0; i < req->n_tag; i++) {
free(req->tags[i]);
}
free(req->tags);
}
static inline int neu_req_del_tag_copy(neu_req_del_tag_t *dst,
neu_req_del_tag_t *src)
{
strcpy(dst->driver, src->driver);
strcpy(dst->group, src->group);
dst->tags = (char **) calloc(src->n_tag, sizeof(src->tags[0]));
if (NULL == dst->tags) {
return -1;
}
dst->n_tag = src->n_tag;
for (uint16_t i = 0; i < src->n_tag; ++i) {
dst->tags[i] = strdup(src->tags[i]);
if (NULL == dst->tags[i]) {
while (i-- > 0) {
free(dst->tags[i]);
}
free(dst->tags);
}
}
return 0;
}
typedef struct neu_req_get_tag {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
@@ -491,6 +545,19 @@ typedef struct neu_req_node_setting {
char *setting;
} neu_req_node_setting_t;
static inline void neu_req_node_setting_fini(neu_req_node_setting_t *req)
{
free(req->setting);
}
static inline int neu_req_node_setting_copy(neu_req_node_setting_t *dst,
neu_req_node_setting_t *src)
{
strcpy(dst->node, src->node);
dst->setting = strdup(src->setting);
return dst->setting ? 0 : -1;
}
typedef struct neu_req_get_node_setting {
char node[NEU_NODE_NAME_LEN];
} neu_req_get_node_setting_t;
plugins/monitor/CMakeLists.txt
-1
View File
@@ -6,7 +6,6 @@ file(COPY ${CMAKE_SOURCE_DIR}/plugins/monitor/monitor.json DESTINATION ${CMAKE_B
set(src
monitor.c
monitor_config.c
metric_handle.c
mqtt_handle.c
)
plugins/monitor/monitor.c
+1 -78
View File
@@ -23,7 +23,6 @@
#include "errcodes.h"
#include "event/event.h"
#include "metric_handle.h"
#include "monitor.h"
#include "mqtt_handle.h"
#include "utils/http_handler.h"
@@ -48,59 +47,6 @@ static void disconnect_cb(void *data)
neu_plugin_module.module_name);
}
static nng_http_server *server_init()
{
nng_url * url = NULL;
nng_http_server *server = NULL;
nlog_notice("monitor bind url: %s", host_port);
int ret = nng_url_parse(&url, host_port);
if (0 != ret) {
return NULL;
}
ret = nng_http_server_hold(&server, url);
if (0 != ret) {
nlog_error("monitor server bind error: %d", ret);
}
nng_url_free(url);
return server;
}
static struct neu_http_handler api_handlers[] = {
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/metrics",
.value.handler = handle_get_metric,
},
};
static int add_handlers(nng_http_server *server)
{
struct neu_http_handler cors = {
.method = NEU_HTTP_METHOD_OPTIONS,
.type = NEU_HTTP_HANDLER_FUNCTION,
.value.handler = neu_http_handle_cors,
};
for (size_t i = 0; i < sizeof(api_handlers) / sizeof(api_handlers[0]);
++i) {
if (0 != neu_http_add_handler(server, &api_handlers[0])) {
return -1;
}
cors.url = api_handlers[i].url;
if (0 != neu_http_add_handler(server, &cors)) {
return -1;
}
}
return 0;
}
static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
const monitor_config_t *config)
{
@@ -253,26 +199,6 @@ static neu_plugin_t *monitor_plugin_open(void)
neu_plugin_common_init(&plugin->common);
if (0 != pthread_mutex_init(&plugin->mutex, NULL)) {
free(plugin);
return NULL;
}
plugin->api_server = server_init();
if (NULL == plugin->api_server) {
nlog_error("fail to create monitor server");
free(plugin);
return NULL;
}
if (0 != add_handlers(plugin->api_server) ||
0 != nng_http_server_start(plugin->api_server)) {
nlog_error("fail to start monitor server");
nng_http_server_release(plugin->api_server);
free(plugin);
return NULL;
}
g_monitor_plugin_ = plugin;
return plugin;
}
@@ -281,9 +207,6 @@ static int monitor_plugin_close(neu_plugin_t *plugin)
{
int rv = 0;
nng_http_server_stop(plugin->api_server);
nng_http_server_release(plugin->api_server);
pthread_mutex_destroy(&plugin->mutex);
plog_notice(plugin, "Success to free plugin: %s",
neu_plugin_module.module_name);
free(plugin);
@@ -493,7 +416,7 @@ const neu_plugin_module_t neu_plugin_module = {
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_APP,
.display = true,
.single = true,
.single = false,
.single_name = "monitor",
};
plugins/monitor/monitor.h
-2
View File
@@ -33,11 +33,9 @@
struct neu_plugin {
neu_plugin_common_t common;
pthread_mutex_t mutex;
bool started;
neu_events_t * events;
neu_event_timer_t * heartbeat_timer;
nng_http_server * api_server;
monitor_config_t * config;
neu_mqtt_client_t * mqtt_client;
};
plugins/monitor/mqtt_handle.c
+3 -11
View File
@@ -322,20 +322,12 @@ 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);
neu_req_node_setting_fini(data);
} 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++) {
neu_tag_fini(&add_tag->tags[i]);
}
free(add_tag->tags);
neu_req_add_tag_fini(data);
} 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);
neu_req_del_tag_fini(data);
}
return rv;
}
plugins/restful/global_config_handle.c
+4 -1
View File
@@ -103,7 +103,10 @@ static int add_setting(context_t *ctx, neu_json_node_setting_req_t *data);
static inline bool is_static_node(const char *name, const char *plugin)
{
return 0 == strcmp("monitor", name) && 0 == strcmp("Monitor", plugin);
(void) name;
(void) plugin;
return false;
// return 0 == strcmp("monitor", name) && 0 == strcmp("Monitor", plugin);
}
static context_t *context_new(nng_aio *aio, bool get)
plugins/restful/handle.c
+10
View File
@@ -30,6 +30,7 @@
#include "global_config_handle.h"
#include "group_config_handle.h"
#include "log_handle.h"
#include "metric_handle.h"
#include "ndriver_handle.h"
#include "normal_handle.h"
#include "plugin_handle.h"
@@ -143,6 +144,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/ndriver/tag",
},
{
.url = "/api/v2/metrics",
},
};
static struct neu_http_handler rest_handlers[] = {
@@ -502,6 +506,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/ndriver/tag",
.value.handler = handle_get_ndriver_tags,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/metrics",
.value.handler = handle_get_metric,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
plugins/monitor/metric_handle.c → plugins/restful/metric_handle.c
+9 -11
View File
@@ -1,6 +1,6 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
* Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@@ -22,12 +22,11 @@
#include "define.h"
#include "metrics.h"
#include "utils/asprintf.h"
#include "utils/http.h"
#include "utils/http_handler.h"
#include "utils/log.h"
#include "metric_handle.h"
#include "monitor.h"
#include "utils/http.h"
#include "utils/http_handler.h"
// clang-format off
#define METRIC_GLOBAL_TMPL \
@@ -298,19 +297,18 @@ static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx)
void handle_get_metric(nng_aio *aio)
{
int status = NNG_HTTP_STATUS_OK;
char * result = NULL;
size_t len = 0;
FILE * stream = NULL;
neu_plugin_t *plugin = neu_monitor_get_plugin();
int status = NNG_HTTP_STATUS_OK;
char * result = NULL;
size_t len = 0;
FILE * stream = NULL;
neu_metrics_category_e cat = NEU_METRICS_CATEGORY_ALL;
size_t cat_param_len = 0;
const char *cat_param = neu_http_get_param(aio, "category", &cat_param_len);
if (NULL != cat_param &&
!parse_metrics_catgory(cat_param, cat_param_len, &cat)) {
plog_error(plugin, "invalid metrics category: %.*s",
(int) cat_param_len, cat_param);
nlog_error("invalid metrics category: %.*s", (int) cat_param_len,
cat_param);
status = NNG_HTTP_STATUS_BAD_REQUEST;
goto end;
}
plugins/monitor/metric_handle.h → plugins/restful/metric_handle.h
+3 -3
View File
@@ -1,6 +1,6 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
* Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@@ -17,8 +17,8 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef NEU_PLUGIN_MONITOR_METRIC_H
#define NEU_PLUGIN_MONITOR_METRIC_H
#ifndef NEU_PLUGIN_REST_METRIC_H
#define NEU_PLUGIN_REST_METRIC_H
#include <nng/nng.h>
src/adapter/adapter.c
+1 -1
View File
@@ -1347,7 +1347,7 @@ inline static void notify_monitor(neu_adapter_t * adapter,
neu_reqresp_type_e event, void *data)
{
neu_reqresp_head_t header = {
.receiver = "monitor",
.receiver = "manager",
.type = event,
};
src/core/manager.c
+83 -27
View File
@@ -58,8 +58,8 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
nng_pipe pipe, msg_dup dup);
inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
const char *node);
inline static void notify_monitor(neu_manager_t * manager,
neu_reqresp_type_e event, void *data);
inline static void notify_monitors(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,
@@ -588,7 +588,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);
notify_monitors(manager, NEU_REQ_ADD_NODE_EVENT, cmd);
}
header->type = NEU_RESP_ERROR;
@@ -600,12 +600,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_req_update_node_t *cmd = (neu_req_update_node_t *) &header[1];
neu_resp_error_t e = { 0 };
if (0 == strcmp("monitor", cmd->node)) {
e.error = NEU_ERR_NODE_NOT_ALLOW_UPDATE;
} else if (0 == strcmp("monitor", cmd->new_name)) {
e.error = NEU_ERR_NODE_EXIST;
} else if (NULL ==
neu_node_manager_find(manager->node_manager, cmd->node)) {
if (NULL == neu_node_manager_find(manager->node_manager, cmd->node)) {
e.error = NEU_ERR_NODE_NOT_EXIST;
} else if (NULL !=
neu_node_manager_find(manager->node_manager,
@@ -657,7 +652,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);
notify_monitors(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);
@@ -936,16 +931,14 @@ 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_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: {
if (neu_node_manager_find(manager->node_manager, header->receiver)) {
forward_msg(manager, msg, header->receiver);
}
notify_monitors(manager, header->type, &header[1]);
break;
}
@@ -1027,8 +1020,7 @@ 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_EVENT: {
case NEU_REQ_NODE_SETTING: {
neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1];
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
@@ -1225,30 +1217,94 @@ 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)
struct notify_monitor_ctx {
neu_manager_t * manager;
neu_reqresp_type_e event;
void * data;
};
static int notify_monitor(const char *name, nng_pipe pipe, void *arg)
{
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, "monitor");
struct notify_monitor_ctx *ctx = arg;
void * data = NULL;
neu_req_node_setting_t setting = { 0 };
neu_req_add_tag_t mod_tag = { 0 };
neu_req_del_tag_t del_tag = { 0 };
// monitor not ready, ignore
if (0 == pipe.id) {
nlog_error("no monitor node");
return;
return 0;
}
neu_reqresp_head_t header = {
.sender = "manager",
.receiver = "monitor",
.type = event,
};
// copy message data
switch (ctx->event) {
case NEU_REQ_NODE_SETTING_EVENT: {
if (0 != neu_req_node_setting_copy(&setting, ctx->data)) {
return -1;
}
data = &setting;
break;
}
case NEU_REQ_ADD_TAG_EVENT:
case NEU_REQ_UPDATE_TAG_EVENT: {
if (0 != neu_req_add_tag_copy(&mod_tag, ctx->data)) {
return -1;
}
data = &mod_tag;
break;
}
case NEU_REQ_DEL_TAG_EVENT: {
if (0 != neu_req_del_tag_copy(&del_tag, ctx->data)) {
return -1;
}
data = &del_tag;
break;
}
default:
data = ctx->data;
}
// message header
neu_reqresp_head_t header = {
.sender = "manager",
.type = ctx->event,
};
strcpy(header.receiver, name);
// send message
nng_msg *msg = neu_msg_gen(&header, data);
nng_msg_set_pipe(msg, pipe);
int ret = nng_sendmsg(manager->socket, msg, 0);
int ret = nng_sendmsg(ctx->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));
}
return 0;
}
inline static void notify_monitors(neu_manager_t * manager,
neu_reqresp_type_e event, void *data)
{
struct notify_monitor_ctx ctx = {
.manager = manager,
.event = event,
.data = data,
};
neu_node_manager_for_each_monitor(manager->node_manager, notify_monitor,
&ctx);
if (NEU_REQ_NODE_SETTING_EVENT == event) {
neu_req_node_setting_fini(data);
} else if (NEU_REQ_ADD_TAG_EVENT == event ||
NEU_REQ_UPDATE_TAG_EVENT == event) {
neu_req_add_tag_fini(data);
} else if (NEU_REQ_DEL_TAG_EVENT == event) {
neu_req_del_tag_fini(data);
}
}
static void start_static_adapter(neu_manager_t *manager, const char *name)
src/core/manager_internal.c
+1 -1
View File
@@ -684,7 +684,7 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group,
const char *params)
{
if (0 == strcmp(app, "monitor")) {
if (neu_node_manager_is_monitor(manager->node_manager, app)) {
// filter out monitor node
return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE;
}
src/core/node_manager.c
+42
View File
@@ -32,6 +32,7 @@ typedef struct node_entity {
bool is_static;
bool display;
bool single;
bool is_monitor;
nng_pipe pipe;
UT_hash_handle hh;
@@ -40,12 +41,17 @@ typedef struct node_entity {
struct neu_node_manager {
node_entity_t *nodes;
UT_array * monitors;
};
neu_node_manager_t *neu_node_manager_create()
{
neu_node_manager_t *node_manager = calloc(1, sizeof(neu_node_manager_t));
if (node_manager) {
utarray_new(node_manager->monitors, &ut_ptr_icd);
}
return node_manager;
}
@@ -60,6 +66,7 @@ void neu_node_manager_destroy(neu_node_manager_t *mgr)
free(el);
}
utarray_free(mgr->monitors);
free(mgr);
}
@@ -70,9 +77,15 @@ int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter)
node->adapter = adapter;
node->name = strdup(adapter->name);
node->display = true;
node->is_monitor =
(0 == strcmp(node->adapter->module->module_name, "Monitor"));
HASH_ADD_STR(mgr->nodes, name, node);
if (node->is_monitor) {
utarray_push_back(mgr->monitors, &node);
}
return 0;
}
@@ -149,6 +162,10 @@ void neu_node_manager_del(neu_node_manager_t *mgr, const char *name)
HASH_FIND_STR(mgr->nodes, name, node);
if (node != NULL) {
HASH_DEL(mgr->nodes, node);
if (node->is_monitor) {
utarray_erase(mgr->monitors, utarray_eltidx(mgr->monitors, node),
1);
}
free(node->name);
free(node);
}
@@ -354,6 +371,31 @@ nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name)
return pipe;
}
bool neu_node_manager_is_monitor(neu_node_manager_t *mgr, const char *name)
{
node_entity_t *node = NULL;
HASH_FIND_STR(mgr->nodes, name, node);
return node ? node->is_monitor : false;
}
int neu_node_manager_for_each_monitor(neu_node_manager_t *mgr,
int (*cb)(const char *name, nng_pipe pipe,
void *data),
void *data)
{
int rv = 0;
utarray_foreach(mgr->monitors, node_entity_t **, node)
{
rv = cb((*node)->name, (*node)->pipe, data);
if (0 != rv) {
break;
}
}
return rv;
}
UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr)
{
UT_icd icd = { sizeof(neu_nodes_state_t), NULL, NULL, NULL };
src/core/node_manager.h
+6
View File
@@ -62,6 +62,12 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, int type);
UT_array *neu_node_manager_get_pipes_all(neu_node_manager_t *mgr);
nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name);
bool neu_node_manager_is_monitor(neu_node_manager_t *mgr, const char *name);
int neu_node_manager_for_each_monitor(neu_node_manager_t *mgr,
int (*)(const char *name, nng_pipe pipe,
void *data),
void *data);
// neu_nodes_state_t array
UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr);
src/persist/persist.c
+6 -15
View File
@@ -600,21 +600,12 @@ void neu_persister_destroy()
int neu_persister_store_node(neu_persist_node_info_t *info)
{
int rv = 0;
if (strcmp(info->plugin_name, "Monitor") == 0) {
rv = execute_sql(
global_db,
"INSERT OR IGNORE INTO nodes (name, type, state, plugin_name) "
"VALUES (%Q, %i, %i, %Q)",
info->name, info->type, info->state, info->plugin_name);
} else {
rv =
execute_sql(global_db,
"INSERT INTO nodes (name, type, state, plugin_name) "
"VALUES (%Q, %i, %i, %Q)",
info->name, info->type, info->state, info->plugin_name);
if (0 == rv) {
node_count_add(1);
}
rv = execute_sql(global_db,
"INSERT INTO nodes (name, type, state, plugin_name) "
"VALUES (%Q, %i, %i, %Q)",
info->name, info->type, info->state, info->plugin_name);
if (0 == rv) {
node_count_add(1);
}
return rv;
}