Merge pull request #1969 from fengzeroz/v2.7

version: 2.7.5

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.7.2
+  VERSION: 2.7.5
   UI_VERSION: 2.6.3
   BRANCH: v2.7
 

cmake/cross.cmake

@@ -21,3 +21,7 @@ link_directories(${CMAKE_STAGING_PREFIX})
 if(CMAKE_SYSTEM_PROCESSOR MATCHES "mips32")
 	link_libraries(atomic)
 endif()
+
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "riscv64")
+	link_libraries(atomic)
+endif()

include/neuron/define.h

@@ -24,14 +24,14 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 7
-#define NEU_VERSION_FIX 2
+#define NEU_VERSION_FIX 5
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
 #define NEU_GET_VERSION_FIX(V) ((0x000000ff & V))
 #define NEU_VERSION(MAJOR, MINOR, FIX) ((MAJOR << 16) | (MINOR << 8) | FIX)
 
-#define NEU_TAG_NAME_LEN 64
+#define NEU_TAG_NAME_LEN 128
 #define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
 #define NEU_GROUP_MAX_PER_NODE 512

include/neuron/metrics.h

@@ -26,24 +26,37 @@ extern "C" {
 
 #include <stdint.h>
 
+#include <pthread.h>
+
 #include "define.h"
 #include "type.h"
+#include "utils/rolling_counter.h"
+#include "utils/utextend.h"
 #include "utils/uthash.h"
 
+extern int64_t global_timestamp;
+
 typedef enum {
     NEU_METRIC_TYPE_COUNTER,
     NEU_METRIC_TYPE_GAUAGE,
     NEU_METRIC_TYPE_COUNTER_SET,
+    NEU_METRIC_TYPE_ROLLING_COUNTER,
+
+    NEU_METRIC_TYPE_FLAG_NO_RESET = 0x80,
 } neu_metric_type_e;
 
+#define NEU_METRIC_TYPE_MASK 0x0F
+
 // 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_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #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_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_LINK_STATE_HELP "Node link state"
 
 // last round trip time in millisends
@@ -79,13 +92,15 @@ typedef enum {
 // maintained by neuron core
 // number of tags in group
 #define NEU_METRIC_TAGS_TOTAL "tags_total"
-#define NEU_METRIC_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_TAGS_TOTAL_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_TAGS_TOTAL_HELP "Total number of tags in the node"
 
 // maintained by neuron core
 // number of tags in group
 #define NEU_METRIC_GROUP_TAGS_TOTAL "group_tags_total"
-#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_GROUP_TAGS_TOTAL_HELP "Total number of tags in the group"
 
 // number of messages sent in last group timer
@@ -131,6 +146,96 @@ typedef enum {
 #define NEU_METRIC_RECV_MSGS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER
 #define NEU_METRIC_RECV_MSGS_TOTAL_HELP "Total number of messages received"
 
+// number of trans data message within the last 5 seconds
+#define NEU_METRIC_TRANS_DATA_5S "last_5s_trans_data_msgs"
+#define NEU_METRIC_TRANS_DATA_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_TRANS_DATA_5S_HELP \
+    "Number of internal trans data message within the last 5 seconds"
+
+// number of trans data message within the last 30 seconds
+#define NEU_METRIC_TRANS_DATA_30S "last_30s_trans_data_msgs"
+#define NEU_METRIC_TRANS_DATA_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_TRANS_DATA_30S_HELP \
+    "Number of internal trans data message within the last 30 seconds"
+
+// number of trans data message within the last 60 seconds
+#define NEU_METRIC_TRANS_DATA_60S "last_60s_trans_data_msgs"
+#define NEU_METRIC_TRANS_DATA_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_TRANS_DATA_60S_HELP \
+    "Number of internal trans data message within the last 60 seconds"
+
+// number of bytes sent within the last 5 seconds
+#define NEU_METRIC_SEND_BYTES_5S "last_5s_send_bytes"
+#define NEU_METRIC_SEND_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_SEND_BYTES_5S_HELP \
+    "Number of bytes sent within the last 5 seconds"
+
+// number of bytes sent within the last 30 seconds
+#define NEU_METRIC_SEND_BYTES_30S "last_30s_send_bytes"
+#define NEU_METRIC_SEND_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_SEND_BYTES_30S_HELP \
+    "Number of bytes sent within the last 30 seconds"
+
+// number of bytes sent within the last 60 seconds
+#define NEU_METRIC_SEND_BYTES_60S "last_60s_send_bytes"
+#define NEU_METRIC_SEND_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_SEND_BYTES_60S_HELP \
+    "Number of bytes sent within the last 60 seconds"
+
+// number of bytes received within the last 5 seconds
+#define NEU_METRIC_RECV_BYTES_5S "last_5s_recv_bytes"
+#define NEU_METRIC_RECV_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_BYTES_5S_HELP \
+    "Number of bytes received within the last 5 seconds"
+
+// number of bytes received within the last 30 seconds
+#define NEU_METRIC_RECV_BYTES_30S "last_30s_recv_bytes"
+#define NEU_METRIC_RECV_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_BYTES_30S_HELP \
+    "Number of bytes received within the last 30 seconds"
+
+// number of bytes received within the last 60 seconds
+#define NEU_METRIC_RECV_BYTES_60S "last_60s_recv_bytes"
+#define NEU_METRIC_RECV_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_BYTES_60S_HELP \
+    "Number of bytes received within the last 60 seconds"
+
+// number of messages received within the last 5 seconds
+#define NEU_METRIC_RECV_MSGS_5S "last_5s_recv_msgs"
+#define NEU_METRIC_RECV_MSGS_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_MSGS_5S_HELP \
+    "Number of messages received within the last 5 seconds"
+
+// number of messages received within the last 30 seconds
+#define NEU_METRIC_RECV_MSGS_30S "last_30s_recv_msgs"
+#define NEU_METRIC_RECV_MSGS_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_MSGS_30S_HELP \
+    "Number of messages received within the last 30 seconds"
+
+// number of messages received within the last 60 seconds
+#define NEU_METRIC_RECV_MSGS_60S "last_60s_recv_msgs"
+#define NEU_METRIC_RECV_MSGS_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_RECV_MSGS_60S_HELP \
+    "Number of messages received within the last 60 seconds"
+
+// number of disconnection within the last 60 seconds
+#define NEU_METRIC_DISCONNECTION_60S "last_60s_disconnections"
+#define NEU_METRIC_DISCONNECTION_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_DISCONNECTION_60S_HELP \
+    "Number of disconnection within the last 60 seconds"
+
+// number of disconnection within the last 600 seconds
+#define NEU_METRIC_DISCONNECTION_600S "last_600s_disconnections"
+#define NEU_METRIC_DISCONNECTION_600S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_DISCONNECTION_600S_HELP \
+    "Number of disconnection within the last 600 seconds"
+
+// number of disconnection within the last 1800 seconds
+#define NEU_METRIC_DISCONNECTION_1800S "last_1800s_disconnections"
+#define NEU_METRIC_DISCONNECTION_1800S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
+#define NEU_METRIC_DISCONNECTION_1800S_HELP \
+    "Number of disconnection within the last 1800 seconds"
+
 typedef enum {
     NEU_METRICS_CATEGORY_GLOBAL,
     NEU_METRICS_CATEGORY_DRIVER,
@@ -140,11 +245,12 @@ typedef enum {
 
 // metric entry
 typedef struct {
-    const char *      name;  // NOTE: should points to string literal
-    const char *      help;  // NOTE: should points to string literal
-    neu_metric_type_e type;  //
-    uint64_t          value; //
-    UT_hash_handle    hh;    // ordered by name
+    const char *           name;  // NOTE: should points to string literal
+    const char *           help;  // NOTE: should points to string literal
+    neu_metric_type_e      type;  //
+    uint64_t               value; //
+    neu_rolling_counter_t *rcnt;  //
+    UT_hash_handle         hh;    // ordered by name
 } neu_metric_entry_t;
 
 // group metrics
@@ -156,6 +262,7 @@ typedef struct {
 
 // node metrics
 typedef struct {
+    pthread_mutex_t      lock;          // spin lock
     neu_node_type_e      type;          // node type
     char *               name;          // node name
     neu_metric_entry_t * entries;       // node metric entries
@@ -192,9 +299,34 @@ typedef struct {
     neu_metric_entry_t *registered_metrics;
 } neu_metrics_t;
 
+void neu_metrics_init();
+void neu_metrics_add_node(const neu_adapter_t *adapter);
+void neu_metrics_del_node(const neu_adapter_t *adapter);
+int  neu_metrics_register_entry(const char *name, const char *help,
+                                neu_metric_type_e type);
+void neu_metrics_unregister_entry(const char *name);
+
+typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
+void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
+
+static inline bool neu_metric_type_is_counter(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_COUNTER == (type & NEU_METRIC_TYPE_MASK);
+}
+
+static inline bool neu_metric_type_is_rolling_counter(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_ROLLING_COUNTER == (type & NEU_METRIC_TYPE_MASK);
+}
+
+static inline bool neu_metric_type_no_reset(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_FLAG_NO_RESET & type;
+}
+
 static inline const char *neu_metric_type_str(neu_metric_type_e type)
 {
-    if (NEU_METRIC_TYPE_COUNTER == type) {
+    if (neu_metric_type_is_counter(type)) {
         return "counter";
     } else {
         return "gauge";
@@ -207,6 +339,9 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
 
 static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
 {
+    if (neu_metric_type_is_rolling_counter(entry->type)) {
+        neu_rolling_counter_free(entry->rcnt);
+    }
     free(entry);
 }
 
@@ -220,22 +355,41 @@ static inline void neu_group_metrics_free(neu_group_metrics_t *group_metrics)
     HASH_ITER(hh, group_metrics->entries, e, tmp)
     {
         HASH_DEL(group_metrics->entries, e);
+        neu_metrics_unregister_entry(e->name);
         neu_metric_entry_free(e);
     }
     free(group_metrics->name);
     free(group_metrics);
 }
 
+static inline neu_node_metrics_t *
+neu_node_metrics_new(neu_adapter_t *adapter, neu_node_type_e type, char *name)
+{
+    neu_node_metrics_t *node_metrics =
+        (neu_node_metrics_t *) calloc(1, sizeof(*node_metrics));
+    if (NULL != node_metrics) {
+        pthread_mutex_init(&node_metrics->lock, NULL);
+        node_metrics->type    = type;
+        node_metrics->name    = name;
+        node_metrics->adapter = adapter;
+        // neu_metrics_add_node(adapter);
+    }
+    return node_metrics;
+}
+
 static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
 {
     if (NULL == node_metrics) {
         return;
     }
 
+    pthread_mutex_destroy(&node_metrics->lock);
+
     neu_metric_entry_t *e = NULL, *etmp = NULL;
     HASH_ITER(hh, node_metrics->entries, e, etmp)
     {
         HASH_DEL(node_metrics->entries, e);
+        neu_metrics_unregister_entry(e->name);
         neu_metric_entry_free(e);
     }
 
@@ -249,15 +403,152 @@ static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
     free(node_metrics);
 }
 
-void neu_metrics_init();
-void neu_metrics_add_node(const neu_adapter_t *adapter);
-void neu_metrics_del_node(const neu_adapter_t *adapter);
-int  neu_metrics_register_entry(const char *name, const char *help,
-                                neu_metric_type_e type);
-void neu_metrics_unregister_entry(const char *name);
+static inline int neu_node_metrics_add(neu_node_metrics_t *node_metrics,
+                                       const char *group_name, const char *name,
+                                       const char *help, neu_metric_type_e type,
+                                       uint64_t init)
+{
+    int rv = 0;
 
-typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
-void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
+    if (0 > neu_metrics_register_entry(name, help, type)) {
+        return -1;
+    }
+
+    pthread_mutex_lock(&node_metrics->lock);
+    if (NULL == group_name) {
+        rv = neu_metric_entries_add(&node_metrics->entries, name, help, type,
+                                    init);
+    } else {
+        neu_group_metrics_t *group_metrics = NULL;
+        HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
+        if (NULL != group_metrics) {
+            rv = neu_metric_entries_add(&group_metrics->entries, name, help,
+                                        type, init);
+        } else {
+            group_metrics =
+                (neu_group_metrics_t *) calloc(1, sizeof(*group_metrics));
+            if (NULL != group_metrics &&
+                NULL != (group_metrics->name = strdup(group_name))) {
+                HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
+                rv = neu_metric_entries_add(&group_metrics->entries, name, help,
+                                            type, init);
+            } else {
+                free(group_metrics);
+                rv = -1;
+            }
+        }
+    }
+    pthread_mutex_unlock(&node_metrics->lock);
+
+    if (0 != rv) {
+        neu_metrics_unregister_entry(name);
+    }
+    return rv;
+}
+
+static inline int neu_node_metrics_update(neu_node_metrics_t *node_metrics,
+                                          const char *        group,
+                                          const char *metric_name, uint64_t n)
+{
+    neu_metric_entry_t *entry = NULL;
+
+    pthread_mutex_lock(&node_metrics->lock);
+    if (NULL == group) {
+        HASH_FIND_STR(node_metrics->entries, metric_name, entry);
+    } else if (NULL != node_metrics->group_metrics) {
+        neu_group_metrics_t *g = NULL;
+        HASH_FIND_STR(node_metrics->group_metrics, group, g);
+        if (NULL != g) {
+            HASH_FIND_STR(g->entries, metric_name, entry);
+        }
+    }
+
+    if (NULL == entry) {
+        pthread_mutex_unlock(&node_metrics->lock);
+        return -1;
+    }
+
+    if (neu_metric_type_is_counter(entry->type)) {
+        entry->value += n;
+    } else if (neu_metric_type_is_rolling_counter(entry->type)) {
+        entry->value =
+            neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
+    } else {
+        entry->value = n;
+    }
+    pthread_mutex_unlock(&node_metrics->lock);
+
+    return 0;
+}
+
+static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
+{
+    neu_metric_entry_t *entry = NULL;
+
+    pthread_mutex_lock(&node_metrics->lock);
+    HASH_LOOP(hh, node_metrics->entries, entry)
+    {
+        if (!neu_metric_type_no_reset(entry->type)) {
+            entry->value = 0;
+            if (neu_metric_type_is_rolling_counter(entry->type)) {
+                neu_rolling_counter_reset(entry->rcnt);
+            }
+        }
+    }
+
+    neu_group_metrics_t *g = NULL;
+    HASH_LOOP(hh, node_metrics->group_metrics, g)
+    {
+        HASH_LOOP(hh, g->entries, entry)
+        {
+            if (!neu_metric_type_no_reset(entry->type)) {
+                entry->value = 0;
+                if (neu_metric_type_is_rolling_counter(entry->type)) {
+                    neu_rolling_counter_reset(entry->rcnt);
+                }
+            }
+        }
+    }
+    pthread_mutex_unlock(&node_metrics->lock);
+}
+
+static inline int
+neu_node_metrics_update_group(neu_node_metrics_t *node_metrics,
+                              const char *        group_name,
+                              const char *        new_group_name)
+{
+    int                  rv            = -1;
+    neu_group_metrics_t *group_metrics = NULL;
+
+    pthread_mutex_lock(&node_metrics->lock);
+    HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
+    if (NULL != group_metrics) {
+        char *name = strdup(new_group_name);
+        if (NULL != name) {
+            HASH_DEL(node_metrics->group_metrics, group_metrics);
+            free(group_metrics->name);
+            group_metrics->name = name;
+            HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
+            rv = 0;
+        }
+    }
+    pthread_mutex_unlock(&node_metrics->lock);
+
+    return rv;
+}
+
+static inline void neu_node_metrics_del_group(neu_node_metrics_t *node_metrics,
+                                              const char *        group_name)
+{
+    neu_group_metrics_t *gm = NULL;
+    pthread_mutex_lock(&node_metrics->lock);
+    HASH_FIND_STR(node_metrics->group_metrics, group_name, gm);
+    if (NULL != gm) {
+        HASH_DEL(node_metrics->group_metrics, gm);
+        neu_group_metrics_free(gm);
+    }
+    pthread_mutex_unlock(&node_metrics->lock);
+}
 
 #ifdef __cplusplus
 }

include/neuron/msg.h

@@ -137,6 +137,10 @@ typedef enum neu_reqresp_type {
 
     NEU_REQ_UPDATE_LOG_LEVEL,
 
+    NEU_REQ_PRGFILE_UPLOAD,
+    NEU_REQ_PRGFILE_PROCESS,
+    NEU_RESP_PRGFILE_PROCESS,
+
 } neu_reqresp_type_e;
 
 static const char *neu_reqresp_type_string_t[] = {
@@ -236,6 +240,9 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_GET_NDRIVER_TAGS]        = "NEU_RESP_GET_NDRIVER_TAGS",
 
     [NEU_REQ_UPDATE_LOG_LEVEL] = "NEU_REQ_UPDATE_LOG_LEVEL",
+    [NEU_REQ_PRGFILE_UPLOAD]   = "NEU_REQ_PRGFILE_UPLOAD",
+    [NEU_REQ_PRGFILE_PROCESS]  = "NEU_REQ_PRGFILE_PROCESS",
+    [NEU_RESP_PRGFILE_PROCESS] = "NEU_RESP_PRGFILE_PROCESS",
 };
 
 inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -1010,6 +1017,8 @@ static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value,
         tag_json->value.val_bit = tag_value->value.value.u8;
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         tag_json->t             = NEU_JSON_STR;
         tag_json->value.val_str = tag_value->value.value.str;
         break;
@@ -1138,6 +1147,25 @@ inline static void neu_msg_exchange(neu_reqresp_head_t *header)
     strcpy(header->receiver, tmp);
 }
 
+typedef struct {
+    char driver[NEU_NODE_NAME_LEN];
+    char path[128];
+    int  prg_int_param1;
+    int  prg_int_param2;
+    char prg_str_param1[128];
+    char prg_str_param2[128];
+} neu_req_prgfile_upload_t;
+
+typedef struct {
+    char driver[NEU_NODE_NAME_LEN];
+} neu_req_prgfile_process_t;
+
+typedef struct {
+    char path[128];
+    int  state;
+    char reason[256];
+} neu_resp_prgfile_process_t;
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/plugin.h

@@ -36,6 +36,14 @@ extern "C" {
 #define NEURON_PLUGIN_VER_1_0 \
     NEU_VERSION(NEU_VERSION_MAJOR, NEU_VERSION_MINOR, NEU_VERSION_FIX)
 
+#define NEU_PLUGIN_REGISTER_METRIC(plugin, name, init) \
+    plugin->common.adapter_callbacks->register_metric( \
+        plugin->common.adapter, name, name##_HELP, name##_TYPE, init)
+
+#define NEU_PLUGIN_UPDATE_METRIC(plugin, name, val, grp)                    \
+    plugin->common.adapter_callbacks->update_metric(plugin->common.adapter, \
+                                                    name, val, grp)
+
 extern int64_t global_timestamp;
 
 typedef struct neu_plugin_common {

include/neuron/tag.h

@@ -59,6 +59,7 @@ typedef union {
     } value16;
     struct {
         neu_datatag_endian_e endian;
+        bool                 is_default;
     } value32;
     struct {
         neu_datatag_endian_e endian;
@@ -66,6 +67,7 @@ typedef union {
     struct {
         uint16_t                  length;
         neu_datatag_string_type_e type;
+        bool                      is_default;
     } string;
 
     struct {

include/neuron/type.h

@@ -31,25 +31,27 @@ extern "C" {
 #include <stdio.h>
 
 typedef enum {
-    NEU_TYPE_INT8   = 1,
-    NEU_TYPE_UINT8  = 2,
-    NEU_TYPE_INT16  = 3,
-    NEU_TYPE_UINT16 = 4,
-    NEU_TYPE_INT32  = 5,
-    NEU_TYPE_UINT32 = 6,
-    NEU_TYPE_INT64  = 7,
-    NEU_TYPE_UINT64 = 8,
-    NEU_TYPE_FLOAT  = 9,
-    NEU_TYPE_DOUBLE = 10,
-    NEU_TYPE_BIT    = 11,
-    NEU_TYPE_BOOL   = 12,
-    NEU_TYPE_STRING = 13,
-    NEU_TYPE_BYTES  = 14,
-    NEU_TYPE_ERROR  = 15,
-    NEU_TYPE_WORD   = 16,
-    NEU_TYPE_DWORD  = 17,
-    NEU_TYPE_LWORD  = 18,
-    NEU_TYPE_PTR    = 19,
+    NEU_TYPE_INT8          = 1,
+    NEU_TYPE_UINT8         = 2,
+    NEU_TYPE_INT16         = 3,
+    NEU_TYPE_UINT16        = 4,
+    NEU_TYPE_INT32         = 5,
+    NEU_TYPE_UINT32        = 6,
+    NEU_TYPE_INT64         = 7,
+    NEU_TYPE_UINT64        = 8,
+    NEU_TYPE_FLOAT         = 9,
+    NEU_TYPE_DOUBLE        = 10,
+    NEU_TYPE_BIT           = 11,
+    NEU_TYPE_BOOL          = 12,
+    NEU_TYPE_STRING        = 13,
+    NEU_TYPE_BYTES         = 14,
+    NEU_TYPE_ERROR         = 15,
+    NEU_TYPE_WORD          = 16,
+    NEU_TYPE_DWORD         = 17,
+    NEU_TYPE_LWORD         = 18,
+    NEU_TYPE_PTR           = 19,
+    NEU_TYPE_TIME          = 20,
+    NEU_TYPE_DATA_AND_TIME = 21,
 } neu_type_e;
 
 inline static const char *neu_type_string(neu_type_e type)
@@ -93,6 +95,10 @@ inline static const char *neu_type_string(neu_type_e type)
         return "NEU_TYPE_LWORD";
     case NEU_TYPE_PTR:
         return "NEU_TYPE_PTR";
+    case NEU_TYPE_TIME:
+        return "NEU_TYPE_TIME";
+    case NEU_TYPE_DATA_AND_TIME:
+        return "NEU_TYPE_DATA_AND_TIME";
     }
 
     return "";
@@ -188,7 +194,9 @@ static inline char *neu_value_str(neu_type_e type, neu_value_u value)
         snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
                  value.boolean);
         break;
-    case NEU_TYPE_STRING: {
+    case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME: {
         snprintf(str, sizeof(str), "type: %s, value: %c%c%c",
                  neu_type_string(type), value.str[0], value.str[1],
                  value.str[2]);

include/neuron/utils/rolling_counter.h

@@ -0,0 +1,115 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 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 NEURON_UTILS_ROLLING_COUNTER_H
+#define NEURON_UTILS_ROLLING_COUNTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <assert.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+/** Rolling counter.
+ *
+ * This counter is for counting values within some latest time span, like
+ * network bytes sent within the last 5 seconds etc.
+ */
+typedef struct {
+    uint64_t val;      // accumulator
+    uint64_t ts;       // head time stamp in milliseconds
+    uint32_t res : 21; // time resolution in milliseconds
+    uint32_t hd : 5;   // head position
+    uint32_t n : 6;    // number of counters
+    uint32_t counts[]; // bins of counters
+} neu_rolling_counter_t;
+
+/** Create rolling counter.
+ *
+ * @param   span   time span in milliseconds
+ */
+static inline neu_rolling_counter_t *neu_rolling_counter_new(unsigned span)
+{
+    unsigned n = span <= 6000 ? 4 : span <= 32000 ? 8 : span <= 64000 ? 16 : 32;
+    assert(span / n < (1 << 22)); // should not overflow ti
+
+    neu_rolling_counter_t *counter = (neu_rolling_counter_t *) calloc(
+        1, sizeof(*counter) + sizeof(counter->counts[0]) * n);
+    if (counter) {
+        counter->res = span / n;
+        counter->n   = n;
+    }
+    return counter;
+}
+
+/** Destructs the rolling counter.
+ */
+static inline void neu_rolling_counter_free(neu_rolling_counter_t *counter)
+{
+    if (counter) {
+        free(counter);
+    }
+}
+
+/** Increment the rolling counter and return the value;
+ *
+ * @param   ts    time stamp in milliseconds, should be monotonic
+ * @param   dt    delta value to increment by
+ */
+static inline uint64_t neu_rolling_counter_inc(neu_rolling_counter_t *counter,
+                                               uint64_t ts, unsigned dt)
+{
+    uint64_t step = (ts - counter->ts) / counter->res;
+    for (unsigned i = 0; i < step && i < counter->n; ++i) {
+        counter->hd = (counter->hd + 1) & (counter->n - 1);
+        counter->val -= counter->counts[counter->hd];
+        counter->counts[counter->hd] = 0;
+    }
+    counter->val += dt;
+    counter->counts[counter->hd] += dt;
+    counter->ts += step * counter->res;
+    return counter->val;
+}
+
+/** Reset the counter.
+ */
+static inline void neu_rolling_counter_reset(neu_rolling_counter_t *counter)
+{
+    counter->val = 0;
+    counter->hd  = 0;
+    memset(counter->counts, 0, counter->n * sizeof(counter->counts[0]));
+}
+
+/** Return the counter value.
+ *
+ * NOTE: may return stale value if the counter is not updated frequent enough.
+ */
+static inline uint64_t neu_rolling_counter_value(neu_rolling_counter_t *counter)
+{
+    return counter->val;
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

persistence/0060_2.7.3_tag_name_length.sql

@@ -0,0 +1,40 @@
+/**
+ * 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.
+ **/
+BEGIN TRANSACTION;
+
+ALTER TABLE tags RENAME TO temp_tags;
+CREATE TABLE IF NOT EXISTS
+  tags (
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    name TEXT NULL check(length(name) <= 128),
+    address TEXT NULL check(length(address) <= 128),
+    attribute INTEGER NOT NULL check(attribute BETWEEN 0 AND 15),
+    precision INTEGER NOT NULL check(precision BETWEEN 0 AND 17),
+    decimal REAL NOT NULL,
+    type INTEGER NOT NULL check(type BETWEEN 0 AND 19),
+    description TEXT NULL check(length(description) <= 512),
+    value TEXT,
+    UNIQUE (driver_name, group_name, name),
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO tags SELECT * FROM temp_tags;
+
+DROP TABLE temp_tags;
+COMMIT;

persistence/0070_2.7.5_tag_time_type.sql

@@ -0,0 +1,43 @@
+/**
+* NEURON IIoT System for Industry 4.0
+* Copyright (C) 2020-2024 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.
+**/
+BEGIN TRANSACTION;
+
+ALTER TABLE tags RENAME TO temp_tags;
+
+CREATE TABLE
+  IF NOT EXISTS tags (
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    name TEXT NULL check (length (name) <= 128),
+    address TEXT NULL check (length (address) <= 128),
+    attribute INTEGER NOT NULL check (attribute BETWEEN 0 AND 15),
+    precision INTEGER NOT NULL check (precision BETWEEN 0 AND 17),
+    decimal REAL NOT NULL,
+    type INTEGER NOT NULL check (type BETWEEN 0 AND 21),
+    description TEXT NULL check (length (description) <= 512),
+    value TEXT,
+    UNIQUE (driver_name, group_name, name),
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+
+INSERT INTO tags SELECT * FROM temp_tags;
+
+DROP TABLE temp_tags;
+
+COMMIT;

plugins/ekuiper/json_rw.c

@@ -185,7 +185,8 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
 
     json_obj = neu_json_decode_newb(buf, len);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {
@@ -208,27 +209,20 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
     };
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
-    if (ret != 0) {
-        goto decode_fail;
-    }
-
     req->node_name  = req_elems[0].v.val_str;
     req->group_name = req_elems[1].v.val_str;
     req->tag_name   = req_elems[2].v.val_str;
     req->t          = req_elems[3].t;
     req->value      = req_elems[3].v;
 
-    *result = req;
-    goto decode_exit;
-
-decode_fail:
-    free(req);
-    ret = -1;
-
-decode_exit:
-    if (json_obj != NULL) {
-        neu_json_decode_free(json_obj);
+    if (0 == ret) {
+        *result = req;
+    } else {
+        json_decode_write_req_free(req);
+        ret = -1;
     }
+
+    neu_json_decode_free(json_obj);
     return ret;
 }
 
@@ -243,6 +237,8 @@ void json_decode_write_req_free(json_write_req_t *req)
     free(req->tag_name);
     if (req->t == NEU_JSON_STR) {
         free(req->value.val_str);
+    } else if (req->t == NEU_JSON_BYTES) {
+        free(req->value.val_bytes.bytes);
     }
 
     free(req);

plugins/ekuiper/plugin_ekuiper.c

@@ -71,6 +71,10 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     nng_mtx_lock(plugin->mtx);
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
     nng_mtx_unlock(plugin->mtx);
+
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL);
 }
 
 static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
@@ -96,6 +100,22 @@ static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
 
     plugin->recv_aio = recv_aio;
 
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000);
+
     plog_notice(plugin, "plugin initialized");
     return rv;
 }
@@ -114,10 +134,9 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin)
     return rv;
 }
 
-static int ekuiper_plugin_start(neu_plugin_t *plugin)
+static inline int start(neu_plugin_t *plugin, const char *url)
 {
-    int   rv  = 0;
-    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+    int rv = 0;
 
     rv = nng_pair0_open(&plugin->sock);
     if (rv != 0) {
@@ -143,16 +162,43 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
         return rv;
     }
 
+    nng_mtx_lock(plugin->mtx);
+    while (plugin->receiving) {
+        nng_mtx_unlock(plugin->mtx);
+        nng_msleep(10);
+        nng_mtx_lock(plugin->mtx);
+    }
+    plugin->receiving = true;
     nng_recv_aio(plugin->sock, plugin->recv_aio);
+    nng_mtx_unlock(plugin->mtx);
+
+    return NEU_ERR_SUCCESS;
+}
+
+static int ekuiper_plugin_start(neu_plugin_t *plugin)
+{
+    int   rv  = 0;
+    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+
+    rv = start(plugin, url);
+    if (rv != 0) {
+        return rv;
+    }
+
     plugin->started = true;
     plog_notice(plugin, "start successfully");
 
     return NEU_ERR_SUCCESS;
 }
 
-static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+static inline void stop(neu_plugin_t *plugin)
 {
     nng_close(plugin->sock);
+}
+
+static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+{
+    stop(plugin);
     plugin->started = false;
     plog_notice(plugin, "stop successfully");
     return NEU_ERR_SUCCESS;
@@ -201,41 +247,6 @@ error:
     return -1;
 }
 
-static inline int check_url_listenable(neu_plugin_t *plugin, const char *url,
-                                       const char *host, uint16_t port)
-{
-    if (NULL != plugin->host &&                  // already configured and
-        port == plugin->port &&                  // port is the same, then if
-        (0 == strcmp(plugin->host, host)         // 1. host is the same
-         || 0 == strcmp("0.0.0.0", host)         // 2. to bind to any address
-         || 0 == strcmp("0.0.0.0", plugin->host) // 3. bound to any address
-         )) {
-        // early return, no need check url is listenable
-        return 0;
-    }
-
-    nng_socket sock = NNG_SOCKET_INITIALIZER;
-    int        rv   = nng_pair0_open(&sock);
-    if (0 != rv) {
-        plog_error(plugin, "nng_pair0_open: %s", nng_strerror(rv));
-        return NEU_ERR_EINTERNAL;
-    }
-
-    if (0 != (rv = nng_listen(sock, url, NULL, 0))) {
-        plog_error(plugin, "nng_listen: %s", nng_strerror(rv));
-        if (NNG_EADDRINVAL == rv) {
-            rv = NEU_ERR_IP_ADDRESS_INVALID;
-        } else if (NNG_EADDRINUSE == rv) {
-            rv = NEU_ERR_IP_ADDRESS_IN_USE;
-        } else {
-            rv = NEU_ERR_EINTERNAL;
-        }
-    }
-
-    nng_close(sock);
-    return rv;
-}
-
 static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
 {
     int      rv   = 0;
@@ -255,23 +266,31 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (0 != (rv = check_url_listenable(plugin, url, host, port))) {
+    if (plugin->started) {
+        stop(plugin);
+    }
+
+    // check we could start the plugin with the new setting
+    if (0 != (rv = start(plugin, url))) {
+        // recover with old setting
+        if (plugin->started && 0 != start(plugin, plugin->url)) {
+            plog_warn(plugin, "restart host:%s port:%" PRIu16 " fail",
+                      plugin->host, plugin->port);
+        }
         goto error;
     }
 
+    if (!plugin->started) {
+        stop(plugin);
+    }
+
     plog_notice(plugin, "config success");
 
     free(plugin->host);
+    free(plugin->url);
     plugin->host = host;
     plugin->port = port;
-    free(plugin->url);
-    plugin->url = url;
-
-    if (plugin->started) {
-        // restart service
-        ekuiper_plugin_stop(plugin);
-        ekuiper_plugin_start(plugin);
-    }
+    plugin->url  = url;
 
     return rv;
 
@@ -303,6 +322,10 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
     case NEU_REQRESP_TRANS_DATA: {
         neu_reqresp_trans_data_t *trans_data = data;
 
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL);
+
         if (disconnected) {
             plog_debug(plugin, "not connected");
         } else {
@@ -310,12 +333,17 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
         }
         break;
     }
+    case NEU_REQRESP_NODE_DELETED: {
+        break;
+    }
     case NEU_REQ_UPDATE_NODE: {
         break;
     }
     case NEU_REQ_UPDATE_GROUP: {
         break;
     }
+    case NEU_REQ_DEL_GROUP:
+        break;
     case NEU_REQ_SUBSCRIBE_GROUP:
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
         neu_req_subscribe_t *sub_info = data;

plugins/ekuiper/plugin_ekuiper.h

@@ -37,6 +37,7 @@ struct neu_plugin {
     nng_socket          sock;
     nng_mtx *           mtx;
     bool                started;
+    bool                receiving;
     nng_aio *           recv_aio;
     char *              host;
     uint16_t            port;

plugins/ekuiper/read_write.c

@@ -30,9 +30,6 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
 {
     int rv = 0;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     char *           json_str = NULL;
     json_read_resp_t resp     = {
         .plugin     = plugin,
@@ -59,15 +56,18 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
     rv = nng_sendmsg(plugin->sock, msg,
                      NNG_FLAG_NONBLOCK); // TODO: use aio to send message
     if (0 == rv) {
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, json_len,
-                      NULL);
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1,
-                      NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
+                                 NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, json_len,
+                                 NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, json_len,
+                                 NULL);
     } else {
         plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
         nng_msg_free(msg);
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
-                      1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
     }
 }
 
@@ -80,12 +80,12 @@ void recv_data_callback(void *arg)
     char *            json_str = NULL;
     json_write_req_t *req      = NULL;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     rv = nng_aio_result(plugin->recv_aio);
     if (0 != rv) {
         plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
+        nng_mtx_lock(plugin->mtx);
+        plugin->receiving = false;
+        nng_mtx_unlock(plugin->mtx);
         return;
     }
 
@@ -93,9 +93,13 @@ void recv_data_callback(void *arg)
     json_str = nng_msg_body(msg);
     json_len = nng_msg_len(msg);
     plog_debug(plugin, "<< %.*s", (int) json_len, json_str);
-    update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, json_len,
-                  NULL);
-    update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, json_len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, json_len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, json_len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
     if (json_decode_write_req(json_str, json_len, &req) < 0) {
         plog_error(plugin, "fail decode write request json: %.*s",
                    (int) nng_msg_len(msg), json_str);

plugins/modbus/modbus_req.c

@@ -699,7 +699,7 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
                 plog_recv_protocol(plugin, recv_buf, ret);
             }
             neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
-            int ret_s = modbus_stack_recv(plugin->stack, 0, &pbuf);
+            int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
             if (ret_s == MODBUS_DEVICE_ERR) {
                 ret = ret_s;
             } else {

plugins/modbus/modbus_stack.c

@@ -84,10 +84,8 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
         return -1;
     }
 
-    if (stack->protocol == MODBUS_PROTOCOL_TCP) {
-        if (code.slave_id != slave_id) {
-            return -1;
-        }
+    if (code.slave_id != slave_id) {
+        return -1;
     }
 
     switch (code.function) {
@@ -102,18 +100,36 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
             return -1;
         }
 
-        if (stack->protocol == MODBUS_PROTOCOL_TCP && data.n_byte == 0xff) {
-            bytes = neu_protocol_unpack_buf(buf,
-                                            header.len -
-                                                sizeof(struct modbus_code) -
-                                                sizeof(struct modbus_data));
-            stack->value_fn(stack->ctx, code.slave_id,
-                            header.len - sizeof(struct modbus_code) -
-                                sizeof(struct modbus_data),
-                            bytes, 0);
-        } else {
+        switch (stack->protocol) {
+        case MODBUS_PROTOCOL_TCP:
+            if (data.n_byte == 0xff) {
+                bytes = neu_protocol_unpack_buf(buf,
+                                                header.len -
+                                                    sizeof(struct modbus_code) -
+                                                    sizeof(struct modbus_data));
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id,
+                                header.len - sizeof(struct modbus_code) -
+                                    sizeof(struct modbus_data),
+                                bytes, 0);
+            } else {
+                bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes,
+                                0);
+            }
+            break;
+        case MODBUS_PROTOCOL_RTU:
             bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+            if (bytes == NULL) {
+                return -1;
+            }
             stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, 0);
+            break;
         }
 
         break;

plugins/mqtt/mqtt_handle.c

@@ -276,15 +276,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
 
     neu_plugin_t *plugin = data;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     if (0 == errcode) {
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1,
-                      NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, len, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, len, NULL);
     } else {
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
-                      1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
     }
 
     free(payload);
@@ -293,16 +292,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
 static inline int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
                           char *payload, size_t payload_len)
 {
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
 
     int rv =
         neu_mqtt_client_publish(plugin->client, qos, topic, (uint8_t *) payload,
                                 (uint32_t) payload_len, plugin, publish_cb);
     if (0 != rv) {
         plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos);
-        update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
-                      1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
         free(payload);
         rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
     }
@@ -320,9 +317,13 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
     (void) qos;
     (void) topic;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-    update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
 
     char *json_str = malloc(len + 1);
     if (NULL == json_str) {
@@ -407,9 +408,13 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
     (void) qos;
     (void) topic;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-    update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
 
     char *json_str = malloc(len + 1);
     if (NULL == json_str) {
@@ -601,6 +606,14 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
     return 0;
 }
 
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
+{
+    route_tbl_del(&plugin->route_tbl, req->driver, req->group);
+    plog_notice(plugin, "del route driver:%s group:%s", req->driver,
+                req->group);
+    return 0;
+}
+
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
 {
     route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,

plugins/mqtt/mqtt_handle.h

@@ -49,6 +49,7 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
                              neu_req_unsubscribe_t *unsub_info);
 
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd);
 
 int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
 int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);

plugins/mqtt/mqtt_plugin.c

@@ -37,6 +37,9 @@ static void disconnect_cb(void *data)
 {
     neu_plugin_t *plugin      = data;
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL);
+    NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL);
     plog_notice(plugin, "plugin `%s` disconnected",
                 neu_plugin_module.module_name);
 }
@@ -60,13 +63,26 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
 static int mqtt_plugin_init(neu_plugin_t *plugin, bool load)
 {
     (void) load;
+
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_CACHED_MSGS_NUM, 0);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000);
+    NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000);
+
     plog_notice(plugin, "initialize plugin `%s` success",
                 neu_plugin_module.module_name);
-    neu_adapter_register_metric_cb_t register_metric =
-        plugin->common.adapter_callbacks->register_metric;
-    register_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM,
-                    NEU_METRIC_CACHED_MSGS_NUM_HELP,
-                    NEU_METRIC_CACHED_MSGS_NUM_TYPE, 0);
     return NEU_ERR_SUCCESS;
 }
 
