Merge pull request #1029 from fengzeroz/v2.3

version: 2.3.4

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.3.1
-  UI_VERSION: 2.3.1
+  VERSION: 2.3.4
+  UI_VERSION: 2.3.2
   BRANCH: v2.3
 
 jobs:

include/neuron/adapter.h

@@ -414,6 +414,7 @@ typedef struct {
 typedef struct {
     char                 driver[NEU_NODE_NAME_LEN];
     char                 group[NEU_GROUP_NAME_LEN];
+    int64_t              timestamp;
     uint16_t             n_tag;
     neu_resp_tag_value_t tags[];
 } neu_reqresp_trans_data_t;

include/neuron/metrics.h

@@ -39,6 +39,16 @@ typedef enum {
     NEU_METRIC_TYPE_COUNTER_SET,
 } neu_metric_type_e;
 
+// node running state
+#define NEU_METRIC_RUNNING_STATE "running_state"
+#define NEU_METRIC_RUNNING_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_RUNNING_STATE_HELP "Node running state"
+
+// node link state
+#define NEU_METRIC_LINK_STATE "link_state"
+#define NEU_METRIC_LINK_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_LINK_STATE_HELP "Node link state"
+
 // last round trip time in millisends
 #define NEU_METRIC_LAST_RTT_MS "last_rtt_ms"
 #define NEU_METRIC_LAST_RTT_MS_TYPE NEU_METRIC_TYPE_GAUAGE

include/neuron/neuron.h

@@ -27,6 +27,7 @@ extern "C" {
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
+#include <string.h>
 
 #include "errcodes.h"
 

plugins/ekuiper/json_rw.c

@@ -201,7 +201,7 @@ int json_encode_read_resp(void *json_object, void *param)
 
     json_read_resp_header_t header = { .group_name = trans_data->group,
                                        .node_name  = trans_data->driver,
-                                       .timestamp  = global_timestamp };
+                                       .timestamp  = trans_data->timestamp };
 
     ret = json_encode_read_resp_header(json_object, &header);
     if (0 != ret) {

src/adapter/adapter.c

@@ -67,6 +67,10 @@ static const adapter_callbacks_t callback_funs = {
     adapter_register_metric(adapter, name, name##_HELP, name##_TYPE, init);
 
 #define REGISTER_DRIVER_METRICS(adapter)                     \
+    REGISTER_METRIC(adapter, NEU_METRIC_LINK_STATE,          \
+                    NEU_NODE_LINK_STATE_DISCONNECTED);       \
+    REGISTER_METRIC(adapter, NEU_METRIC_RUNNING_STATE,       \
+                    NEU_NODE_RUNNING_STATE_INIT);            \
     REGISTER_METRIC(adapter, NEU_METRIC_LAST_RTT_MS,         \
                     NEU_METRIC_LAST_RTT_MS_MAX);             \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_BYTES, 0);      \
@@ -75,6 +79,10 @@ static const adapter_callbacks_t callback_funs = {
     REGISTER_METRIC(adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, 0);
 
 #define REGISTER_APP_METRICS(adapter)                              \
+    REGISTER_METRIC(adapter, NEU_METRIC_LINK_STATE,                \
+                    NEU_NODE_LINK_STATE_DISCONNECTED);             \
+    REGISTER_METRIC(adapter, NEU_METRIC_RUNNING_STATE,             \
+                    NEU_NODE_RUNNING_STATE_INIT);                  \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_MSGS_TOTAL, 0);       \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 0); \
     REGISTER_METRIC(adapter, NEU_METRIC_RECV_MSGS_TOTAL, 0);

src/adapter/driver/driver.c

@@ -467,13 +467,14 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
             .second      = interval / 1000,
             .millisecond = interval % 1000,
             .usr_data    = (void *) find,
-            .type        = NEU_EVENT_TIMER_BLOCK,
+            .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
         neu_group_set_interval(find->group, interval);
 
         param.cb     = report_callback;
         find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+        param.type   = driver->adapter.module->timer_type;
         param.cb     = read_callback;
         find->read   = neu_event_add_timer(driver->driver_events, param);
 
@@ -769,6 +770,7 @@ static int report_callback(void *usr_data)
                sizeof(neu_reqresp_trans_data_t) +
                    utarray_len(tags) * sizeof(neu_resp_tag_value_t));
 
+    data->timestamp = global_timestamp;
     strcpy(data->driver, group->driver->adapter.name);
     strcpy(data->group, group->name);
     data->n_tag = read_report_group(global_timestamp,

src/adapter/metrics.c

@@ -163,6 +163,11 @@ void neu_metrics_visist(neu_metrics_cb_t cb, void *data)
         neu_plugin_common_t *common =
             neu_plugin_to_plugin_common(n->adapter->plugin);
 
+        n->adapter->cb_funs.update_metric(n->adapter, NEU_METRIC_RUNNING_STATE,
+                                          n->adapter->state, NULL);
+        n->adapter->cb_funs.update_metric(n->adapter, NEU_METRIC_LINK_STATE,
+                                          common->link_state, NULL);
+
         if (NEU_NA_TYPE_DRIVER == n->adapter->module->type) {
             ++g_metrics_.south_nodes;
             if (NEU_NODE_RUNNING_STATE_RUNNING == n->adapter->state) {

src/core/manager.c

@@ -58,7 +58,6 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
                                const char *ndoe);
 static void start_static_adapter(neu_manager_t *manager, const char *name);
 static int  update_timestamp(void *usr_data);
-static int  report_nodes_state(void *usr_data);
 static void start_single_adapter(neu_manager_t *manager, const char *name,
                                  const char *plugin_name, bool display);
 
@@ -75,13 +74,7 @@ neu_manager_t *neu_manager_create()
         .second      = 0,
         .millisecond = 10,
         .cb          = update_timestamp,
-        .type        = NEU_EVENT_TIMER_BLOCK,
-    };
-    neu_event_timer_param_t timer_param = {
-        .second      = 1,
-        .millisecond = 0,
-        .cb          = report_nodes_state,
-        .type        = NEU_EVENT_TIMER_BLOCK,
+        .type        = NEU_EVENT_TIMER_NOBLOCK,
     };
 
     neu_event_timer_param_t timer_level = {
@@ -138,9 +131,6 @@ neu_manager_t *neu_manager_create()
     timer_level.usr_data = (void *) manager;
     manager->timer_lev   = neu_event_add_timer(manager->events, timer_level);
 
-    timer_param.usr_data = (void *) manager;
-    manager->timer       = neu_event_add_timer(manager->events, timer_param);
-
     timestamp_timer_param.usr_data = (void *) manager;
     manager->timer_timestamp =
         neu_event_add_timer(manager->events, timestamp_timer_param);
@@ -155,7 +145,6 @@ void neu_manager_destroy(neu_manager_t *manager)
     nng_msg *           uninit_msg = NULL;
     UT_array *pipes = neu_node_manager_get_pipes_all(manager->node_manager);
 
-    neu_event_del_timer(manager->events, manager->timer);
     neu_event_del_timer(manager->events, manager->timer_lev);
     neu_event_del_timer(manager->events, manager->timer_timestamp);
     strcpy(header.sender, "manager");
@@ -733,46 +722,6 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header,
     }
 }
 
-static int report_nodes_state(void *usr_data)
-{
-    neu_manager_t *manager = (neu_manager_t *) usr_data;
-    UT_array *     apps    = neu_sub_msg_manager_get(manager->sub_msg_manager,
-                                             NEU_SUBSCRIBE_NODES_STATE);
-
-    neu_manager_count_tag(manager);
-    if (apps == NULL) {
-        return 0;
-    }
-
-    UT_array *states = neu_node_manager_get_state(manager->node_manager);
-
-    utarray_foreach(apps, char *, app)
-    {
-        nng_msg *                  msg         = NULL;
-        neu_reqresp_head_t         header      = { 0 };
-        neu_resp_get_nodes_state_t node_states = { 0 };
-        nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
-
-        header.type = NEU_REQRESP_NODES_STATE;
-        strcpy(header.sender, "manager");
-        strcpy(header.receiver, app);
-        node_states.states = utarray_clone(states);
-
-        msg = neu_msg_gen(&header, (void *) &node_states);
-        nng_msg_set_pipe(msg, pipe);
-
-        if (nng_sendmsg(manager->socket, msg, 0) != 0) {
-            nng_msg_free(msg);
-            nlog_warn("manager -> %s nodes state fail", app);
-        }
-    }
-
-    utarray_free(apps);
-    utarray_free(states);
-
-    return 0;
-}
-
 static int update_timestamp(void *usr_data)
 {
     (void) usr_data;

src/core/manager_internal.h

@@ -41,7 +41,6 @@ typedef struct neu_manager {
     neu_subscribe_mgr_t * subscribe_manager;
     neu_sub_msg_mgr_t *   sub_msg_manager;
 
-    neu_event_timer_t *timer;
     neu_event_timer_t *timer_timestamp;
 
     neu_event_timer_t *timer_lev;

version

@@ -1 +1 @@
-2.3.1
+2.3.4