From e201ba6a3e3e2066e6615d78fca5e4560d7bf271 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 18 Sep 2023 10:29:51 +0100 Subject: [PATCH 1/5] feat(restful): add metrics api --- CMakeLists.txt | 1 + plugins/restful/handle.c | 10 + plugins/restful/metric_handle.c | 362 ++++++++++++++++++++++++++++++++ plugins/restful/metric_handle.h | 29 +++ 4 files changed, 402 insertions(+) create mode 100644 plugins/restful/metric_handle.c create mode 100644 plugins/restful/metric_handle.h diff --git a/CMakeLists.txt b/CMakeLists.txt index d545ff00..c7ad71c4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -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 diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 1a37797a..af0aecfa 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -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) diff --git a/plugins/restful/metric_handle.c b/plugins/restful/metric_handle.c new file mode 100644 index 00000000..575b8196 --- /dev/null +++ b/plugins/restful/metric_handle.c @@ -0,0 +1,362 @@ +/** + * NEURON IIoT System for Industry 4.0 + * 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 + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#include + +#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" + +// clang-format off +#define METRIC_GLOBAL_TMPL \ + "# HELP os_info OS distro and kernel version\n" \ + "# TYPE os_info gauge\n" \ + "os_info{version=\"%s\"} 0\n" \ + "os_info{kernel=\"%s\"} 0\n" \ + "# HELP cpu_percent Total CPU utilisation percentage\n" \ + "# TYPE cpu_percent gauge\n" \ + "cpu_percent %u\n" \ + "# HELP cpu_cores Number of CPU cores\n" \ + "# TYPE cpu_cores counter\n" \ + "cpu_cores %u\n" \ + "# HELP mem_total_bytes Total installed memory in bytes\n" \ + "# TYPE mem_total_bytes counter\n" \ + "mem_total_bytes %zu\n" \ + "# HELP mem_used_bytes Used memory in bytes\n" \ + "# TYPE mem_used_bytes gauge\n" \ + "mem_used_bytes %zu\n" \ + "# HELP mem_cache Memory buffer/cache size in bytes\n" \ + "# TYPE mem_cache gauge\n" \ + "mem_cache_bytes %zu\n" \ + "# HELP rss_bytes RSS (Resident Set Size) in bytes\n" \ + "# TYPE rss_bytes gauge\n" \ + "rss_bytes %zu\n" \ + "# HELP disk_size_gibibytes Disk size in gibibytes\n" \ + "# TYPE disk_size_gibibytes counter\n" \ + "disk_size_gibibytes %zu\n" \ + "# HELP disk_used_gibibytes Used disk size in gibibytes\n" \ + "# TYPE disk_used_gibibytes gauge\n" \ + "disk_used_gibibytes %zu\n" \ + "# HELP disk_avail_gibibytes Available disk size in gibibytes\n" \ + "# TYPE disk_avail_gibibytes gauge\n" \ + "disk_avail_gibibytes %zu\n" \ + "# HELP core_dumped Whether there is any core dump\n" \ + "# TYPE core_dumped gauge\n" \ + "core_dumped %d\n" \ + "# HELP uptime_seconds Uptime in seconds\n" \ + "# TYPE uptime_seconds counter\n" \ + "uptime_seconds %" PRIu64 "\n" \ + "# HELP north_nodes_total Number of north nodes\n" \ + "# TYPE north_nodes_total gauge\n" \ + "north_nodes_total %zu\n" \ + "# HELP north_running_nodes_total Number of north nodes in running state\n" \ + "# TYPE north_running_nodes_total gauge\n" \ + "north_running_nodes_total %zu\n" \ + "# HELP north_disconnected_nodes_total Number of north nodes disconnected\n" \ + "# TYPE north_disconnected_nodes_total gauge\n" \ + "north_disconnected_nodes_total %zu\n" \ + "# HELP south_nodes_total Number of south nodes\n" \ + "# TYPE south_nodes_total gauge\n" \ + "south_nodes_total %zu\n" \ + "# HELP south_running_nodes_total Number of south nodes in running state\n" \ + "# TYPE south_running_nodes_total gauge\n" \ + "south_running_nodes_total %zu\n" \ + "# HELP south_disconnected_nodes_total Number of south nodes disconnected\n" \ + "# TYPE south_disconnected_nodes_total gauge\n" \ + "south_disconnected_nodes_total %zu\n" +// clang-format on + +static int response(nng_aio *aio, char *content, enum nng_http_status status) +{ + nng_http_res *res = NULL; + + nng_http_res_alloc(&res); + + nng_http_res_set_header(res, "Content-Type", "text/plain"); + nng_http_res_set_header(res, "Access-Control-Allow-Origin", "*"); + nng_http_res_set_header(res, "Access-Control-Allow-Methods", + "POST,GET,PUT,DELETE,OPTIONS"); + nng_http_res_set_header(res, "Access-Control-Allow-Headers", "*"); + + if (content != NULL && strlen(content) > 0) { + nng_http_res_copy_data(res, content, strlen(content)); + } else { + nng_http_res_set_data(res, NULL, 0); + } + + nng_http_res_set_status(res, status); + + nng_http_req *nng_req = nng_aio_get_input(aio, 0); + nlog_info("%s %s [%d]", nng_http_req_get_method(nng_req), + nng_http_req_get_uri(nng_req), status); + + nng_aio_set_output(aio, 0, res); + nng_aio_finish(aio, 0); + + return 0; +} + +static inline bool parse_metrics_catgory(const char *s, size_t len, + neu_metrics_category_e *cat) +{ + char *domain[] = { + [NEU_METRICS_CATEGORY_GLOBAL] = "global", + [NEU_METRICS_CATEGORY_DRIVER] = "driver", + [NEU_METRICS_CATEGORY_APP] = "app", + [NEU_METRICS_CATEGORY_ALL] = NULL, + }; + + if (NULL == s || NULL == cat) { + return false; + } + + for (int i = 0; i < NEU_METRICS_CATEGORY_ALL; ++i) { + if (strlen(domain[i]) == len && 0 == strncmp(s, domain[i], len)) { + *cat = i; + return true; + } + } + + return false; +} + +static inline void gen_global_metrics(const neu_metrics_t *metrics, + FILE * stream) +{ + fprintf(stream, METRIC_GLOBAL_TMPL, metrics->distro, metrics->kernel, + metrics->cpu_percent, metrics->cpu_cores, metrics->mem_total_bytes, + metrics->mem_used_bytes, metrics->mem_cache_bytes, + metrics->mem_used_bytes, metrics->disk_size_gibibytes, + metrics->disk_used_gibibytes, metrics->disk_avail_gibibytes, + metrics->core_dumped, metrics->uptime_seconds, metrics->north_nodes, + metrics->north_running_nodes, metrics->north_disconnected_nodes, + metrics->south_nodes, metrics->south_running_nodes, + metrics->south_disconnected_nodes); +} + +static inline void +gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream) +{ + fprintf(stream, + "# HELP node_type Driver(1) or APP(2)\n" + "# TYPE node_type gauge\n" + "node_type{node=\"%s\"} %d\n", + node_metrics->name, node_metrics->type); + + neu_metric_entry_t *e = NULL; + + HASH_LOOP(hh, node_metrics->entries, e) + { + fprintf(stream, + "# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\"} %" PRIu64 "\n", + e->name, e->help, e->name, neu_metric_type_str(e->type), + e->name, node_metrics->name, e->value); + } + + neu_group_metrics_t *g = NULL; + HASH_LOOP(hh, node_metrics->group_metrics, g) + { + HASH_LOOP(hh, g->entries, e) + { + fprintf(stream, + "# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\",group=\"%s\"} " + "%" PRIu64 "\n", + e->name, e->help, e->name, neu_metric_type_str(e->type), + e->name, node_metrics->name, g->name, e->value); + } + } +} + +static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name) +{ + neu_metric_entry_t * e = NULL; + neu_group_metrics_t *g = NULL; + HASH_FIND_STR(node_metrics->entries, name, e); + if (e) { + return true; + } + + HASH_LOOP(hh, node_metrics->group_metrics, g) + { + HASH_FIND_STR(g->entries, name, e); + if (e) { + return true; + } + } + return false; +} + +static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter, + FILE *stream) +{ + neu_metric_entry_t * e = NULL, *r = NULL; + neu_group_metrics_t *g = NULL; + neu_node_metrics_t * n = NULL; + + bool commented = false; + HASH_LOOP(hh, metrics->node_metrics, n) + { + if (!(type_filter & n->type)) { + continue; + } + + if (!commented) { + commented = true; + fprintf(stream, + "# HELP node_type Driver(1) or APP(2)\n" + "# TYPE node_type gauge\n"); + } + + fprintf(stream, "node_type{node=\"%s\"} %d\n", n->name, n->type); + } + + n = NULL; + + HASH_LOOP(hh, metrics->registered_metrics, r) + { + commented = false; + HASH_LOOP(hh, metrics->node_metrics, n) + { + if (!(type_filter & n->type)) { + continue; + } + + if (!has_entry(n, r->name)) { + continue; + } + + if (!commented) { + commented = true; + fprintf(stream, "# HELP %s %s\n# TYPE %s %s\n", r->name, + r->help, r->name, neu_metric_type_str(r->type)); + } + + HASH_FIND_STR(n->entries, r->name, e); + if (e) { + fprintf(stream, "%s{node=\"%s\"} %" PRIu64 "\n", e->name, + n->name, e->value); + + continue; + } + + HASH_LOOP(hh, n->group_metrics, g) + { + HASH_FIND_STR(g->entries, r->name, e); + if (e) { + fprintf(stream, + "%s{node=\"%s\",group=\"%s\"} %" PRIu64 "\n", + e->name, n->name, g->name, e->value); + } + } + } + } +} + +struct context { + int filter; + int * status; + FILE * stream; + const char *node; +}; + +static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx) +{ + if (ctx->node[0]) { + neu_node_metrics_t *n = NULL; + HASH_FIND_STR(metrics->node_metrics, ctx->node, n); + if (NULL == n || 0 == (ctx->filter & n->type)) { + *ctx->status = NNG_HTTP_STATUS_NOT_FOUND; + return; + } + gen_single_node_metrics(n, ctx->stream); + } else { + gen_all_node_metrics(metrics, ctx->filter, ctx->stream); + } +} + +void handle_get_metric(nng_aio *aio) +{ + 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)) { + nlog_error("invalid metrics category: %.*s", (int) cat_param_len, + cat_param); + status = NNG_HTTP_STATUS_BAD_REQUEST; + goto end; + } + + char node_name[NEU_NODE_NAME_LEN] = { 0 }; + ssize_t rv = + neu_http_get_param_str(aio, "node", node_name, sizeof(node_name)); + if (-1 == rv || rv >= NEU_NODE_NAME_LEN || + (0 < rv && NEU_METRICS_CATEGORY_GLOBAL == cat)) { + status = NNG_HTTP_STATUS_BAD_REQUEST; + goto end; + } + + stream = open_memstream(&result, &len); + if (NULL == stream) { + status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR; + goto end; + } + + struct context ctx = { + .status = &status, + .stream = stream, + .node = node_name, + }; + + switch (cat) { + case NEU_METRICS_CATEGORY_GLOBAL: + neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream); + break; + case NEU_METRICS_CATEGORY_DRIVER: + ctx.filter = NEU_NA_TYPE_DRIVER; + neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); + break; + case NEU_METRICS_CATEGORY_APP: + ctx.filter = NEU_NA_TYPE_APP | NEU_NA_TYPE_NDRIVER; + neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); + break; + case NEU_METRICS_CATEGORY_ALL: + ctx.filter = NEU_NA_TYPE_DRIVER | NEU_NA_TYPE_NDRIVER | NEU_NA_TYPE_APP; + neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream); + neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); + break; + } + +end: + if (NULL != stream) { + fclose(stream); + } + response(aio, NNG_HTTP_STATUS_OK == status ? result : NULL, status); + free(result); +} diff --git a/plugins/restful/metric_handle.h b/plugins/restful/metric_handle.h new file mode 100644 index 00000000..ebd92a2f --- /dev/null +++ b/plugins/restful/metric_handle.h @@ -0,0 +1,29 @@ +/** + * NEURON IIoT System for Industry 4.0 + * 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 + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEU_PLUGIN_REST_METRIC_H +#define NEU_PLUGIN_REST_METRIC_H + +#include + +#include "json/neu_json_fn.h" + +void handle_get_metric(nng_aio *aio); + +#endif From 812cc72a6cf870365b5b105a0850c63a3fb09e7d Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 18 Sep 2023 10:23:32 +0800 Subject: [PATCH 2/5] feat(monitor): remove metrics http api --- plugins/monitor/CMakeLists.txt | 1 - plugins/monitor/metric_handle.c | 364 -------------------------------- plugins/monitor/metric_handle.h | 29 --- plugins/monitor/monitor.c | 77 ------- plugins/monitor/monitor.h | 2 - 5 files changed, 473 deletions(-) delete mode 100644 plugins/monitor/metric_handle.c delete mode 100644 plugins/monitor/metric_handle.h diff --git a/plugins/monitor/CMakeLists.txt b/plugins/monitor/CMakeLists.txt index 10bc1e4e..47691e71 100644 --- a/plugins/monitor/CMakeLists.txt +++ b/plugins/monitor/CMakeLists.txt @@ -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 ) diff --git a/plugins/monitor/metric_handle.c b/plugins/monitor/metric_handle.c deleted file mode 100644 index 7c43c889..00000000 --- a/plugins/monitor/metric_handle.c +++ /dev/null @@ -1,364 +0,0 @@ -/** - * NEURON IIoT System for Industry 4.0 - * Copyright (C) 2020-2022 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 - * License as published by the Free Software Foundation; either - * version 3 of the License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public License - * along with this program; if not, write to the Free Software Foundation, - * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - **/ - -#include - -#include "define.h" -#include "metrics.h" -#include "utils/asprintf.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 \ - "# HELP os_info OS distro and kernel version\n" \ - "# TYPE os_info gauge\n" \ - "os_info{version=\"%s\"} 0\n" \ - "os_info{kernel=\"%s\"} 0\n" \ - "# HELP cpu_percent Total CPU utilisation percentage\n" \ - "# TYPE cpu_percent gauge\n" \ - "cpu_percent %u\n" \ - "# HELP cpu_cores Number of CPU cores\n" \ - "# TYPE cpu_cores counter\n" \ - "cpu_cores %u\n" \ - "# HELP mem_total_bytes Total installed memory in bytes\n" \ - "# TYPE mem_total_bytes counter\n" \ - "mem_total_bytes %zu\n" \ - "# HELP mem_used_bytes Used memory in bytes\n" \ - "# TYPE mem_used_bytes gauge\n" \ - "mem_used_bytes %zu\n" \ - "# HELP mem_cache Memory buffer/cache size in bytes\n" \ - "# TYPE mem_cache gauge\n" \ - "mem_cache_bytes %zu\n" \ - "# HELP rss_bytes RSS (Resident Set Size) in bytes\n" \ - "# TYPE rss_bytes gauge\n" \ - "rss_bytes %zu\n" \ - "# HELP disk_size_gibibytes Disk size in gibibytes\n" \ - "# TYPE disk_size_gibibytes counter\n" \ - "disk_size_gibibytes %zu\n" \ - "# HELP disk_used_gibibytes Used disk size in gibibytes\n" \ - "# TYPE disk_used_gibibytes gauge\n" \ - "disk_used_gibibytes %zu\n" \ - "# HELP disk_avail_gibibytes Available disk size in gibibytes\n" \ - "# TYPE disk_avail_gibibytes gauge\n" \ - "disk_avail_gibibytes %zu\n" \ - "# HELP core_dumped Whether there is any core dump\n" \ - "# TYPE core_dumped gauge\n" \ - "core_dumped %d\n" \ - "# HELP uptime_seconds Uptime in seconds\n" \ - "# TYPE uptime_seconds counter\n" \ - "uptime_seconds %" PRIu64 "\n" \ - "# HELP north_nodes_total Number of north nodes\n" \ - "# TYPE north_nodes_total gauge\n" \ - "north_nodes_total %zu\n" \ - "# HELP north_running_nodes_total Number of north nodes in running state\n" \ - "# TYPE north_running_nodes_total gauge\n" \ - "north_running_nodes_total %zu\n" \ - "# HELP north_disconnected_nodes_total Number of north nodes disconnected\n" \ - "# TYPE north_disconnected_nodes_total gauge\n" \ - "north_disconnected_nodes_total %zu\n" \ - "# HELP south_nodes_total Number of south nodes\n" \ - "# TYPE south_nodes_total gauge\n" \ - "south_nodes_total %zu\n" \ - "# HELP south_running_nodes_total Number of south nodes in running state\n" \ - "# TYPE south_running_nodes_total gauge\n" \ - "south_running_nodes_total %zu\n" \ - "# HELP south_disconnected_nodes_total Number of south nodes disconnected\n" \ - "# TYPE south_disconnected_nodes_total gauge\n" \ - "south_disconnected_nodes_total %zu\n" -// clang-format on - -static int response(nng_aio *aio, char *content, enum nng_http_status status) -{ - nng_http_res *res = NULL; - - nng_http_res_alloc(&res); - - nng_http_res_set_header(res, "Content-Type", "text/plain"); - nng_http_res_set_header(res, "Access-Control-Allow-Origin", "*"); - nng_http_res_set_header(res, "Access-Control-Allow-Methods", - "POST,GET,PUT,DELETE,OPTIONS"); - nng_http_res_set_header(res, "Access-Control-Allow-Headers", "*"); - - if (content != NULL && strlen(content) > 0) { - nng_http_res_copy_data(res, content, strlen(content)); - } else { - nng_http_res_set_data(res, NULL, 0); - } - - nng_http_res_set_status(res, status); - - nng_http_req *nng_req = nng_aio_get_input(aio, 0); - nlog_info("%s %s [%d]", nng_http_req_get_method(nng_req), - nng_http_req_get_uri(nng_req), status); - - nng_aio_set_output(aio, 0, res); - nng_aio_finish(aio, 0); - - return 0; -} - -static inline bool parse_metrics_catgory(const char *s, size_t len, - neu_metrics_category_e *cat) -{ - char *domain[] = { - [NEU_METRICS_CATEGORY_GLOBAL] = "global", - [NEU_METRICS_CATEGORY_DRIVER] = "driver", - [NEU_METRICS_CATEGORY_APP] = "app", - [NEU_METRICS_CATEGORY_ALL] = NULL, - }; - - if (NULL == s || NULL == cat) { - return false; - } - - for (int i = 0; i < NEU_METRICS_CATEGORY_ALL; ++i) { - if (strlen(domain[i]) == len && 0 == strncmp(s, domain[i], len)) { - *cat = i; - return true; - } - } - - return false; -} - -static inline void gen_global_metrics(const neu_metrics_t *metrics, - FILE * stream) -{ - fprintf(stream, METRIC_GLOBAL_TMPL, metrics->distro, metrics->kernel, - metrics->cpu_percent, metrics->cpu_cores, metrics->mem_total_bytes, - metrics->mem_used_bytes, metrics->mem_cache_bytes, - metrics->mem_used_bytes, metrics->disk_size_gibibytes, - metrics->disk_used_gibibytes, metrics->disk_avail_gibibytes, - metrics->core_dumped, metrics->uptime_seconds, metrics->north_nodes, - metrics->north_running_nodes, metrics->north_disconnected_nodes, - metrics->south_nodes, metrics->south_running_nodes, - metrics->south_disconnected_nodes); -} - -static inline void -gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream) -{ - fprintf(stream, - "# HELP node_type Driver(1) or APP(2)\n" - "# TYPE node_type gauge\n" - "node_type{node=\"%s\"} %d\n", - node_metrics->name, node_metrics->type); - - neu_metric_entry_t *e = NULL; - - HASH_LOOP(hh, node_metrics->entries, e) - { - fprintf(stream, - "# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\"} %" PRIu64 "\n", - e->name, e->help, e->name, neu_metric_type_str(e->type), - e->name, node_metrics->name, e->value); - } - - neu_group_metrics_t *g = NULL; - HASH_LOOP(hh, node_metrics->group_metrics, g) - { - HASH_LOOP(hh, g->entries, e) - { - fprintf(stream, - "# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\",group=\"%s\"} " - "%" PRIu64 "\n", - e->name, e->help, e->name, neu_metric_type_str(e->type), - e->name, node_metrics->name, g->name, e->value); - } - } -} - -static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name) -{ - neu_metric_entry_t * e = NULL; - neu_group_metrics_t *g = NULL; - HASH_FIND_STR(node_metrics->entries, name, e); - if (e) { - return true; - } - - HASH_LOOP(hh, node_metrics->group_metrics, g) - { - HASH_FIND_STR(g->entries, name, e); - if (e) { - return true; - } - } - return false; -} - -static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter, - FILE *stream) -{ - neu_metric_entry_t * e = NULL, *r = NULL; - neu_group_metrics_t *g = NULL; - neu_node_metrics_t * n = NULL; - - bool commented = false; - HASH_LOOP(hh, metrics->node_metrics, n) - { - if (!(type_filter & n->type)) { - continue; - } - - if (!commented) { - commented = true; - fprintf(stream, - "# HELP node_type Driver(1) or APP(2)\n" - "# TYPE node_type gauge\n"); - } - - fprintf(stream, "node_type{node=\"%s\"} %d\n", n->name, n->type); - } - - n = NULL; - - HASH_LOOP(hh, metrics->registered_metrics, r) - { - commented = false; - HASH_LOOP(hh, metrics->node_metrics, n) - { - if (!(type_filter & n->type)) { - continue; - } - - if (!has_entry(n, r->name)) { - continue; - } - - if (!commented) { - commented = true; - fprintf(stream, "# HELP %s %s\n# TYPE %s %s\n", r->name, - r->help, r->name, neu_metric_type_str(r->type)); - } - - HASH_FIND_STR(n->entries, r->name, e); - if (e) { - fprintf(stream, "%s{node=\"%s\"} %" PRIu64 "\n", e->name, - n->name, e->value); - - continue; - } - - HASH_LOOP(hh, n->group_metrics, g) - { - HASH_FIND_STR(g->entries, r->name, e); - if (e) { - fprintf(stream, - "%s{node=\"%s\",group=\"%s\"} %" PRIu64 "\n", - e->name, n->name, g->name, e->value); - } - } - } - } -} - -struct context { - int filter; - int * status; - FILE * stream; - const char *node; -}; - -static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx) -{ - if (ctx->node[0]) { - neu_node_metrics_t *n = NULL; - HASH_FIND_STR(metrics->node_metrics, ctx->node, n); - if (NULL == n || 0 == (ctx->filter & n->type)) { - *ctx->status = NNG_HTTP_STATUS_NOT_FOUND; - return; - } - gen_single_node_metrics(n, ctx->stream); - } else { - gen_all_node_metrics(metrics, ctx->filter, ctx->stream); - } -} - -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(); - - 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); - status = NNG_HTTP_STATUS_BAD_REQUEST; - goto end; - } - - char node_name[NEU_NODE_NAME_LEN] = { 0 }; - ssize_t rv = - neu_http_get_param_str(aio, "node", node_name, sizeof(node_name)); - if (-1 == rv || rv >= NEU_NODE_NAME_LEN || - (0 < rv && NEU_METRICS_CATEGORY_GLOBAL == cat)) { - status = NNG_HTTP_STATUS_BAD_REQUEST; - goto end; - } - - stream = open_memstream(&result, &len); - if (NULL == stream) { - status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR; - goto end; - } - - struct context ctx = { - .status = &status, - .stream = stream, - .node = node_name, - }; - - switch (cat) { - case NEU_METRICS_CATEGORY_GLOBAL: - neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream); - break; - case NEU_METRICS_CATEGORY_DRIVER: - ctx.filter = NEU_NA_TYPE_DRIVER; - neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); - break; - case NEU_METRICS_CATEGORY_APP: - ctx.filter = NEU_NA_TYPE_APP | NEU_NA_TYPE_NDRIVER; - neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); - break; - case NEU_METRICS_CATEGORY_ALL: - ctx.filter = NEU_NA_TYPE_DRIVER | NEU_NA_TYPE_NDRIVER | NEU_NA_TYPE_APP; - neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream); - neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx); - break; - } - -end: - if (NULL != stream) { - fclose(stream); - } - response(aio, NNG_HTTP_STATUS_OK == status ? result : NULL, status); - free(result); -} diff --git a/plugins/monitor/metric_handle.h b/plugins/monitor/metric_handle.h deleted file mode 100644 index dcad7739..00000000 --- a/plugins/monitor/metric_handle.h +++ /dev/null @@ -1,29 +0,0 @@ -/** - * NEURON IIoT System for Industry 4.0 - * Copyright (C) 2020-2022 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 - * License as published by the Free Software Foundation; either - * version 3 of the License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public License - * along with this program; if not, write to the Free Software Foundation, - * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - **/ - -#ifndef NEU_PLUGIN_MONITOR_METRIC_H -#define NEU_PLUGIN_MONITOR_METRIC_H - -#include - -#include "json/neu_json_fn.h" - -void handle_get_metric(nng_aio *aio); - -#endif diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 3120c05d..2973d44e 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -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); diff --git a/plugins/monitor/monitor.h b/plugins/monitor/monitor.h index 726e33ce..4581bc17 100644 --- a/plugins/monitor/monitor.h +++ b/plugins/monitor/monitor.h @@ -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; }; From e8dffb1ee25315d5922ce103a1bd7c4d9d601bc4 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 18 Sep 2023 11:43:50 +0800 Subject: [PATCH 3/5] feat(monitor): make as non-singleton plugin --- include/neuron/adapter.h | 67 +++++++++++++++ plugins/monitor/monitor.c | 2 +- plugins/monitor/mqtt_handle.c | 14 +--- plugins/restful/global_config_handle.c | 5 +- src/adapter/adapter.c | 2 +- src/core/manager.c | 110 +++++++++++++++++++------ src/core/manager_internal.c | 2 +- src/core/node_manager.c | 42 ++++++++++ src/core/node_manager.h | 6 ++ src/persist/persist.c | 21 ++--- 10 files changed, 214 insertions(+), 57 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index b0605556..316f9e90 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -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; diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 2973d44e..f396a34f 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -416,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", }; diff --git a/plugins/monitor/mqtt_handle.c b/plugins/monitor/mqtt_handle.c index d9852c33..a57e505f 100644 --- a/plugins/monitor/mqtt_handle.c +++ b/plugins/monitor/mqtt_handle.c @@ -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; } diff --git a/plugins/restful/global_config_handle.c b/plugins/restful/global_config_handle.c index 5b1c1386..d41ce0d1 100644 --- a/plugins/restful/global_config_handle.c +++ b/plugins/restful/global_config_handle.c @@ -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) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index ed0bb6ac..7d694bb4 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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, }; diff --git a/src/core/manager.c b/src/core/manager.c index 2627fd51..5567cc25 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -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) diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 71938ec6..74c7eb28 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -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; } diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 0d96b9b6..79f9ff0e 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -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 }; diff --git a/src/core/node_manager.h b/src/core/node_manager.h index 6df3abe2..3b7640d8 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -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); diff --git a/src/persist/persist.c b/src/persist/persist.c index 0540a4b8..23cb8b3a 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -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; } From b87da6154c6eec270b97eabcd7c9d4da1d3b38fe Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 18 Sep 2023 18:17:12 +0800 Subject: [PATCH 4/5] test: remove update node name test for monitor --- ft/metrics_test.robot | 16 +++++++++++++--- ft/node_test.robot | 12 ------------ ft/resource/common.resource | 1 + 3 files changed, 14 insertions(+), 15 deletions(-) diff --git a/ft/metrics_test.robot b/ft/metrics_test.robot index cedf2c80..d161506b 100644 --- a/ft/metrics_test.robot +++ b/ft/metrics_test.robot @@ -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} \ No newline at end of file + 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 diff --git a/ft/node_test.robot b/ft/node_test.robot index 8a65b412..72d30d6f 100644 --- a/ft/node_test.robot +++ b/ft/node_test.robot @@ -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 diff --git a/ft/resource/common.resource b/ft/resource/common.resource index 8a7b846b..4080dbf2 100644 --- a/ft/resource/common.resource +++ b/ft/resource/common.resource @@ -18,6 +18,7 @@ ${test_data_dir} ft/data ${PLUGIN_MODBUS_TCP} Modbus TCP +${PLUGIN_MONITOR} Monitor ${PLUGIN_MQTT} MQTT ${MQTT_CMD_NONE} From 78cd5f51053725b097e157e1dfe0f8f835a6ffb1 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 18 Sep 2023 15:21:05 +0800 Subject: [PATCH 5/5] ci: run pyft depending on branch --- .github/workflows/check.yml | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 1fe7dfe1..802e07e1 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -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: |