@@ -333,15 +349,12 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
 {
     neu_err_code_e error = NEU_ERR_SUCCESS;
 
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
     // update cached messages number per seconds
     if (NULL != plugin->client &&
         (global_timestamp - plugin->cache_metric_update_ts) >= 1000) {
-        update_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM,
-                      neu_mqtt_client_get_cached_msgs_num(plugin->client),
-                      NULL);
+        NEU_PLUGIN_UPDATE_METRIC(
+            plugin, NEU_METRIC_CACHED_MSGS_NUM,
+            neu_mqtt_client_get_cached_msgs_num(plugin->client), NULL);
         plugin->cache_metric_update_ts = global_timestamp;
     }
 
@@ -353,6 +366,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
         error = handle_read_response(plugin, head->ctx, data);
         break;
     case NEU_REQRESP_TRANS_DATA: {
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL);
         error = handle_trans_data(plugin, data);
         break;
     }
@@ -368,6 +384,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_GROUP:
         error = handle_update_group(plugin, data);
         break;
+    case NEU_REQ_DEL_GROUP:
+        error = handle_del_group(plugin, data);
+        break;
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;

plugins/restful/metric_handle.c

@@ -19,8 +19,11 @@
 
 #include <stdio.h>
 
+#include <pthread.h>
+
 #include "define.h"
 #include "metrics.h"
