feat(monitor): handling global metrics generation

include/neuron/adapter.h

@@ -52,6 +52,13 @@ typedef enum {
     NEU_NODE_STAT_MAX,
 } neu_node_stat_e;
 
+typedef enum {
+    NEU_METRICS_CATEGORY_GLOBAL,
+    NEU_METRICS_CATEGORY_DRIVER,
+    NEU_METRICS_CATEGORY_APP,
+    NEU_METRICS_CATEGORY_ALL,
+} neu_metrics_category_e;
+
 typedef struct {
     bool     core_dumped;
     uint64_t uptime_seconds;

plugins/monitor/metric_handle.c

@@ -21,6 +21,7 @@
 #include "utils/log.h"
 
 #include "metric_handle.h"
+#include "monitor.h"
 #include "restful/handle.h"
 #include "restful/http.h"
 
@@ -64,6 +65,8 @@ static int response(nng_aio *aio, char *content, enum nng_http_status status)
 
     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);
@@ -78,13 +81,79 @@ static int response(nng_aio *aio, char *content, enum nng_http_status status)
     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,
+                                      char **              result)
+{
+    neu_asprintf(result, METRIC_GLOBAL_TMPL, 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);
+}
+
 void handle_get_metric(nng_aio *aio)
 {
-    char *result = NULL;
+    char *        result = 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 = http_get_param(aio, "category", &cat_param_len);
+    if (NULL != cat_param &&
+        !parse_metrics_catgory(cat_param, cat_param_len, &cat)) {
+        plog_info(plugin, "invalid metrics category: %.*s", (int) cat_param_len,
+                  cat_param);
+        response(aio, NULL, NNG_HTTP_STATUS_BAD_REQUEST);
+        return;
+    }
+
+    pthread_mutex_lock(&plugin->mutex);
+    if (plugin->metrics_updating) {
+        // a metrics request to core is in progress
+        // in the assumption that this case is rare, we just finish error
+        pthread_mutex_unlock(&plugin->mutex);
+        nng_aio_finish(aio, NNG_EBUSY);
+        response(aio, NULL, NNG_HTTP_STATUS_SERVICE_UNAVAILABLE);
+        return;
+    }
+    switch (cat) {
+    case NEU_METRICS_CATEGORY_GLOBAL:
+        gen_global_metrics(&plugin->metrics, &result);
+        break;
+    case NEU_METRICS_CATEGORY_DRIVER:
+    case NEU_METRICS_CATEGORY_APP:
+    case NEU_METRICS_CATEGORY_ALL:
+        break;
+    }
+    pthread_mutex_unlock(&plugin->mutex);
 
-    neu_asprintf(&result, METRIC_GLOBAL_TMPL, 0, 0, 0, 0, 0, 0, 0, 0);
     if (NULL == result) {
-        nng_aio_finish(aio, NNG_EINTERNAL);
+        response(aio, NULL, NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR);
         return;
     }
 

plugins/monitor/monitor.c

@@ -21,9 +21,6 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include <nng/nng.h>
-#include <nng/supplemental/http/http.h>
-
 #include "metric_handle.h"
 #include "monitor.h"
 #include "restful/handle.h"
@@ -31,10 +28,7 @@
 
 #define neu_plugin_module default_monitor_plugin_module
 
-struct neu_plugin {
-    neu_plugin_common_t common;
-    nng_http_server *   api_server;
-};
+static struct neu_plugin *g_monitor_plugin_;
 
 static nng_http_server *server_init()
 {
@@ -99,6 +93,11 @@ 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");
@@ -114,6 +113,7 @@ static neu_plugin_t *monitor_plugin_open(void)
         return NULL;
     }
 
+    g_monitor_plugin_ = plugin;
     return plugin;
 }
 
@@ -123,17 +123,45 @@ static int monitor_plugin_close(neu_plugin_t *plugin)
 
     nng_http_server_stop(plugin->api_server);
     nng_http_server_release(plugin->api_server);
+    pthread_mutex_destroy(&plugin->mutex);
     free(plugin);
     nlog_info("Success to free plugin: %s", neu_plugin_module.module_name);
 
     return rv;
 }
 
+static inline int send_metrics_req(neu_plugin_t *plugin)
+{
+    neu_reqresp_head_t header = {
+        .type = NEU_REQRESP_METRICS,
+    };
+    neu_reqresp_metrics_t cmd = { .metrics = &plugin->metrics };
+
+    pthread_mutex_lock(&plugin->mutex);
+    plugin->metrics_updating = true;
+    pthread_mutex_unlock(&plugin->mutex);
+
+    return neu_plugin_op(plugin, header, &cmd);
+}
+
+static int monitor_timer_cb(void *data)
+{
+    neu_plugin_t *plugin = data;
+    return send_metrics_req(plugin);
+}
+
 static int monitor_plugin_init(neu_plugin_t *plugin)
 {
     int rv = 0;
 
-    (void) plugin;
+    plugin->events                = neu_event_new();
+    neu_event_timer_param_t param = {
+        .second      = 2,
+        .millisecond = 0,
+        .cb          = monitor_timer_cb,
+        .usr_data    = plugin,
+    };
+    plugin->timer = neu_event_add_timer(plugin->events, param);
 
     nlog_info("Initialize plugin: %s", neu_plugin_module.module_name);
     return rv;
@@ -143,7 +171,15 @@ static int monitor_plugin_uninit(neu_plugin_t *plugin)
 {
     int rv = 0;
 
-    (void) plugin;
+    if (NULL != plugin->timer) {
+        neu_event_del_timer(plugin->events, plugin->timer);
+        plugin->events = NULL;
+    }
+
+    if (NULL != plugin->events) {
+        neu_event_close(plugin->events);
+        plugin->events = NULL;
+    }
 
     nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name);
     return rv;
@@ -171,6 +207,12 @@ static int monitor_plugin_request(neu_plugin_t *      plugin,
         nlog_warn("recv error code: %d", resp_err->error);
         break;
     }
+    case NEU_REQRESP_METRICS: {
+        pthread_mutex_lock(&plugin->mutex);
+        plugin->metrics_updating = false;
+        pthread_mutex_unlock(&plugin->mutex);
+        break;
+    }
     default:
         nlog_fatal("recv unhandle msg: %s",
                    neu_reqresp_type_string(header->type));
@@ -214,3 +256,8 @@ const neu_plugin_module_t default_monitor_plugin_module = {
     .kind         = NEU_PLUGIN_KIND_SYSTEM,
     .type         = NEU_NA_TYPE_APP,
 };
+
+neu_plugin_t *neu_monitor_get_plugin()
+{
+    return g_monitor_plugin_;
+}

plugins/monitor/monitor.h

@@ -20,8 +20,26 @@
 #ifndef NEU_PLUGIN_MONITOR_H
 #define NEU_PLUGIN_MONITOR_H
 
+#include <pthread.h>
+
+#include <nng/nng.h>
+#include <nng/supplemental/http/http.h>
+
+#include "adapter.h"
+#include "event/event.h"
 #include "plugin.h"
 
+struct neu_plugin {
+    neu_plugin_common_t common;
+    pthread_mutex_t     mutex;
+    bool                metrics_updating;
+    neu_metrics_t       metrics;
+    neu_events_t *      events;
+    neu_event_timer_t * timer;
+    nng_http_server *   api_server;
+};
+
 extern const neu_plugin_module_t default_monitor_plugin_module;
+neu_plugin_t *                   neu_monitor_get_plugin();
 
 #endif