+#include "plugin.h"
 #include "utils/asprintf.h"
 #include "utils/http.h"
 #include "utils/http_handler.h"
@@ -164,8 +167,8 @@ static inline void gen_global_metrics(const neu_metrics_t *metrics,
             metrics->south_running_nodes, metrics->south_disconnected_nodes);
 }
 
-static inline void
-gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
+static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
+                                           FILE *              stream)
 {
     fprintf(stream,
             "# HELP node_type Driver(1) or APP(2)\n"
@@ -175,8 +178,13 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
 
     neu_metric_entry_t *e = NULL;
 
+    pthread_mutex_lock(&node_metrics->lock);
     HASH_LOOP(hh, node_metrics->entries, e)
     {
+        if (neu_metric_type_is_rolling_counter(e->type)) {
+            // force clean stale value
+            e->value = neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
+        }
         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),
@@ -188,6 +196,11 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
     {
         HASH_LOOP(hh, g->entries, e)
         {
+            if (neu_metric_type_is_rolling_counter(e->type)) {
+                // force clean stale value
+                e->value =
+                    neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
+            }
             fprintf(stream,
                     "# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\",group=\"%s\"} "
                     "%" PRIu64 "\n",
@@ -195,6 +208,7 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
                     e->name, node_metrics->name, g->name, e->value);
         }
     }
+    pthread_mutex_unlock(&node_metrics->lock);
 }
 
 static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name)
@@ -216,7 +230,7 @@ static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name)
     return false;
 }
 
-static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
+static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
                                  FILE *stream)
 {
     neu_metric_entry_t * e = NULL, *r = NULL;
@@ -261,11 +275,18 @@ static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
                         r->help, r->name, neu_metric_type_str(r->type));
             }
 
+            pthread_mutex_lock(&n->lock);
             HASH_FIND_STR(n->entries, r->name, e);
             if (e) {
+                if (neu_metric_type_is_rolling_counter(e->type)) {
+                    // force clean stale value
+                    e->value =
+                        neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
+                }
                 fprintf(stream, "%s{node=\"%s\"} %" PRIu64 "\n", e->name,
                         n->name, e->value);
 
+                pthread_mutex_unlock(&n->lock);
                 continue;
             }
 
@@ -273,11 +294,17 @@ static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
             {
                 HASH_FIND_STR(g->entries, r->name, e);
                 if (e) {
+                    if (neu_metric_type_is_rolling_counter(e->type)) {
+                        // force clean stale value
+                        e->value = neu_rolling_counter_inc(e->rcnt,
+                                                           global_timestamp, 0);
+                    }
                     fprintf(stream,
                             "%s{node=\"%s\",group=\"%s\"} %" PRIu64 "\n",
                             e->name, n->name, g->name, e->value);
                 }
             }
+            pthread_mutex_unlock(&n->lock);
         }
     }
 }
@@ -289,7 +316,7 @@ struct context {
     const char *node;
 };
 
-static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx)
+static void gen_node_metrics(neu_metrics_t *metrics, struct context *ctx)
 {
     if (ctx->node[0]) {
         neu_node_metrics_t *n = NULL;

src/adapter/adapter.c

@@ -389,22 +389,19 @@ static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
                                    uint64_t init)
 {
     if (NULL == adapter->metrics) {
-        adapter->metrics = calloc(1, sizeof(*adapter->metrics));
+        adapter->metrics =
+            neu_node_metrics_new(adapter, adapter->module->type, adapter->name);
         if (NULL == adapter->metrics) {
             return -1;
         }
-        adapter->metrics->type    = adapter->module->type;
-        adapter->metrics->name    = adapter->name;
-        adapter->metrics->adapter = adapter;
         neu_metrics_add_node(adapter);
     }
 
-    if (0 > neu_metric_entries_add(&adapter->metrics->entries, name, help, type,
-                                   init)) {
+    if (0 >
+        neu_node_metrics_add(adapter->metrics, NULL, name, help, type, init)) {
         return -1;
     }
 
-    neu_metrics_register_entry(name, help, type);
     return 0;
 }
 
@@ -412,48 +409,11 @@ static int adapter_update_metric(neu_adapter_t *adapter,
                                  const char *metric_name, uint64_t n,
                                  const char *group)
 {
-    neu_metric_entry_t *entry = NULL;
     if (NULL == adapter->metrics) {
         return -1;
     }
 
-    if (NULL == group) {
-        HASH_FIND_STR(adapter->metrics->entries, metric_name, entry);
-    } else if (NULL != adapter->metrics->group_metrics) {
-        neu_group_metrics_t *g = NULL;
-        HASH_FIND_STR(adapter->metrics->group_metrics, group, g);
-        if (NULL != g) {
-            HASH_FIND_STR(g->entries, metric_name, entry);
-        }
-    }
-
-    if (NULL == entry) {
-        return -1;
-    }
-
-    if (NEU_METRIC_TYPE_COUNTER == entry->type) {
-        entry->value += n;
-    } else {
-        entry->value = n;
-    }
-
-    return 0;
-}
-
-static void adapter_reset_metrics(neu_adapter_t *adapter)
-{
-    neu_metric_entry_t *entry = NULL;
-    if (NULL == adapter->metrics) {
-        return;
-    }
-
-    HASH_LOOP(hh, adapter->metrics->entries, entry) { entry->value = 0; }
-
-    neu_group_metrics_t *g = NULL;
-    HASH_LOOP(hh, adapter->metrics->group_metrics, g)
-    {
-        HASH_LOOP(hh, g->entries, entry) { entry->value = 0; }
-    }
+    return neu_node_metrics_update(adapter->metrics, group, metric_name, n);
 }
 
 static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
@@ -546,6 +506,20 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
         strcpy(pheader->receiver, cmd->node);
         break;
     }
+    case NEU_REQ_PRGFILE_UPLOAD: {
+        neu_req_prgfile_upload_t *cmd = (neu_req_prgfile_upload_t *) data;
+        strcpy(pheader->receiver, cmd->driver);
+        break;
+    }
+    case NEU_REQ_PRGFILE_PROCESS: {
+        neu_req_prgfile_process_t *cmd = (neu_req_prgfile_process_t *) data;
+        strcpy(pheader->receiver, cmd->driver);
+        break;
+    }
+    case NEU_RESP_PRGFILE_PROCESS: {
+        strcpy(pheader->receiver, header.sender);
+        break;
+    }
     default:
         break;
     }
@@ -729,6 +703,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_GET_GROUP:
     case NEU_RESP_ERROR:
     case NEU_REQRESP_NODES_STATE:
+    case NEU_REQ_PRGFILE_PROCESS:
+    case NEU_RESP_PRGFILE_PROCESS:
         adapter->module->intf_funs->request(
             adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
         neu_msg_free(msg);
@@ -838,11 +814,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_get_node_state_t *resp =
             (neu_resp_get_node_state_t *) &header[1];
 
-        neu_metric_entry_t *e = NULL;
         if (NULL != adapter->metrics) {
+            pthread_mutex_lock(&adapter->metrics->lock);
+            neu_metric_entry_t *e = NULL;
             HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e);
+            resp->rtt = NULL != e ? e->value : 0;
+            pthread_mutex_unlock(&adapter->metrics->lock);
         }
-        resp->rtt    = NULL != e ? e->value : 0;
         resp->state  = neu_adapter_get_state(adapter);
         header->type = NEU_RESP_GET_NODE_STATE;
         neu_msg_exchange(header);
@@ -950,7 +928,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             error.error = neu_adapter_driver_del_group(
                 (neu_adapter_driver_t *) adapter, cmd->group);
         } else {
-            error.error = NEU_ERR_GROUP_NOT_ALLOW;
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+            neu_msg_free(msg);
+            break;
         }
 
         if (error.error == NEU_ERR_SUCCESS) {
@@ -1083,12 +1064,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_add_tag_t  resp = { 0 };
         if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
             resp.error = NEU_ERR_GROUP_NOT_ALLOW;
-        } else if (neu_adapter_driver_new_group_count(
-                       (neu_adapter_driver_t *) adapter, cmd) +
-                       neu_adapter_driver_group_count(
-                           (neu_adapter_driver_t *) adapter) >
-                   NEU_GROUP_MAX_PER_NODE) {
-            resp.error = NEU_ERR_GROUP_MAX_GROUPS;
         } else {
             if (validate_gtags(adapter, cmd, &resp) == 0 &&
                 try_add_gtags(adapter, cmd, &resp) == 0 &&
@@ -1256,7 +1231,11 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         break;
     }
-
+    case NEU_REQ_PRGFILE_UPLOAD: {
+        adapter->module->intf_funs->request(
+            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        break;
+    }
     default:
         nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name,
                   neu_reqresp_type_string(header->type));
@@ -1270,22 +1249,30 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 int validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
                    neu_resp_add_tag_t *resp)
 {
+    neu_adapter_driver_t *driver_adapter = (neu_adapter_driver_t *) adapter;
+    if (neu_adapter_driver_new_group_count(driver_adapter, cmd) +
+            neu_adapter_driver_group_count(driver_adapter) >
+        NEU_GROUP_MAX_PER_NODE) {
+        resp->error = NEU_ERR_GROUP_MAX_GROUPS;
+        resp->index = 0;
+        return NEU_ERR_GROUP_MAX_GROUPS;
+    }
+
     for (int group_index = 0; group_index < cmd->n_group; group_index++) {
-        for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
-             tag_index++) {
+        neu_gdatatag_t *current_group = &cmd->groups[group_index];
+        for (int tag_index = 0; tag_index < current_group->n_tag; tag_index++) {
             int validation_result = neu_adapter_driver_validate_tag(
-                (neu_adapter_driver_t *) adapter,
-                cmd->groups[group_index].group,
-                &cmd->groups[group_index].tags[tag_index]);
-            if (validation_result == 0) {
-                resp->index += 1;
-            } else {
+                driver_adapter, current_group->group,
+                &current_group->tags[tag_index]);
+            if (validation_result != 0) {
                 resp->error = validation_result;
                 resp->index = 0;
                 return validation_result;
             }
+            resp->index += 1;
         }
     }
+
     return 0;
 }
 
@@ -1325,8 +1312,8 @@ int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
             if (add_tag_result != 0) {
                 for (int added_group_index = 0; added_group_index < group_index;
                      added_group_index++) {
-                    for (int added_tag_index = 0; added_tag_index <=
-                         cmd->groups[added_group_index].n_tag;
+                    for (int added_tag_index = 0;
+                         added_tag_index < cmd->groups[added_group_index].n_tag;
                          added_tag_index++) {
                         neu_adapter_driver_del_tag(
                             (neu_adapter_driver_t *) adapter,
@@ -1343,12 +1330,17 @@ int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
                         cmd->groups[group_index].group,
                         cmd->groups[group_index].tags[added_tag_index].name);
                 }
-                for (int groups_count = 0; groups_count < cmd->n_group;
-                     groups_count++) {
+
+                neu_adapter_driver_try_del_tag((neu_adapter_driver_t *) adapter,
+                                               cmd->groups[group_index].n_tag -
+                                                   tag_index);
+
+                for (++group_index; group_index < cmd->n_group; group_index++) {
                     neu_adapter_driver_try_del_tag(
                         (neu_adapter_driver_t *) adapter,
-                        cmd->groups[groups_count].n_tag);
+                        cmd->groups[group_index].n_tag);
                 }
+
                 resp->index = 0;
                 resp->error = add_tag_result;
                 return add_tag_result;
@@ -1368,11 +1360,6 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
 
     if (NULL != adapter->metrics) {
         neu_metrics_del_node(adapter);
-        neu_metric_entry_t *e = NULL;
-        HASH_LOOP(hh, adapter->metrics->entries, e)
-        {
-            neu_metrics_unregister_entry(e->name);
-        }
         neu_node_metrics_free(adapter->metrics);
     }
 
@@ -1450,6 +1437,10 @@ int neu_adapter_start(neu_adapter_t *adapter)
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
         adapter_storage_state(adapter->name, adapter->state);
+        if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
+            neu_adapter_driver_start_group_timer(
+                (neu_adapter_driver_t *) adapter);
+        }
     }
 
     return error;
@@ -1488,7 +1479,11 @@ int neu_adapter_stop(neu_adapter_t *adapter)
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
         adapter_storage_state(adapter->name, adapter->state);
-        adapter_reset_metrics(adapter);
+        if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
+            neu_adapter_driver_stop_group_timer(
+                (neu_adapter_driver_t *) adapter);
+        }
+        neu_adapter_reset_metrics(adapter);
     }
 
     return error;
@@ -1567,110 +1562,43 @@ int neu_adapter_register_group_metric(neu_adapter_t *adapter,
                                       const char *help, neu_metric_type_e type,
                                       uint64_t init)
 {
-    neu_group_metrics_t *group_metrics = NULL;
-
     if (NULL == adapter->metrics) {
         return -1;
     }
 
-    if (0 > neu_metrics_register_entry(name, help, type)) {
-        return -1;
-    }
-
-    HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
-    if (NULL == group_metrics) {
-        group_metrics = calloc(1, sizeof(*group_metrics));
-        if (NULL == group_metrics) {
-            return -1;
-        }
-        group_metrics->name = strdup(group_name);
-        if (NULL == group_metrics->name) {
-            free(group_metrics);
-            return -1;
-        }
-        HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
-    }
-
-    if (0 > neu_metric_entries_add(&group_metrics->entries, name, help, type,
-                                   init)) {
-        return -1;
-    }
-
-    return 0;
+    return neu_node_metrics_add(adapter->metrics, group_name, name, help, type,
+                                init);
 }
 
 int neu_adapter_update_group_metric(neu_adapter_t *adapter,
                                     const char *   group_name,
                                     const char *metric_name, uint64_t n)
 {
-    neu_metric_entry_t * entry         = NULL;
-    neu_group_metrics_t *group_metrics = NULL;
-
     if (NULL == adapter->metrics) {
         return -1;
     }
 
-    HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
-    if (NULL == group_metrics) {
-        return -1;
-    }
-
-    HASH_FIND_STR(group_metrics->entries, metric_name, entry);
-    if (NULL == entry) {
-        return -1;
-    }
-
-    if (NEU_METRIC_TYPE_COUNTER == entry->type) {
-        entry->value += n;
-    } else {
-        entry->value = n;
-    }
-
-    return 0;
+    return neu_node_metrics_update(adapter->metrics, group_name, metric_name,
+                                   n);
 }
 
 int neu_adapter_metric_update_group_name(neu_adapter_t *adapter,
                                          const char *   group_name,
                                          const char *   new_group_name)
 {
-    neu_group_metrics_t *group_metrics = NULL;
-
     if (NULL == adapter->metrics) {
         return -1;
     }
 
-    HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
-    if (NULL == group_metrics) {
-        return -1;
-    }
-
-    char *name = strdup(new_group_name);
-    if (NULL == name) {
-        return -1;
-    }
-
-    HASH_DEL(adapter->metrics->group_metrics, group_metrics);
-    free(group_metrics->name);
-    group_metrics->name = name;
-    HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
-
-    return 0;
+    return neu_node_metrics_update_group(adapter->metrics, group_name,
+                                         new_group_name);
 }
 
 void neu_adapter_del_group_metrics(neu_adapter_t *adapter,
                                    const char *   group_name)
 {
-    if (NULL == adapter->metrics) {
-        return;
-    }
-
-    neu_group_metrics_t *gm = NULL;
-    HASH_FIND_STR(adapter->metrics->group_metrics, group_name, gm);
-    if (NULL != gm) {
-        HASH_DEL(adapter->metrics->group_metrics, gm);
-        neu_metric_entry_t *e = NULL;
-        HASH_LOOP(hh, gm->entries, e) { neu_metrics_unregister_entry(e->name); }
-        neu_group_metrics_free(gm);
+    if (NULL != adapter->metrics) {
+        neu_node_metrics_del_group(adapter->metrics, group_name);
     }
 }
 

src/adapter/adapter_internal.h

@@ -100,6 +100,13 @@ neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter);
 
 int neu_adapter_validate_tag(neu_adapter_t *adapter, neu_datatag_t *tag);
 
+static inline void neu_adapter_reset_metrics(neu_adapter_t *adapter)
+{
+    if (NULL != adapter->metrics) {
+        neu_node_metrics_reset(adapter->metrics);
+    }
+}
+
 int  neu_adapter_register_group_metric(neu_adapter_t *adapter,
                                        const char *group_name, const char *name,
                                        const char *help, neu_metric_type_e type,

src/adapter/driver/cache.c

@@ -157,6 +157,8 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
             case NEU_TYPE_BIT:
             case NEU_TYPE_BOOL:
             case NEU_TYPE_STRING:
+            case NEU_TYPE_TIME:
+            case NEU_TYPE_DATA_AND_TIME:
             case NEU_TYPE_WORD:
             case NEU_TYPE_DWORD:
             case NEU_TYPE_LWORD:
@@ -299,6 +301,8 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;
@@ -381,6 +385,8 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;
@@ -461,6 +467,8 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;
@@ -539,6 +547,8 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;

src/adapter/driver/driver.c

@@ -90,6 +90,7 @@ static void read_group(int64_t timestamp, int64_t timeout,
                        neu_driver_cache_t *cache, const char *group,
                        UT_array *tags, UT_array *tag_values);
 static void read_report_group(int64_t timestamp, int64_t timeout,
+                              neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values);
 static void update(neu_adapter_t *adapter, const char *group, const char *tag,
@@ -101,8 +102,11 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
                              const char *tag, neu_dvalue_t value,
                              neu_tag_meta_t *metas, int n_meta);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
-static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
-static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
+static group_t *   find_group(neu_adapter_driver_t *driver, const char *name);
+static void        store_write_tag(group_t *group, to_be_write_tag_t *tag);
+static inline void start_group_timer(neu_adapter_driver_t *driver,
+                                     group_t *             grp);
+static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp);
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -167,7 +171,7 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
         update_metric(&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
                       NEU_TYPE_ERROR == value.type, NULL);
     }
-    nlog_info(
+    nlog_debug(
         "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64
         " n_meta: %d",
         driver->adapter.name, group, tag, neu_type_string(value.type),
@@ -200,19 +204,19 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     if (first == NULL ||
         !neu_tag_attribute_test(first, NEU_ATTRIBUTE_SUBSCRIBE)) {
         utarray_free(tags);
-        nlog_info("update immediately, driver: %s, "
-                  "group: %s, tag: %s, type: %s, "
-                  "timestamp: %" PRId64,
-                  driver->adapter.name, group, tag, neu_type_string(value.type),
-                  global_timestamp);
+        nlog_debug("update immediately, driver: %s, "
+                   "group: %s, tag: %s, type: %s, "
+                   "timestamp: %" PRId64,
+                   driver->adapter.name, group, tag,
+                   neu_type_string(value.type), global_timestamp);
         return;
     }
 
-    nlog_info("update and report immediately, driver: %s, "
-              "group: %s, tag: %s, type: %s, "
-              "timestamp: %" PRId64,
-              driver->adapter.name, group, tag, neu_type_string(value.type),
-              global_timestamp);
+    nlog_debug("update and report immediately, driver: %s, "
+               "group: %s, tag: %s, type: %s, "
+               "timestamp: %" PRId64,
+               driver->adapter.name, group, tag, neu_type_string(value.type),
+               global_timestamp);
 
     neu_reqresp_head_t header = {
         .type = NEU_REQRESP_TRANS_DATA,
@@ -224,8 +228,9 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     data->group  = strdup(group);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
 
-    read_report_group(global_timestamp, 0, driver->cache, group, tags,
-                      data->tags);
+    read_report_group(global_timestamp, 0,
+                      neu_adapter_get_tag_cache_type(&driver->adapter),
+                      driver->cache, group, tags, data->tags);
 
     if (utarray_len(data->tags) > 0) {
         group_t *find = NULL;
@@ -334,9 +339,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         HASH_DEL(driver->groups, el);
 
         neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group));
-        neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
-        neu_event_del_timer(driver->driver_events, el->read);
-        neu_event_del_timer(driver->driver_events, el->write);
+        stop_group_timer(driver, el);
         if (el->grp.group_free != NULL) {
             el->grp.group_free(&el->grp);
         }
@@ -359,34 +362,67 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
     return 0;
 }
 
+static inline void start_group_timer(neu_adapter_driver_t *driver, group_t *grp)
+{
+    uint32_t interval = neu_group_get_interval(grp->group);
+
+    neu_event_timer_param_t param = {
+        .second      = interval / 1000,
+        .millisecond = interval % 1000,
+        .usr_data    = (void *) grp,
+        .type        = NEU_EVENT_TIMER_NOBLOCK,
+    };
+
+    param.type = driver->adapter.module->timer_type;
+    param.cb   = read_callback;
+    grp->read  = neu_event_add_timer(driver->driver_events, param);
+
+    struct timespec t1 = {
+        .tv_sec  = 0,
+        .tv_nsec = 1000 * 1000 * 20,
+    };
+    struct timespec t2 = { 0 };
+    nanosleep(&t1, &t2);
+
+    param.type  = NEU_EVENT_TIMER_NOBLOCK;
+    param.cb    = report_callback;
+    grp->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+
+    param.second      = 0;
+    param.millisecond = 3;
+    param.cb          = write_callback;
+    grp->write        = neu_event_add_timer(driver->driver_events, param);
+}
+
 void neu_adapter_driver_start_group_timer(neu_adapter_driver_t *driver)
 {
     group_t *el = NULL, *tmp = NULL;
 
     HASH_ITER(hh, driver->groups, el, tmp)
     {
-        uint32_t                interval = neu_group_get_interval(el->group);
-        neu_event_timer_param_t param    = {
-            .second      = interval / 1000,
-            .millisecond = interval % 1000,
-            .usr_data    = el,
-            .type        = NEU_EVENT_TIMER_NOBLOCK,
-        };
+        start_group_timer(driver, el);
+        neu_adapter_update_group_metric(
+            &driver->adapter, neu_group_get_name(el->group),
+            NEU_METRIC_GROUP_TAGS_TOTAL, neu_group_tag_size(el->group));
+    }
 
-        param.type = driver->adapter.module->timer_type;
-        param.cb   = read_callback;
-        el->read   = neu_event_add_timer(driver->driver_events, param);
+    driver->adapter.cb_funs.update_metric(
+        &driver->adapter, NEU_METRIC_TAGS_TOTAL, driver->tag_cnt, NULL);
+}
 
-        struct timespec t1 = {
-            .tv_sec  = 0,
-            .tv_nsec = 1000 * 1000 * 20,
-        };
-        struct timespec t2 = { 0 };
-        nanosleep(&t1, &t2);
-
-        param.type = NEU_EVENT_TIMER_NOBLOCK;
-        param.cb   = report_callback;
-        el->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp)
+{
+    if (grp->report) {
+        neu_adapter_del_timer((neu_adapter_t *) driver, grp->report);
+        grp->report = NULL;
+    }
+    if (grp->read) {
+        neu_event_del_timer(driver->driver_events, grp->read);
+        grp->read = NULL;
+    }
+    if (grp->write) {
+        neu_event_del_timer(driver->driver_events, grp->write);
+        grp->write = NULL;
     }
 }
 
@@ -394,13 +430,7 @@ void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver)
 {
     group_t *el = NULL, *tmp = NULL;
 
-    HASH_ITER(hh, driver->groups, el, tmp)
-    {
-        neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
-        el->report = NULL;
-        neu_event_del_timer(driver->driver_events, el->read);
-        el->read = NULL;
-    }
+    HASH_ITER(hh, driver->groups, el, tmp) { stop_group_timer(driver, el); }
 }
 
 void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
@@ -482,6 +512,8 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
     switch (tag->type) {
     case NEU_TYPE_BOOL:
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
     case NEU_TYPE_BYTES:
         break;
     case NEU_TYPE_BIT:
@@ -898,13 +930,6 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
     if (find == NULL) {
         find = calloc(1, sizeof(group_t));
 
-        neu_event_timer_param_t param = {
-            .second      = interval / 1000,
-            .millisecond = interval % 1000,
-            .usr_data    = (void *) find,
-            .type        = NEU_EVENT_TIMER_NOBLOCK,
-        };
-
         pthread_mutex_init(&find->wt_mtx, NULL);
         pthread_mutex_init(&find->apps_mtx, NULL);
 
@@ -918,25 +943,9 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
         neu_group_split_static_tags(find->group, &find->static_tags,
                                     &find->grp.tags);
 
-        param.type = driver->adapter.module->timer_type;
-        param.cb   = read_callback;
-        find->read = neu_event_add_timer(driver->driver_events, param);
-
-        struct timespec t1 = {
-            .tv_sec  = 0,
-            .tv_nsec = 1000 * 1000 * 20,
-        };
-        struct timespec t2 = { 0 };
-        nanosleep(&t1, &t2);
-
-        param.type   = NEU_EVENT_TIMER_NOBLOCK;
-        param.cb     = report_callback;
-        find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
-
-        param.second      = 0;
-        param.millisecond = 3;
-        param.cb          = write_callback;
-        find->write       = neu_event_add_timer(driver->driver_events, param);
+        if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+            start_group_timer(driver, find);
+        }
 
         REGISTER_GROUP_METRIC(&driver->adapter, find->name,
                               NEU_METRIC_GROUP_TAGS_TOTAL,
@@ -983,8 +992,9 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
     }
 
     // stop the timer first to avoid race condition
-    neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
-    neu_event_del_timer(driver->driver_events, find->read);
+    if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+        stop_group_timer(driver, find);
+    }
 
     // a diminutive value should keep the interval untouched
     if (interval < NEU_GROUP_INTERVAL_LIMIT) {
@@ -1013,31 +1023,12 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
         }
     }
 
-    neu_event_timer_param_t param = {
-        .second      = interval / 1000,
-        .millisecond = interval % 1000,
-        .usr_data    = (void *) find,
-        .type        = NEU_EVENT_TIMER_NOBLOCK,
-    };
-
     neu_group_set_interval(find->group, interval);
 
     // restore the timers
-
-    param.type = driver->adapter.module->timer_type;
-    param.cb   = read_callback;
-    find->read = neu_event_add_timer(driver->driver_events, param);
-
-    struct timespec t1 = {
-        .tv_sec  = 0,
-        .tv_nsec = 1000 * 1000 * 20,
-    };
-    struct timespec t2 = { 0 };
-    nanosleep(&t1, &t2);
-
-    param.type   = NEU_EVENT_TIMER_NOBLOCK;
-    param.cb     = report_callback;
-    find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+    if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+        start_group_timer(driver, find);
+    }
 
     return ret;
 }
@@ -1053,9 +1044,10 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(find->group));
 
-        neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
-        neu_event_del_timer(driver->driver_events, find->read);
-        neu_event_del_timer(driver->driver_events, find->write);
+        if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+            stop_group_timer(driver, find);
+        }
+
         if (find->grp.group_free != NULL) {
             find->grp.group_free(&find->grp);
         }
@@ -1510,6 +1502,7 @@ static int report_callback(void *usr_data)
     read_report_group(global_timestamp,
                       neu_group_get_interval(group->group) *
                           NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
+                      neu_adapter_get_tag_cache_type(&group->driver->adapter),
                       group->driver->cache, group->name, tags, data->tags);
 
     if (utarray_len(data->tags) > 0) {
@@ -1683,6 +1676,7 @@ static int read_callback(void *usr_data)
 }
 
 static void read_report_group(int64_t timestamp, int64_t timeout,
+                              neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values)
 {
@@ -1775,7 +1769,8 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
             break;
         }
 
-        if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
+        if (cache_type != NEU_TAG_CACHE_TYPE_NEVER &&
+            !neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
             (timestamp - value.timestamp) > timeout && timeout > 0) {
             if (value.value.type == NEU_TYPE_PTR) {
                 free(value.value.value.ptr.ptr);

src/base/metrics.c

@@ -212,16 +212,17 @@ static unsigned cpu_usage()
     return (double) work / total * 100.0 * sysconf(_SC_NPROCESSORS_CONF);
 }
 
-static bool has_core_dumps()
+static bool has_core_dump_in_dir(const char *dir, const char *prefix)
 {
-    DIR *dp = opendir("core");
-    if (NULL == dp) {
+    DIR *dp = opendir(dir);
+    if (dp == NULL) {
         return false;
     }
 
-    bool found = false;
-    for (struct dirent *de = NULL; NULL != (de = readdir(dp));) {
-        if (0 == strncmp("core", de->d_name, 4)) {
+    struct dirent *de;
+    bool           found = false;
+    while ((de = readdir(dp)) != NULL) {
+        if (strncmp(prefix, de->d_name, strlen(prefix)) == 0) {
             found = true;
             break;
         }
@@ -231,6 +232,46 @@ static bool has_core_dumps()
     return found;
 }
 
+static char *get_core_dir(const char *core_pattern)
+{
+    char *last_slash = strrchr(core_pattern, '/');
+    if (last_slash != NULL) {
+        static char core_dir[256];
+        ptrdiff_t   path_length = last_slash - core_pattern + 1;
+        strncpy(core_dir, core_pattern, path_length);
+        core_dir[path_length] = '\0';
+        return core_dir;
+    }
+    return NULL;
+}
+
+static bool has_core_dumps()
+{
+    if (has_core_dump_in_dir("core", "core-neuron")) {
+        return true;
+    }
+
+    FILE *fp = fopen("/proc/sys/kernel/core_pattern", "r");
+    if (fp == NULL) {
+        return false;
+    }
+
+    char core_pattern[256];
+    if (fgets(core_pattern, sizeof(core_pattern), fp) == NULL) {
+        fclose(fp);
+        return false;
+    }
+    fclose(fp);
+
+    char *core_dir =
+        (core_pattern[0] == '|') ? "/var/crash/" : get_core_dir(core_pattern);
+    if (core_dir == NULL) {
+        return false;
+    }
+
+    return has_core_dump_in_dir(core_dir, "core-neuron");
+}
+
 static inline void metrics_unregister_entry(const char *name)
 {
     neu_metric_entry_t *e = NULL;
@@ -262,10 +303,19 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
         return -1;
     }
 
-    entry->name  = name;
-    entry->type  = type;
-    entry->help  = help;
-    entry->value = init;
+    if (NEU_METRIC_TYPE_ROLLING_COUNTER == type) {
+        // only allocate rolling counter for nonzero time span
+        if (init > 0 && NULL == (entry->rcnt = neu_rolling_counter_new(init))) {
+            free(entry);
+            return -1;
+        }
+    } else {
+        entry->value = init;
+    }
+
+    entry->name = name;
+    entry->type = type;
+    entry->help = help;
     HASH_ADD_STR(*entries, name, entry);
     return 0;
 }
@@ -302,9 +352,11 @@ int neu_metrics_register_entry(const char *name, const char *help,
     neu_metric_entry_t *e  = NULL;
 
     pthread_rwlock_wrlock(&g_metrics_mtx_);
+    // use `value` field as reference counter, initialize to zero
+    // and we don't need to allocate rolling counter for register entries
     rv = neu_metric_entries_add(&g_metrics_.registered_metrics, name, help,
-                                type, 1);
-    if (1 == rv) {
+                                type, 0);
+    if (-1 != rv) {
         HASH_FIND_STR(g_metrics_.registered_metrics, name, e);
         ++e->value;
         rv = 0;

src/base/msg_internal.h

@@ -115,7 +115,10 @@ extern "C" {
     XX(NEU_REQ_UPDATE_NDRIVER_TAG_INFO, neu_req_update_ndriver_tag_info_t)   \
     XX(NEU_REQ_GET_NDRIVER_TAGS, neu_req_get_ndriver_tags_t)                 \
     XX(NEU_RESP_GET_NDRIVER_TAGS, neu_resp_get_ndriver_tags_t)               \
-    XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t)
+    XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t)                 \
+    XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t)                     \
+    XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t)                   \
+    XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t)
 
 static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
 {

src/base/tag.c

@@ -161,19 +161,24 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
 
             switch (t) {
             case 'H':
-                option->string.type = NEU_DATATAG_STRING_TYPE_H;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_H;
+                option->string.is_default = false;
                 break;
             case 'L':
-                option->string.type = NEU_DATATAG_STRING_TYPE_L;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_L;
+                option->string.is_default = false;
                 break;
             case 'D':
-                option->string.type = NEU_DATATAG_STRING_TYPE_D;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_D;
+                option->string.is_default = false;
                 break;
             case 'E':
-                option->string.type = NEU_DATATAG_STRING_TYPE_D;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_E;
+                option->string.is_default = false;
                 break;
             default:
-                option->string.type = NEU_DATATAG_STRING_TYPE_H;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_H;
+                option->string.is_default = true;
                 break;
             }
 
@@ -210,10 +215,13 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
     }
     case NEU_TYPE_FLOAT:
     case NEU_TYPE_INT32:
-    case NEU_TYPE_UINT32: {
+    case NEU_TYPE_UINT32:
+    case NEU_TYPE_DATA_AND_TIME:
+    case NEU_TYPE_TIME: {
         char *op = find_last_character(datatag->address, '#');
 
-        option->value32.endian = NEU_DATATAG_ENDIAN_LL32;
+        option->value32.endian     = NEU_DATATAG_ENDIAN_LL32;
+        option->value32.is_default = true;
         if (op != NULL) {
             char e1 = 0;
             char e2 = 0;
@@ -221,16 +229,20 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
 
             if (n == 2) {
                 if (e1 == 'B' && e2 == 'B') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_BB32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_BB32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'B' && e2 == 'L') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_BL32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_BL32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'L' && e2 == 'L') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_LL32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_LL32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'L' && e2 == 'B') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_LB32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_LB32;
+                    option->value32.is_default = false;
                 }
             }
         }
@@ -523,6 +535,8 @@ int neu_tag_get_static_value_json(neu_datatag_t *tag, neu_json_type_e *t,
         v->val_double = value->d64;
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         *t         = NEU_JSON_STR;
         v->val_str = value->str;
         break;
@@ -625,7 +639,8 @@ int neu_tag_set_static_value_json(neu_datatag_t *tag, neu_json_type_e t,
         break;
     }
     case NEU_JSON_STR: {
-        if (NEU_TYPE_STRING == tag->type) {
+        if (NEU_TYPE_STRING == tag->type || NEU_TYPE_TIME == tag->type ||
+            NEU_TYPE_DATA_AND_TIME == tag->type) {
             snprintf(value.str, sizeof(value.str), "%s", v->val_str);
         } else {
             rv = -1;
@@ -695,6 +710,8 @@ char *neu_tag_dump_static_value(const neu_datatag_t *tag)
         jval = json_real(value.d64);
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         jval = json_string(value.str);
         break;
     default:

src/connection/connection.c

@@ -1290,6 +1290,12 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
                 break;
             } else {
                 pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
@@ -1320,10 +1326,18 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
                 neu_conn_tcp_server_close_client(conn, fd);
                 break;
             } else {
+                pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
                                              conn->offset);
+                pthread_mutex_unlock(&conn->mtx);
             }
         }
     }

src/core/manager.c

@@ -535,6 +535,35 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
                 break;
             }
 
+            if (!neu_plugin_manager_create_instance_by_lib_name(
+                    manager->plugin_manager, cmd->library, &ins)) {
+                nlog_warn("library %s not instance", cmd->library);
+                free(cmd->so_file);
+                free(cmd->schema_file);
+                free(so_tmp_path);
+                header->type = NEU_RESP_ERROR;
+                e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
+                strcpy(header->receiver, header->sender);
+                reply(manager, header, &e);
+                break;
+            }
+
+            if (strcmp(module_name, ins.module->module_name) != 0) {
+                nlog_warn("library %s module name mismatch!", cmd->library);
+                neu_plugin_manager_destroy_instance(manager->plugin_manager,
+                                                    &ins);
+                free(cmd->so_file);
+                free(cmd->schema_file);
+                free(so_tmp_path);
+                header->type = NEU_RESP_ERROR;
+                e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
+                strcpy(header->receiver, header->sender);
+                reply(manager, header, &e);
+                break;
+            }
+
+            neu_plugin_manager_destroy_instance(manager->plugin_manager, &ins);
+
             if (!neu_plugin_manager_exists(manager->plugin_manager,
                                            module_name)) {
 
@@ -1267,7 +1296,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         break;
     }
-
+    case NEU_REQ_PRGFILE_PROCESS:
+    case NEU_REQ_PRGFILE_UPLOAD:
     case NEU_REQ_GET_NODE_STATE: {
         if (neu_node_manager_find(manager->node_manager, header->receiver) ==
             NULL) {
@@ -1370,9 +1400,24 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_msg_exchange(header);
             reply(manager, header, &e);
         } else {
+            UT_array *apps = neu_subscribe_manager_find(
+                manager->subscribe_manager, cmd->driver, cmd->group);
+
             forward_msg(manager, header, header->receiver);
             neu_subscribe_manager_remove(manager->subscribe_manager,
                                          cmd->driver, cmd->group);
+
+            if (NULL == apps) {
+                break;
+            }
+
+            // notify app node about group deletion
+            utarray_foreach(apps, neu_app_subscribe_t *, app)
+            {
+                forward_msg_copy(manager, header, app->app_name);
+            }
+
+            utarray_free(apps);
         }
         break;
     }
@@ -1465,6 +1510,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_RESP_GET_NODE_SETTING:
     case NEU_RESP_ERROR:
     case NEU_RESP_READ_GROUP:
+    case NEU_RESP_PRGFILE_PROCESS:
         forward_msg(manager, header, header->receiver);
         break;
 

src/core/manager_internal.c

@@ -698,7 +698,6 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
     // this is not an elegant solution due to the current architecture
     if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
         neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
-        neu_json_decode((char *) params, 1, &elem);
         neu_parse_param(params, NULL, 1, &elem);
         if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
             free(elem.v.val_str);

src/core/plugin_manager.c

@@ -490,6 +490,38 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
     return true;
 }
 
+bool neu_plugin_manager_create_instance_by_lib_name(
+    neu_plugin_manager_t *mgr, const char *lib_name,
+    neu_plugin_instance_t *instance)
+{
+    (void) mgr;
+    char lib_paths[3][64] = { 0 };
+    snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
+    snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
+    snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
+
+    for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
+        char lib_path[256] = { 0 };
+
+        snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
+                 lib_name);
+        instance->handle = dlopen(lib_path, RTLD_NOW);
+
+        if (instance->handle == NULL) {
+            continue;
+        }
+
+        instance->module = (neu_plugin_module_t *) dlsym(instance->handle,
+                                                         "neu_plugin_module");
+        assert(instance->module != NULL);
+        return true;
+    }
+
+    assert(instance->handle != NULL);
+
+    return false;
+}
+
 bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
                                        const char *          library)
 {

src/core/plugin_manager.h

@@ -68,6 +68,10 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
                                                 neu_plugin_instance_t *instance,
                                                 int *                  error);
 
+bool neu_plugin_manager_create_instance_by_lib_name(
+    neu_plugin_manager_t *mgr, const char *lib_name,
+    neu_plugin_instance_t *instance);
+
 bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
                                        const char *          library);
 

src/main.c

@@ -125,8 +125,6 @@ static int neuron_run(const neu_cli_args_t *args)
     int           rv = 0;
 
     signal(SIGINT, sig_handler);
-    signal(SIGABRT, sig_handler);
-    signal(SIGSEGV, sig_handler);
     signal(SIGTERM, sig_handler);
     signal(SIGKILL, sig_handler);
 

src/parser/neu_json_rw.c

@@ -407,14 +407,19 @@ void neu_json_decode_write_free(neu_json_write_t *req)
 
 int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
 {
-    int                  ret      = 0;
-    void *               json_obj = NULL;
-    neu_json_read_req_t *req      = calloc(1, sizeof(neu_json_read_req_t));
-    if (req == NULL) {
+    int   ret      = 0;
+    void *json_obj = NULL;
+
+    json_obj = neu_json_decode_new(buf);
+    if (NULL == json_obj) {
         return -1;
     }
 
-    json_obj = neu_json_decode_new(buf);
+    neu_json_read_req_t *req = calloc(1, sizeof(neu_json_read_req_t));
+    if (req == NULL) {
+        neu_json_decode_free(json_obj);
+        return -1;
+    }
 
     neu_json_elem_t req_elems[] = {
         {
@@ -436,15 +441,6 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
             .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
         },
     };
-    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
-                                  req_elems);
-    if (ret != 0) {
-        goto error;
-    }
-
-    req->node  = req_elems[0].v.val_str;
-    req->group = req_elems[1].v.val_str;
-    req->sync  = req_elems[2].v.val_bool;
 
     neu_json_elem_t query_elems[] = {
         {
@@ -458,6 +454,17 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
             .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
         },
     };
+
+    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
+                                  req_elems);
+    if (ret != 0) {
+        goto error;
+    }
+
+    req->node  = req_elems[0].v.val_str;
+    req->group = req_elems[1].v.val_str;
+    req->sync  = req_elems[2].v.val_bool;
+
     if (req_elems[3].v.val_object) {
         ret = neu_json_decode_by_json(req_elems[3].v.val_object,
                                       NEU_JSON_ELEM_SIZE(query_elems),
@@ -579,6 +586,8 @@ void neu_json_metas_to_json(neu_tag_meta_t *metas, int n_meta,
                 json_tag->metas[k].value.val_bit = metas[k].value.value.u8;
                 break;
             case NEU_TYPE_STRING:
+            case NEU_TYPE_TIME:
+            case NEU_TYPE_DATA_AND_TIME:
                 json_tag->metas[k].t             = NEU_JSON_STR;
                 json_tag->metas[k].value.val_str = metas[k].value.value.str;
                 break;

src/parser/neu_json_tag.c

@@ -199,6 +199,8 @@ int neu_json_tag_check_type(neu_json_tag_t *tag)
             return NEU_JSON_BOOL == tag->t;
         case NEU_TYPE_BYTES:
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             return NEU_JSON_STR == tag->t &&
                 strlen(tag->value.val_str) <= NEU_VALUE_SIZE;
         }

src/utils/json.c

@@ -140,9 +140,15 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
     case NEU_JSON_INT:
         ele->v.val_int = json_integer_value(ob);
         break;
-    case NEU_JSON_STR:
-        ele->v.val_str = strdup(json_string_value(ob));
+    case NEU_JSON_STR: {
+        const char *str = json_string_value(ob);
+        if (str != NULL) {
+            ele->v.val_str = strdup(str);
+        } else {
+            return -1;
+        }
         break;
+    }
     case NEU_JSON_FLOAT:
         if (json_is_integer(ob)) {
             ele->v.val_float = (float) json_integer_value(ob);

tests/ft/app/test_mqtt.py

@@ -490,6 +490,36 @@ class TestMQTT:
         finally:
             api.update_group_check(DRIVER, new_name, GROUP)
 
+    @description(
+        given="MQTT node",
+        when="delete subscribed driver group and add back group",
+        then="broker should receive data on upload topic",
+    )
+    def test_mqtt_upload_when_group_deleted(self, mocker, conf_fmt_tags):
+        api.node_setting_check(NODE, conf_fmt_tags)
+        api.node_ctl(NODE, config.NEU_CTL_START)
+
+        # delete driver group
+        api.del_group_check(DRIVER, GROUP)
+        # add back driver group
+        api.add_group_check(DRIVER, GROUP, INTERVAL)
+        api.add_tags_check(DRIVER, GROUP, tags=[TAG0])
+        # add back subscription on new topic
+        topic = UPLOAD_TOPIC + "-new"
+        params = {"topic": topic}
+        api.subscribe_group_check(NODE, DRIVER, GROUP, params)
+
+        msg = mocker.get(topic, timeout=3)
+        try:
+            assert msg is not None
+            assert DRIVER == msg["node"]
+            assert GROUP == msg["group"]
+            assert TAG0["name"] in [tag["name"] for tag in msg["tags"]]
+        finally:
+            params["topic"] = UPLOAD_TOPIC
+            api.unsubscribe_group(NODE, DRIVER, GROUP)
+            api.subscribe_group_check(NODE, DRIVER, GROUP, params)
+
     @description(
         given="MQTT node and conf_cache",
         when="broken network connection restored",

tests/ft/metrics/test_metrics.py

@@ -92,8 +92,23 @@ class TestMetrics:
             "send_msg_errors_total": (0, {}),
             "recv_msgs_total": (0, {}),
             "cached_msgs": (0, {}),
+            "last_5s_trans_data_msgs": (0, {}),
+            "last_30s_trans_data_msgs": (0, {}),
+            "last_60s_trans_data_msgs": (0, {}),
+            "last_5s_send_bytes": (0, {}),
+            "last_30s_send_bytes": (0, {}),
+            "last_60s_send_bytes": (0, {}),
+            "last_5s_recv_bytes": (0, {}),
+            "last_30s_recv_bytes": (0, {}),
+            "last_60s_recv_bytes": (0, {}),
+            "last_5s_recv_msgs": (0, {}),
+            "last_30s_recv_msgs": (0, {}),
+            "last_60s_recv_msgs": (0, {}),
+            "last_60s_disconnections": (0, {}),
+            "last_600s_disconnections": (0, {}),
+            "last_1800s_disconnections": (0, {}),
         }
-        
+
         assert_metrics(resp.content.decode('utf-8'), expected_metrics)
 
     @description(given="neuron is started",

tests/ft/neuron/api.py

@@ -165,6 +165,7 @@ def read_tag(node, group, tag, sync=False):
     finally:
         return x[0]['value']
 
+
 def read_tag_error(node, group, tag, sync=False):
     response = read_tags(node, group, sync)
     assert 200 == response.status_code
@@ -202,6 +203,7 @@ def write_tag(node, group, tag, value):
 def write_tags(node, group, tag_values):
     return requests.post(url=config.BASE_URL + "/api/v2/write/tags", headers={"Authorization": config.default_jwt}, json={"node": node, "group": group, "tags": tag_values})
 
+
 def write_gtags(json):
     return requests.post(url=config.BASE_URL + "/api/v2/write/gtags", headers={"Authorization": config.default_jwt}, json=json)
 
@@ -296,54 +298,78 @@ def del_template_tags(tmpl, group, tags):
 def get_template_tags(tmpl, group, name):
     return requests.get(url=config.BASE_URL + "/api/v2/template/tag", headers={"Authorization": config.default_jwt}, params={"template": tmpl, "group": group, "name": name})
 
+
 @gen_check
 def subscribe_group(app, driver, group, params={}):
     return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
 
+
 @gen_check
 def subscribe_groups(app, groups):
     return requests.post(url=config.BASE_URL + "/api/v2/subscribes", headers={"Authorization": config.default_jwt}, json={"app": app, "groups": groups})
 
+
 @gen_check
 def unsubscribe_group(app, driver, group):
     return requests.delete(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})
 
+
 def get_metrics(category="global", node=""):
     return requests.get(url=config.BASE_URL + "/api/v2/metrics?category=" + category + "&node=" + node, headers={"Authorization": config.default_jwt})
 
+
 def get_global_config():
     return requests.get(url=config.BASE_URL + "/api/v2/global/config", headers={"Authorization": config.default_jwt})
 
+
 def put_global_config(json):
     return requests.put(url=config.BASE_URL + "/api/v2/global/config", headers={"Authorization": config.default_jwt}, json=json)
 
+
 def get_plugin_schema(plugin):
     return requests.get(url=config.BASE_URL + "/api/v2/schema?schema_name=" + plugin, headers={"Authorization": config.default_jwt})
 
+
 def get_plugin():
     return requests.get(url=config.BASE_URL + "/api/v2/plugin", headers={"Authorization": config.default_jwt})
 
+
 def get_version():
     return requests.get(url=config.BASE_URL + "/api/v2/version", headers={"Authorization": config.default_jwt})
 
+
 def post_ndriver_map(ndriver, driver, group):
-    return requests.post(url=config.BASE_URL + "/api/v2/ndriver/map", headers={"Authorization": config.default_jwt}, json={"ndriver": ndriver, "driver": driver, "group":group})
+    return requests.post(url=config.BASE_URL + "/api/v2/ndriver/map", headers={"Authorization": config.default_jwt}, json={"ndriver": ndriver, "driver": driver, "group": group})
+
 
 def delete_ndriver_map(ndriver, driver, group):
-    return requests.delete(url=config.BASE_URL + "/api/v2/ndriver/map", headers={"Authorization": config.default_jwt}, json={"ndriver": ndriver, "driver": driver, "group":group})
+    return requests.delete(url=config.BASE_URL + "/api/v2/ndriver/map", headers={"Authorization": config.default_jwt}, json={"ndriver": ndriver, "driver": driver, "group": group})
+
 
 def get_ndriver_map(ndriver):
     return requests.get(url=config.BASE_URL + "/api/v2/ndriver/map?ndriver=" + ndriver, headers={"Authorization": config.default_jwt})
 
+
 def put_ndriver_tag_param(json):
     return requests.put(url=config.BASE_URL + "/api/v2/ndriver/tag/param", headers={"Authorization": config.default_jwt}, json=json)
 
+
 def put_ndriver_tag_info(json):
     return requests.put(url=config.BASE_URL + "/api/v2/ndriver/tag/info", headers={"Authorization": config.default_jwt}, json=json)
 
+
 def get_ndriver_tag(ndriver, driver, group):
     return requests.get(url=config.BASE_URL + "/api/v2/ndriver/tag?ndriver=" + ndriver + "&driver=" + driver + "&group=" + group, headers={"Authorization": config.default_jwt})
 
+
+def prgfile_upload(driver, path, params):
+    return requests.post(url=config.BASE_URL + "/api/v2/prgfile", headers={"Authorization": config.default_jwt}, json={"driver": driver, "path": path, "parameters": params})
+
+
+def prgfile_process(driver):
+    return requests.get(url=config.BASE_URL + "/api/v2/prgfile?driver=" + driver, headers={"Authorization": config.default_jwt})
+
+
 # plugin setting
 
 
@@ -358,7 +384,20 @@ def modbus_rtu_node_setting(node, port, connection_mode=0, transport_mode=0, int
                                     "max_retries": max_retries, "link": link, "device": device, "stop": stop,
                                     "parity": parity, "baud": baud, "data": data})
 
+
+def modbus_tcp_node_setting_endian(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, endianness=1):
+    return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode, "interval": interval,
+                                    "host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "endianess": endianness})
+
+
+def modbus_rtu_node_setting_endian(node, port=502, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, link=1, device="", stop=0, parity=0, baud=4, data=3, endianness=1):
+    return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode,
+                                    "interval": interval, "host": host, "port": port, "timeout": timeout,
+                                    "max_retries": max_retries, "link": link, "device": device, "stop": stop,
+                                    "parity": parity, "baud": baud, "data": data, "endianess": endianness})
+
+
 def mqtt_node_setting(node):
-    return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos":0, "format": 0,
+    return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos": 0, "format": 0,
                                     "write-req-topic": f"/neuron/{node}/write/req", "write-resp-topic": f"/neuron/{node}/write/resp",
-                                    "offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})
+                                    "offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})

tests/ft/neuron/config.py

@@ -66,6 +66,11 @@ NEU_DATA_6 = 1
 NEU_DATA_7 = 2
 NEU_DATA_8 = 3
 
+ENDIAN_ABCD = 1,
+ENDIAN_BADC = 2,
+ENDIAN_DCBA = 3,
+ENDIAN_CDAB = 4,
+
 PLUGIN_MODBUS_TCP = "Modbus TCP"
 PLUGIN_MODBUS_RTU = "Modbus RTU"
 PLUGIN_MQTT = "MQTT"

tests/ft/neuron/error.py

@@ -69,6 +69,7 @@ NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT = 2313
 NEU_ERR_LIBRARY_MODULE_VERSION_NOT_MATCH = 2314
 NEU_ERR_LIBRARY_NAME_NOT_CONFORM = 2315
 NEU_ERR_LIBRARY_CLIB_NOT_MATCH = 2316
+NEU_ERR_LIBRARY_UPDATE_FAIL = 2317
 
 NEU_ERR_LICENSE_NOT_FOUND = 2400
 NEU_ERR_LICENSE_INVALID = 2401

tests/ft/plugin/test_plugin.py

@@ -200,7 +200,7 @@ class TestPlugin:
         response = api.updata_plugin(
             library_name=plugin_data['library'], so_file=plugin_data['so_file'], schema_file=plugin_data['schema_file'])
         assert 400 == response.status_code
-        assert NEU_ERR_LIBRARY_MODULE_NOT_EXISTS == response.json().get("error")
+        assert NEU_ERR_LIBRARY_UPDATE_FAIL == response.json().get("error")
 
     @description(given="incorrect configuration", when="update plugin", then="fail")
     def test_08_update_plugin_fail1(self):

tests/ft/tag/test_tag.py

@@ -355,4 +355,132 @@ class TestLog:
         assert 2         == response.json()['groups'][1]['tag_count']
         assert 'group_3' == response.json()['groups'][2]['group']
         assert 3000      == response.json()['groups'][2]['interval']
-        assert 2         == response.json()['groups'][2]['tag_count']
+        assert 2         == response.json()['groups'][2]['tag_count']
+
+    @description(given="multiple groups including same name tags in the first group", when="adding", then="add failed")
+    def test_adding_same_name_gtags_first_group(self):
+        response = api.add_gtags(
+            node='modbus-tcp-tag-test', 
+            groups=[
+                {
+                    "group": "group_f",
+                    "interval": 3000,
+                    "tags": [
+                        {
+                            "name": "tag1",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 3,
+                            "decimal": 0.01
+                        },
+                        {
+                            "name": "tag1",
+                            "address": "1!400002",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        }
+                    ]
+                },
+                {
+                    "group": "group_s",
+                    "interval": 3000,
+                    "tags": [
+                        {
+                            "name": "tag1",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        },
+                        {
+                            "name": "tag2",
+                            "address": "1!400002",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        }
+                    ]
+                }
+            ]
+        )
+
+        assert 409 == response.status_code
+        assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
+
+    @description(given="multiple groups including same name tags in the second group", when="adding", then="add failed")
+    def test_adding_same_name_gtags_second_group(self):
+        response = api.add_gtags(
+            node='modbus-tcp-tag-test', 
+            groups=[
+                {
+                    "group": "group_f",
+                    "interval": 3000,
+                    "tags": [
+                        {
+                            "name": "tag1",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 3,
+                            "decimal": 0.01
+                        },
+                        {
+                            "name": "tag2",
+                            "address": "1!400002",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        }
+                    ]
+                },
+                {
+                    "group": "group_s",
+                    "interval": 3000,
+                    "tags": [
+                        {
+                            "name": "tag1",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        },
+                        {
+                            "name": "tag1",
+                            "address": "1!400002",
+                            "attribute": 3,
+                            "type": 9,
+                            "precision": 3
+                        }
+                    ]
+                }
+            ]
+        )
+
+        assert 409 == response.status_code
+        assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
+
+    @description(given="tag name length exceeds 128", when="adding", then="add failed")
+    def test_adding_tag_name_exceeds_128(self):
+        response = api.add_gtags(
+            node='modbus-tcp-tag-test', 
+            groups=[{"group": "group","interval": 3000,
+                     "tags": [{
+                            "name": "Lt7Pa5Q0tKJN2gL3WiYJLnaFzqBq5q5xyJ1vYHxyf2j0IZR7y3RBYxXGQBHwV7dEzD7lIcF1yX9aR3vH8leYxXxHUILk2eVqQBlaQLaROZt9yZ9xQNxKkfy5mySHb79",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 3,
+                            "decimal": 0.01}]}])
+        assert 200 == response.status_code
+        assert NEU_ERR_SUCCESS == response.json()['error']
+
+        response = api.add_gtags(
+            node='modbus-tcp-tag-test', 
+            groups=[{"group": "group","interval": 3000,
+                     "tags": [{
+                            "name": "Lt7Pa5Q0tKJN2gL3WiYJLnaFzqBq5q5xyJ1vYHxyf2j0IZR7y3RBYxXGQBHwV7dEzD7lIcF1yX9aR3vH8leYxXxHUILk2eVqQBlaQLaROZt9yZ9xQNxKkfy5mySHb79Y8",
+                            "address": "1!400001",
+                            "attribute": 3,
+                            "type": 3,
+                            "decimal": 0.01}]}])
+        assert 400 == response.status_code
+        assert NEU_ERR_TAG_NAME_TOO_LONG == response.json()['error']

tests/plugins/c1/c1_2.c

@@ -127,9 +127,12 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
 static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
                           void *data)
 {
-    (void) plugin;
-    (void) head;
     (void) data;
+    switch (head->type) {
+    default:
+        plog_warn(plugin, "unhandle msg type: %d", head->type);
+        break;
+    }
     return 0;
 }
 

tests/ut/CMakeLists.txt

@@ -51,10 +51,18 @@ target_include_directories(modbus_test PRIVATE
 				${CMAKE_SOURCE_DIR}/plugins/modbus)
 target_link_libraries(modbus_test neuron-base gtest_main gtest pthread zlog)
 
+add_executable(rolling_counter_test rolling_counter_test.cc)
+target_include_directories(rolling_counter_test PRIVATE 
+	${CMAKE_SOURCE_DIR}/src
+	${CMAKE_SOURCE_DIR}/include
+)
+target_link_libraries(rolling_counter_test neuron-base gtest_main gtest)
+
 include(GoogleTest)
 gtest_discover_tests(json_test)
 gtest_discover_tests(http_test)
 gtest_discover_tests(jwt_test)
 gtest_discover_tests(base64_test)
 gtest_discover_tests(tag_sort_test)
-gtest_discover_tests(modbus_test)
+gtest_discover_tests(modbus_test)
+gtest_discover_tests(rolling_counter_test)

tests/ut/rolling_counter_test.cc

@@ -0,0 +1,45 @@
+#include <gtest/gtest.h>
+
+#include "utils/log.h"
+#include "utils/rolling_counter.h"
+
+zlog_category_t *neuron = NULL;
+TEST(RollingCounterTest, neu_rolling_counter_inc)
+{
+    uint64_t               ts      = 0;
+    neu_rolling_counter_t *counter = neu_rolling_counter_new(4000);
+    EXPECT_NE(nullptr, counter);
+
+    for (int i = 1; i <= 4; ++i) {
+        ts += 1000;
+        EXPECT_EQ(i, neu_rolling_counter_inc(counter, ts, 1));
+    }
+
+    // wrap around
+    for (int i = 1; i <= 4; ++i) {
+        ts += 1000;
+        EXPECT_EQ(4, neu_rolling_counter_inc(counter, ts, 1));
+    }
+
+    // increment without time stamp update
+    EXPECT_EQ(4, neu_rolling_counter_value(counter));
+    for (int i = 1; i <= 10; ++i) {
+        EXPECT_EQ(4 + i, neu_rolling_counter_inc(counter, ts, 1));
+    }
+
+    EXPECT_EQ(4 + 10, neu_rolling_counter_value(counter));
+
+    // stale value will be clean
+    ts += 4000;
+    EXPECT_EQ(0, neu_rolling_counter_inc(counter, ts, 0));
+
+    neu_rolling_counter_free(counter);
+}
+
+int main(int argc, char **argv)
+{
+    zlog_init("./config/dev.conf");
+    neuron = zlog_get_category("neuron");
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

version

@@ -1 +1 @@
-2.7.2
+2.7.5