Merge pull request #2691 from fengzeroz/v2.14

version: 2.14.0

.github/workflows/check.yml

@@ -27,7 +27,7 @@ jobs:
 
   cppcheck:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.14
 
     steps:
       - uses: actions/checkout@v4
@@ -42,7 +42,7 @@ jobs:
 
   build:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.14
 
     steps:
       - uses: actions/checkout@v4
@@ -65,7 +65,7 @@ jobs:
 
   ut:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.14
     needs: [build]
 
     steps:
@@ -103,7 +103,7 @@ jobs:
 
   pft:
     runs-on: ubuntu-24.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.14
     needs: [build]
     strategy:
       matrix:

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ on:
 env:
   VERSION: 2.14.0
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.14
 
 jobs:
   cross-build:
@@ -19,7 +19,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.14
 
     steps:
       - uses: actions/checkout@v4
@@ -62,7 +62,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.14
 
     steps:
       - uses: actions/checkout@v4

cmake/aarch64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX aarch64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR aarch64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.14/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabi.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabi)
 set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.14/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabihf.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabihf)
 set(CMAKE_SYSTEM_PROCESSOR armv7l)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.14/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/x86_64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX x86_64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR x86_64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.14/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

dev.conf

@@ -10,6 +10,5 @@ syslog = "<P>%d(%Y-%m-%dT%T%z) %H neuron[%p]: level=%v msg="[%c] %m" file="%f:%L
 
 [rules]
 
-*.*     "./logs/%c.log", 25MB * 1 ~ "./logs/%c.#2r.log"; file_format
+*.*     "./logs/%c.log", 50MB * 2 ~ "./logs/%c.#2r.log"; file_format
 *.*    	>stdout; stdout_format
-*.*     $remote_syslog,"%V"; syslog

include/neuron/adapter.h

@@ -86,6 +86,9 @@ typedef struct adapter_callbacks {
             void (*update_im)(neu_adapter_t *adapter, const char *group,
                               const char *tag, neu_dvalue_t value,
                               neu_tag_meta_t *metas, int n_meta);
+            void (*update_im_f_m)(neu_adapter_t *adapter, const char *group,
+                                  const char *tag, neu_dvalue_t value,
+                                  neu_tag_meta_t *metas, int n_meta);
             void (*scan_tags_response)(neu_adapter_t *adapter, void *r,
                                        neu_resp_scan_tags_t *resp_scan);
             void (*test_read_tag_response)(neu_adapter_t *adapter, void *r,

include/neuron/connection/neu_connection.h

@@ -305,5 +305,6 @@ int is_ipv6(const char *ip);
 
 bool     neu_conn_is_connected(neu_conn_t *conn);
 uint16_t neu_conn_udp_act_port(neu_conn_t *conn);
+bool     conn_tcp_server_is_listen(neu_conn_t *conn);
 
 #endif

include/neuron/define.h

@@ -40,6 +40,7 @@
 #define NEU_DEFAULT_GROUP_INTERVAL 100
 #define NEU_NODE_NAME_LEN 128
 #define NEU_PLUGIN_NAME_LEN 32
+#define NEU_NODE_TAGS_LEN 256
 #define NEU_PLUGIN_LIBRARY_LEN 64
 #define NEU_PLUGIN_DESCRIPTION_LEN 512
 #define NEU_TEMPLATE_NAME_LEN 128

include/neuron/errcodes.h

@@ -69,6 +69,9 @@ typedef enum {
     NEU_ERR_NODE_NOT_ALLOW_UPDATE    = 2013,
     NEU_ERR_NODE_NOT_ALLOW_MAP       = 2014,
     NEU_ERR_NODE_NAME_EMPTY          = 2015,
+    NEU_ERR_NODE_TAGS_TOO_LONG       = 2016,
+    NEU_ERR_NODE_TAGS_TOO_MANY       = 2017,
+    NEU_ERR_NODE_TAGS_INVALID        = 2018,
 
     NEU_ERR_GROUP_ALREADY_SUBSCRIBED = 2101,
     NEU_ERR_GROUP_NOT_SUBSCRIBE      = 2102,

include/neuron/msg.h

@@ -78,6 +78,7 @@ typedef enum neu_reqresp_type {
     NEU_RESP_NODE_UNINIT,
     NEU_REQ_ADD_NODE,
     NEU_REQ_UPDATE_NODE,
+    NEU_REQ_UPDATE_NODE_TAG,
     NEU_REQ_DEL_NODE,
     NEU_REQ_GET_NODE,
     NEU_RESP_GET_NODE,
@@ -242,6 +243,7 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_NODE_UNINIT]      = "NEU_RESP_NODE_UNINIT",
     [NEU_REQ_ADD_NODE]          = "NEU_REQ_ADD_NODE",
     [NEU_REQ_UPDATE_NODE]       = "NEU_REQ_UPDATE_NODE",
+    [NEU_REQ_UPDATE_NODE_TAG]   = "NEU_REQ_UPDATE_NODE_TAG",
     [NEU_REQ_DEL_NODE]          = "NEU_REQ_DEL_NODE",
     [NEU_REQ_GET_NODE]          = "NEU_REQ_GET_NODE",
     [NEU_RESP_GET_NODE]         = "NEU_RESP_GET_NODE",
@@ -522,11 +524,14 @@ typedef struct neu_req_add_node {
     char  node[NEU_NODE_NAME_LEN];
     char  plugin[NEU_PLUGIN_NAME_LEN];
     char *setting;
+    char *tags;
 } neu_req_add_node_t;
 
 static inline void neu_req_add_node_fini(neu_req_add_node_t *req)
 {
     free(req->setting);
+    if (req->tags)
+        free(req->tags);
 }
 
 typedef struct neu_req_update_node {
@@ -534,6 +539,11 @@ typedef struct neu_req_update_node {
     char new_name[NEU_NODE_NAME_LEN];
 } neu_req_update_node_t;
 
+typedef struct neu_req_update_node_tag {
+    char node[NEU_NODE_NAME_LEN];
+    char tags[NEU_NODE_TAGS_LEN];
+} neu_req_update_node_tag_t;
+
 typedef struct neu_req_del_node {
     char node[NEU_NODE_NAME_LEN];
 } neu_req_del_node_t;
@@ -557,6 +567,7 @@ typedef struct neu_resp_node_info {
     int64_t delay;
     char    node[NEU_NODE_NAME_LEN];
     char    plugin[NEU_PLUGIN_NAME_LEN];
+    char    tags[NEU_NODE_TAGS_LEN];
 } neu_resp_node_info_t;
 
 typedef struct neu_resp_get_node {
@@ -1148,6 +1159,7 @@ typedef struct {
     char *          node;
     char *          plugin;
     char *          setting;
+    char *          tags;
     uint16_t        n_group;
     neu_gdatatag_t *groups;
 } neu_req_driver_t;
@@ -1157,6 +1169,9 @@ static inline void neu_req_driver_fini(neu_req_driver_t *req)
     free(req->node);
     free(req->plugin);
     free(req->setting);
+    if (req->tags)
+        free(req->tags);
+
     for (uint16_t i = 0; i < req->n_group; i++) {
         for (int j = 0; j < req->groups[i].n_tag; j++) {
             neu_tag_fini(&req->groups[i].tags[j]);

include/neuron/persist/persist.h

@@ -36,6 +36,7 @@ typedef struct {
     int   type;
     char *plugin_name;
     int   state;
+    char *tags;
 } neu_persist_node_info_t;
 
 typedef struct {
@@ -125,6 +126,9 @@ static inline void neu_persist_node_info_fini(neu_persist_node_info_t *info)
 {
     free(info->name);
     free(info->plugin_name);
+    if (info->tags) {
+        free(info->tags);
+    }
 }
 
 static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info)
@@ -250,6 +254,7 @@ int neu_persister_delete_node(const char *node_name);
  * @return 0 on success, none-zero on failure
  */
 int neu_persister_update_node(const char *node_name, const char *new_name);
+int neu_persister_update_node_tags(const char *node_name, const char *tags);
 
 /**
  * Update node state.

persistence/0114_2.14.0_modbus_simulator.sql

@@ -1,6 +1,12 @@
+BEGIN TRANSACTION;
+
 CREATE TABLE IF NOT EXISTS modbus_tcp_simulator (
   id INTEGER PRIMARY KEY CHECK (id = 1),
   enabled INTEGER NOT NULL DEFAULT 0,
   tags_json TEXT NOT NULL DEFAULT '[]',
   updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
+
+alter TABLE nodes add column tags TEXT NULL;
+
+COMMIT;

plugins/modbus/modbus_point.c

@@ -52,6 +52,10 @@ int modbus_tag_to_point(const neu_datatag_t *tag, modbus_point_t *point,
         return NEU_ERR_TAG_ADDRESS_FORMAT_INVALID;
     }
 
+    if (start_address > 65536) {
+        return NEU_ERR_TAG_ADDRESS_FORMAT_INVALID;
+    }
+
     if (start_address == 65536 && address_base == 0) {
         point->start_address = 65535;
     } else if (start_address == 0 && address_base == 1) {

plugins/mqtt/mqtt_plugin_intf.c

@@ -167,6 +167,7 @@ int mqtt_plugin_uninit(neu_plugin_t *plugin)
 
     mqtt_config_fini(&plugin->config);
     if (plugin->client) {
+        neu_mqtt_client_remove_cache_db(plugin->client);
         neu_mqtt_client_close(plugin->client);
         neu_mqtt_client_free(plugin->client);
         plugin->client = NULL;
@@ -395,6 +396,7 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
     }
 
     if (plugin->client != NULL) {
+        neu_mqtt_client_remove_cache_db(plugin->client);
         neu_mqtt_client_close(plugin->client);
         neu_mqtt_client_free(plugin->client);
     }

plugins/restful/adapter_handle.c

@@ -37,6 +37,46 @@
 #include "utils/utextend.h"
 #include "utils/uthash.h"
 
+static int tags_length(const char *tags)
+{
+    int   length = 0;
+    char *tag    = NULL;
+    char *tmp    = strdup(tags);
+    char *save   = NULL;
+
+    tag = strtok_r(tmp, ",", &save);
+    while (tag != NULL) {
+        length++;
+        tag = strtok_r(NULL, ",", &save);
+    }
+
+    free(tmp);
+    return length;
+}
+
+/*
+'!', '@', '#', '$', '%', '^', '&', '*', '(', ')', '-', '+', '=', '[', ']', '{',
+'}',
+            ';', ':', '\'', '"', ',', '<', '>', '?', '\\', '|', '~', '`', '\t',
+'\n',
+*/
+static bool tags_check_symbol(const char *tags)
+{
+    for (size_t i = 0; i < strlen(tags); i++) {
+        char c = tags[i];
+        if (c == '!' || c == '@' || c == '#' || c == '$' || c == '%' ||
+            c == '^' || c == '&' || c == '*' || c == '(' || c == ')' ||
+            c == '-' || c == '+' || c == '=' || c == '[' || c == ']' ||
+            c == '{' || c == '}' || c == ';' || c == ':' || c == '\'' ||
+            c == '"' || c == '<' || c == '>' || c == '?' || c == '\\' ||
+            c == '|' || c == '~' || c == '`' || c == '\t' || c == '\n') {
+            return false;
+        }
+    }
+
+    return true;
+}
+
 void handle_add_adapter(nng_aio *aio)
 {
     neu_plugin_t *plugin = neu_rest_get_plugin();
@@ -45,6 +85,14 @@ void handle_add_adapter(nng_aio *aio)
         aio, neu_json_add_node_req_t, neu_json_decode_add_node_req, {
             if (strlen(req->name) >= NEU_NODE_NAME_LEN) {
                 CHECK_NODE_NAME_LENGTH_ERR;
+            } else if (req->tags != NULL && !tags_check_symbol(req->tags)) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_TAGS_INVALID, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
+            } else if (req->tags != NULL && tags_length(req->tags) > 5) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_TAGS_TOO_MANY, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
             } else if (strcmp(req->name, "monitor") == 0 ||
                        strcmp(req->name, "DataStorage") == 0) {
                 CHECK_NODE_SINGLE_ERR;
@@ -59,7 +107,9 @@ void handle_add_adapter(nng_aio *aio)
                 strcpy(cmd.node, req->name);
                 strcpy(cmd.plugin, req->plugin);
                 cmd.setting  = req->setting;
+                cmd.tags     = req->tags;
                 req->setting = NULL; // moved
+                req->tags    = NULL; // moved
                 ret          = neu_plugin_op(plugin, header, &cmd);
                 if (ret != 0) {
                     neu_req_add_node_fini(&cmd);
@@ -236,6 +286,7 @@ void handle_get_adapter_resp(nng_aio *aio, neu_resp_get_node_t *nodes)
 
         nodes_res.nodes[index].name   = info->node;
         nodes_res.nodes[index].plugin = info->plugin;
+        nodes_res.nodes[index].tags   = info->tags;
         nodes_res.nodes[index].support_import_tags =
             neu_plugin_support_import_tags_simple(info->plugin);
     }
@@ -501,6 +552,12 @@ static int send_drivers(nng_aio *aio, neu_json_drivers_req_t *req)
             ret = NEU_ERR_NODE_NAME_TOO_LONG;
             goto check_end;
         }
+        if (driver->node.tags != NULL &&
+            strlen(driver->node.tags) >= NEU_NODE_TAGS_LEN) {
+            ret = NEU_ERR_NODE_TAGS_TOO_LONG;
+            goto check_end;
+        }
+
         if (strlen(driver->node.plugin) >= NEU_PLUGIN_NAME_LEN) {
             ret = NEU_ERR_PLUGIN_NAME_TOO_LONG;
             goto check_end;
@@ -592,9 +649,11 @@ check_end:
         cmd.drivers[i].plugin     = driver->node.plugin;
         cmd.drivers[i].setting    = driver->node.setting;
         cmd.drivers[i].n_group    = driver->gtags.len;
+        cmd.drivers[i].tags       = driver->node.tags;
         driver->node.name         = NULL; // moved
         driver->node.plugin       = NULL; // moved
         driver->node.setting      = NULL; // moved
+        driver->node.tags         = NULL; // moved
         cmd.n_driver += 1;
 
         ret = json_to_gdatatags(&driver->gtags, &cmd.drivers[i].groups);
@@ -623,3 +682,36 @@ void handle_put_drivers(nng_aio *aio)
             }
         })
 }
+
+void handle_put_node_tag(nng_aio *aio)
+{
+    neu_plugin_t *plugin = neu_rest_get_plugin();
+
+    NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
+        aio, neu_json_update_node_tag_req_t,
+        neu_json_decode_update_node_tag_req, {
+            if (req->tags != NULL && !tags_check_symbol(req->tags)) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_TAGS_INVALID, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
+            } else if (req->tags != NULL && tags_length(req->tags) > 5) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_TAGS_TOO_MANY, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
+                neu_json_decode_update_node_tag_req_free(req);
+            } else {
+                neu_reqresp_head_t        header = { 0 };
+                neu_req_update_node_tag_t cmd    = { 0 };
+                header.ctx                       = aio;
+                header.type                      = NEU_REQ_UPDATE_NODE_TAG;
+                strcpy(cmd.node, req->name);
+                strcpy(cmd.tags, req->tags);
+                int ret = neu_plugin_op(plugin, header, &cmd);
+                if (ret != 0) {
+                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+                        neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+                    });
+                }
+            }
+        })
+}

plugins/restful/adapter_handle.h

@@ -41,5 +41,6 @@ void handle_get_nodes_state_resp(nng_aio *                   aio,
                                  neu_resp_get_nodes_state_t *states);
 
 void handle_put_drivers(nng_aio *aio);
+void handle_put_node_tag(nng_aio *aio);
 
 #endif

plugins/restful/global_config_handle.c

@@ -60,6 +60,7 @@ typedef enum {
     STATE_DEL_APP,
     STATE_DEL_DRIVER,
     STATE_ADD_NODE,
+    STATE_UPDATE_DEFAULT_APP_TAGS,
     STATE_ADD_GROUP,
     STATE_ADD_TAG,
     STATE_GET_APP_SUBS,
@@ -126,6 +127,7 @@ static int add_group(context_t *                             ctx,
 static int add_tag(context_t *ctx, neu_json_add_tags_req_t *data);
 static int add_subscription(context_t *ctx, neu_json_subscribe_req_t *data);
 static int add_setting(context_t *ctx, neu_json_node_setting_req_t *data);
+static int add_app_tags(context_t *ctx, neu_json_app_t *app);
 
 static inline bool is_static_node(const char *name, const char *plugin)
 {
@@ -743,6 +745,29 @@ static void put_apps_context_next(context_t *ctx, neu_reqresp_type_e type,
         }
 
         // After adding nodes, get subscriptions for default apps to unsub
+        ctx->idx       = 0;
+        ctx->unsub_idx = 0;
+        ctx->state     = STATE_UPDATE_DEFAULT_APP_TAGS;
+    }
+        // fall through
+    case STATE_UPDATE_DEFAULT_APP_TAGS: {
+        neu_resp_error_t *   err  = data;
+        neu_json_apps_req_t *apps = ctx->apps_req;
+
+        if (ctx->idx > 0 && 0 != err->error) {
+            ctx->error = err->error;
+            ctx->state = STATE_END;
+            break;
+        }
+
+        for (; ctx->idx < (size_t) apps->n_app; ++ctx->idx) {
+            neu_json_app_t *app = &apps->apps[ctx->idx];
+            NEXT(ctx, add_app_tags, app);
+
+            ++ctx->idx;
+            goto done;
+        }
+
         ctx->idx       = 0;
         ctx->unsub_idx = 0;
         ctx->state     = STATE_GET_APP_SUBS;
@@ -950,6 +975,7 @@ static int get_drivers_resp(context_t *ctx, neu_resp_get_node_t *resp)
         neu_resp_node_info_t *info     = utarray_eltptr(nodes, i);
         drivers.drivers[i].node.name   = info->node;
         drivers.drivers[i].node.plugin = info->plugin;
+        drivers.drivers[i].node.tags   = info->tags;
     }
 
 end:
@@ -1654,6 +1680,7 @@ static int get_apps_resp(context_t *ctx, neu_resp_get_node_t *resp)
         neu_resp_node_info_t *info = utarray_eltptr(nodes, i);
         apps.apps[i].node.name     = info->node;
         apps.apps[i].node.plugin   = info->plugin;
+        apps.apps[i].node.tags     = info->tags;
     }
 
 end:
@@ -1710,12 +1737,17 @@ static int get_app_subscriptions_resp(context_t *                     ctx,
         grps[index].driver = strdup(group->driver);
         grps[index].group  = strdup(group->group);
         grps[index].params = group->params ? strdup(group->params) : NULL;
+        grps[index].static_tags =
+            group->static_tags ? strdup(group->static_tags) : NULL;
 
         if (NULL == grps[index].driver || NULL == grps[index].group) {
             for (int j = 0; j <= index; ++j) {
                 free(grps[j].driver);
                 free(grps[j].group);
                 free(grps[j].params);
+                if (grps[j].static_tags) {
+                    free(grps[j].static_tags);
+                }
             }
             free(grps);
             rv = NEU_ERR_EINTERNAL;
@@ -1766,7 +1798,10 @@ static int add_app_node(context_t *ctx, neu_json_app_t *app)
     neu_req_add_node_t cmd = { 0 };
     strcpy(cmd.node, app->node.name);
     strcpy(cmd.plugin, app->node.plugin);
-    cmd.setting       = app->node.setting;
+    cmd.setting = app->node.setting;
+    if (app->node.tags != NULL) {
+        cmd.tags = strdup(app->node.tags);
+    }
     app->node.setting = NULL;
 
     if (0 != neu_plugin_op(plugin, header, &cmd)) {
@@ -1776,6 +1811,29 @@ static int add_app_node(context_t *ctx, neu_json_app_t *app)
     return 0;
 }
 
+static int add_app_tags(context_t *ctx, neu_json_app_t *app)
+{
+    neu_plugin_t *plugin = neu_rest_get_plugin();
+
+    neu_reqresp_head_t header = {
+        .otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
+        .type            = NEU_REQ_UPDATE_NODE_TAG,
+        .ctx             = ctx->aio,
+    };
+
+    neu_req_update_node_tag_t cmd = { 0 };
+    strcpy(cmd.node, app->node.name);
+    if (app->node.tags != NULL) {
+        strcpy(cmd.tags, app->node.tags);
+    }
+
+    if (0 != neu_plugin_op(plugin, header, &cmd)) {
+        return NEU_ERR_IS_BUSY;
+    }
+
+    return 0;
+}
+
 static int add_app_setting(context_t *ctx, neu_json_app_t *app)
 {
     neu_plugin_t *plugin = neu_rest_get_plugin();
@@ -1813,8 +1871,10 @@ static int add_app_subscription(context_t *ctx, neu_json_app_t *app,
     strcpy(cmd.app, app->node.name);
     strcpy(cmd.driver, grp->driver);
     strcpy(cmd.group, grp->group);
-    cmd.params  = grp->params;
-    grp->params = NULL;
+    cmd.params       = grp->params;
+    cmd.static_tags  = grp->static_tags;
+    grp->params      = NULL;
+    grp->static_tags = NULL;
 
     if (0 != neu_plugin_op(plugin, header, &cmd)) {
         return NEU_ERR_IS_BUSY;

plugins/restful/handle.c

@@ -756,6 +756,12 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/simulator/export",
         .value.handler = handle_simulator_export,
     },
+    {
+        .method        = NEU_HTTP_METHOD_PUT,
+        .type          = NEU_HTTP_HANDLER_FUNCTION,
+        .url           = "/api/v2/node/tag",
+        .value.handler = handle_put_node_tag,
+    },
 };
 
 void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)

simulator/modbus_simulator.c

@@ -90,6 +90,7 @@ static void sig_handler(int sig)
     for (uint32_t i = 0; i < sizeof(c_events) / sizeof(struct client_event);
          i++) {
         if (c_events[i].fd > 0) {
+            close(c_events[i].fd);
             neu_event_del_io(events, c_events[i].client);
         }
     }
@@ -149,6 +150,11 @@ int main(int argc, char *argv[])
 
     events = neu_event_new("modbus_simulator");
     conn   = neu_conn_new(&param, NULL, connected, disconnected);
+    while (conn_tcp_server_is_listen(conn) == false) {
+        neu_msleep(1000);
+        neu_conn_destory(conn);
+        conn = neu_conn_new(&param, NULL, connected, disconnected);
+    }
 
     signal(SIGINT, sig_handler);
     signal(SIGTERM, sig_handler);

simulator/modbus_tcp_simulator.c

@@ -13,6 +13,7 @@
 #include <unistd.h>
 
 #include "neuron.h"
+
 #include "connection/neu_connection.h"
 #include "event/event.h"
 #include "modbus/modbus.h"
@@ -32,9 +33,9 @@ typedef struct {
     bool            running;
     char            ip[64];
     uint16_t        port;
-    neu_events_t   *events;
+    neu_events_t *  events;
     neu_event_io_t *listen_io;
-    neu_conn_t     *conn;
+    neu_conn_t *    conn;
     pthread_t       gen_thread;
     atomic_bool     gen_stop;
     pthread_mutex_t lock;
@@ -46,8 +47,8 @@ typedef struct {
 
     bool               readonly_mask[SIM_MAX_REGS];
     neu_sim_tag_type_e tag_type[SIM_MAX_REGS];
-    char              *tag_name[SIM_MAX_REGS];
-    char              *tag_addr[SIM_MAX_REGS];
+    char *             tag_name[SIM_MAX_REGS];
+    char *             tag_addr[SIM_MAX_REGS];
     int                tag_count;
     int                last_error_code;
 } sim_ctx_t;
@@ -57,7 +58,7 @@ static sim_ctx_t g_sim = { 0 };
 struct client_event {
     neu_event_io_t *client;
     int             fd;
-    void           *user;
+    void *          user;
 };
 static struct client_event g_clients[32] = { 0 };
 
@@ -127,7 +128,7 @@ static int prng_int(int seed, int min, int max)
     x ^= x >> 17;
     x ^= x << 5;
     int span = max - min + 1;
-    return min + (int) (x % (uint32_t) (span > 0 ? span : 1));
+    return min + (int) (x % (uint32_t)(span > 0 ? span : 1));
 }
 
 static int pdu_read_holding(const uint8_t *pdu, int pdu_len,
@@ -151,7 +152,7 @@ static int pdu_read_holding(const uint8_t *pdu, int pdu_len,
     if (*res_len + 1 + 2 * nreg > res_mlen)
         return -1;
     uint8_t *res_data = res + *res_len;
-    res_data[0]       = (uint8_t) (nreg * 2);
+    res_data[0]       = (uint8_t)(nreg * 2);
     pthread_mutex_lock(&g_sim.lock);
     for (uint16_t i = 0; i < nreg; ++i) {
         uint16_t v  = g_sim.hold_regs[start + i];
@@ -185,12 +186,12 @@ static int pdu_read_input_regs(const uint8_t *pdu, int pdu_len,
     if (*res_len + 1 + 2 * nreg > res_mlen)
         return -1;
     uint8_t *res_data = res + *res_len;
-    res_data[0]       = (uint8_t) (nreg * 2);
+    res_data[0]       = (uint8_t)(nreg * 2);
     pthread_mutex_lock(&g_sim.lock);
     for (uint16_t i = 0; i < nreg; ++i) {
         uint16_t v              = g_sim.input_regs[start + i];
-        res_data[1 + i * 2]     = (uint8_t) (v >> 8);
-        res_data[1 + i * 2 + 1] = (uint8_t) (v & 0xFF);
+        res_data[1 + i * 2]     = (uint8_t)(v >> 8);
+        res_data[1 + i * 2 + 1] = (uint8_t)(v & 0xFF);
     }
     pthread_mutex_unlock(&g_sim.lock);
     *res_len += 1 + 2 * nreg;
@@ -225,7 +226,7 @@ static int pdu_read_coils(const uint8_t *pdu, int pdu_len,
     pthread_mutex_lock(&g_sim.lock);
     for (uint16_t i = 0; i < nbits; ++i) {
         if (g_sim.coil_bits[start + i]) {
-            res_data[1 + (i / 8)] |= (uint8_t) (1u << (i % 8));
+            res_data[1 + (i / 8)] |= (uint8_t)(1u << (i % 8));
         }
     }
     pthread_mutex_unlock(&g_sim.lock);
@@ -261,7 +262,7 @@ static int pdu_read_inputs(const uint8_t *pdu, int pdu_len,
     pthread_mutex_lock(&g_sim.lock);
     for (uint16_t i = 0; i < nbits; ++i) {
         if (g_sim.input_bits[start + i]) {
-            res_data[1 + (i / 8)] |= (uint8_t) (1u << (i % 8));
+            res_data[1 + (i / 8)] |= (uint8_t)(1u << (i % 8));
         }
     }
     pthread_mutex_unlock(&g_sim.lock);
@@ -271,7 +272,7 @@ static int pdu_read_inputs(const uint8_t *pdu, int pdu_len,
 }
 
 static int pdu_write_single_holding(const uint8_t *pdu, int pdu_len,
-                                    struct modbus_code   *rc,
+                                    struct modbus_code *  rc,
                                     struct modbus_header *rh, uint8_t *res,
                                     int res_mlen, int *res_len)
 {
@@ -292,7 +293,7 @@ static int pdu_write_single_holding(const uint8_t *pdu, int pdu_len,
     bool is_readonly = g_sim.readonly_mask[start];
     if (!is_readonly) {
         const uint8_t *vbytes  = pdu + offsetof(struct modbus_address, n_reg);
-        uint16_t       v       = (uint16_t) ((vbytes[0] << 8) | vbytes[1]);
+        uint16_t       v       = (uint16_t)((vbytes[0] << 8) | vbytes[1]);
         g_sim.hold_regs[start] = v;
     }
     pthread_mutex_unlock(&g_sim.lock);
@@ -314,7 +315,7 @@ static int pdu_write_single_holding(const uint8_t *pdu, int pdu_len,
 }
 
 static int pdu_write_multiple_holding(const uint8_t *pdu, int pdu_len,
-                                      struct modbus_code   *rc,
+                                      struct modbus_code *  rc,
                                       struct modbus_header *rh, uint8_t *res,
                                       int res_mlen, int *res_len)
 {
@@ -322,7 +323,7 @@ static int pdu_write_multiple_holding(const uint8_t *pdu, int pdu_len,
         (int) (sizeof(struct modbus_address) + sizeof(struct modbus_data)))
         return -1;
     struct modbus_address *addr = (struct modbus_address *) pdu;
-    struct modbus_data    *data =
+    struct modbus_data *   data =
         (struct modbus_data *) (pdu + sizeof(struct modbus_address));
     uint16_t start = ntohs(addr->start_address);
     uint16_t nreg  = ntohs(addr->n_reg);
@@ -346,7 +347,7 @@ static int pdu_write_multiple_holding(const uint8_t *pdu, int pdu_len,
     if (!deny) {
         for (uint16_t i = 0; i < nreg; ++i) {
             const uint8_t *bp          = &data->byte[i * 2];
-            uint16_t       v           = (uint16_t) ((bp[0] << 8) | bp[1]);
+            uint16_t       v           = (uint16_t)((bp[0] << 8) | bp[1]);
             g_sim.hold_regs[start + i] = v;
         }
     }
@@ -369,7 +370,7 @@ static int pdu_write_multiple_holding(const uint8_t *pdu, int pdu_len,
 }
 
 static int pdu_write_single_coil(const uint8_t *pdu, int pdu_len,
-                                 struct modbus_code   *rc,
+                                 struct modbus_code *  rc,
                                  struct modbus_header *rh, uint8_t *res,
                                  int res_mlen, int *res_len)
 {
@@ -400,7 +401,7 @@ static int pdu_write_single_coil(const uint8_t *pdu, int pdu_len,
 }
 
 static int pdu_write_multiple_coils(const uint8_t *pdu, int pdu_len,
-                                    struct modbus_code   *rc,
+                                    struct modbus_code *  rc,
                                     struct modbus_header *rh, uint8_t *res,
                                     int res_mlen, int *res_len)
 {
@@ -408,7 +409,7 @@ static int pdu_write_multiple_coils(const uint8_t *pdu, int pdu_len,
         (int) (sizeof(struct modbus_address) + sizeof(struct modbus_data)))
         return -1;
     struct modbus_address *addr = (struct modbus_address *) pdu;
-    struct modbus_data    *data =
+    struct modbus_data *   data =
         (struct modbus_data *) (pdu + sizeof(struct modbus_address));
     uint16_t start = ntohs(addr->start_address);
     uint16_t nbits = ntohs(addr->n_reg);
@@ -446,8 +447,8 @@ static inline void sim_update_sine_hi_lo(uint16_t addr, int64_t tms)
         uint32_t u;
     } fu;
     fu.f        = fval;
-    uint16_t lo = (uint16_t) (fu.u & 0xFFFFu);
-    uint16_t hi = (uint16_t) ((fu.u >> 16) & 0xFFFFu);
+    uint16_t lo = (uint16_t)(fu.u & 0xFFFFu);
+    uint16_t hi = (uint16_t)((fu.u >> 16) & 0xFFFFu);
     if (addr + 1 < SIM_MAX_REGS) {
         g_sim.hold_regs[addr]     = hi;
         g_sim.hold_regs[addr + 1] = lo;
@@ -471,7 +472,7 @@ static inline void sim_update_square(uint16_t addr, int64_t tms)
     int64_t period_ms     = 10000;
     int64_t phase         = tms % period_ms;
     int     v             = (phase < (period_ms / 2)) ? -10 : 10;
-    g_sim.hold_regs[addr] = (uint16_t) (int16_t) v;
+    g_sim.hold_regs[addr] = (uint16_t)(int16_t) v;
 }
 
 static inline void sim_update_random(uint16_t addr, int64_t tms)
@@ -520,10 +521,21 @@ static int add_client(int fd, neu_event_io_t *client, void *user)
     return -1;
 }
 
+static void clean_client()
+{
+    for (size_t i = 0; i < sizeof(g_clients) / sizeof(g_clients[0]); ++i) {
+        if (g_clients[i].fd != 0) {
+            close(g_clients[i].fd);
+            g_clients[i].fd = 0;
+        }
+    }
+}
+
 static neu_event_io_t *del_client(int fd)
 {
     for (size_t i = 0; i < sizeof(g_clients) / sizeof(g_clients[0]); ++i) {
         if (g_clients[i].fd == fd) {
+            close(g_clients[i].fd);
             g_clients[i].fd = 0;
             return g_clients[i].client;
         }
@@ -553,7 +565,7 @@ static int process_modbus_tcp(uint8_t *req, uint16_t req_len, uint8_t *res,
 
     memcpy(res, req, sizeof(struct modbus_header));
     struct modbus_header *rh = (struct modbus_header *) res;
-    struct modbus_code   *c =
+    struct modbus_code *  c =
         (struct modbus_code *) (req + sizeof(struct modbus_header));
     struct modbus_code *rc =
         (struct modbus_code *) (res + sizeof(struct modbus_header));
@@ -656,7 +668,7 @@ static int listen_io_cb(enum neu_event_io_type type, int lfd, void *usr)
     if (type == NEU_EVENT_IO_READ) {
         int cfd = neu_conn_tcp_server_accept(g_sim.conn);
         if (cfd > 0) {
-            struct cycle_buf    *buf = calloc(1, sizeof(struct cycle_buf));
+            struct cycle_buf *   buf = calloc(1, sizeof(struct cycle_buf));
             neu_event_io_param_t cio = {
                 .fd       = cfd,
                 .usr_data = buf,
@@ -700,7 +712,7 @@ static char *sim_build_tags_json_unlocked(void)
         if (tp == NEU_SIM_TAG_NONE) {
             continue;
         }
-        json_t     *obj  = json_object();
+        json_t *    obj  = json_object();
         const char *name = g_sim.tag_name[idx] ? g_sim.tag_name[idx] : "";
         const char *addr = g_sim.tag_addr[idx] ? g_sim.tag_addr[idx] : "";
         json_object_set_new(obj, "name", json_string(name));
@@ -730,9 +742,9 @@ static void sim_persist_save(void)
     pthread_mutex_unlock(&g_sim.lock);
 
     sqlite3_stmt *stmt = NULL;
-    const char   *sql  = "INSERT OR REPLACE INTO modbus_tcp_simulator (id, "
-                         "enabled, tags_json, updated_at) "
-                         "VALUES (1, ?, ?, CURRENT_TIMESTAMP)";
+    const char *  sql  = "INSERT OR REPLACE INTO modbus_tcp_simulator (id, "
+                      "enabled, tags_json, updated_at) "
+                      "VALUES (1, ?, ?, CURRENT_TIMESTAMP)";
     if (SQLITE_OK == sqlite3_prepare_v2(db, sql, -1, &stmt, NULL)) {
         sqlite3_bind_int(stmt, 1, enabled);
         if (tags) {
@@ -755,7 +767,7 @@ static void sim_persist_load_and_apply(void)
         return;
     }
     sqlite3_stmt *stmt = NULL;
-    const char   *sql =
+    const char *  sql =
         "SELECT enabled, tags_json FROM modbus_tcp_simulator WHERE id=1";
     if (SQLITE_OK != sqlite3_prepare_v2(db, sql, -1, &stmt, NULL)) {
         return;
@@ -768,12 +780,12 @@ static void sim_persist_load_and_apply(void)
 
         if (tags_str && tags_str[0]) {
             json_error_t jerr;
-            json_t      *arr = json_loads(tags_str, 0, &jerr);
+            json_t *     arr = json_loads(tags_str, 0, &jerr);
             if (arr && json_is_array(arr)) {
                 size_t              count   = json_array_size(arr);
-                char              **names   = NULL;
-                char              **addrs   = NULL;
-                uint16_t           *indices = NULL;
+                char **             names   = NULL;
+                char **             addrs   = NULL;
+                uint16_t *          indices = NULL;
                 neu_sim_tag_type_e *types   = NULL;
                 if (count > 0) {
                     names   = calloc(count, sizeof(char *));
@@ -794,18 +806,18 @@ static void sim_persist_load_and_apply(void)
                         continue;
                     uint16_t idx = (uint16_t) json_integer_value(jidx);
                     int      tpv = json_is_integer(jtype)
-                             ? (int) json_integer_value(jtype)
-                             : 0;
+                        ? (int) json_integer_value(jtype)
+                        : 0;
                     if (idx >= SIM_MAX_REGS)
                         continue;
                     indices[m] = idx;
                     types[m]   = (neu_sim_tag_type_e) tpv;
                     names[m]   = jname && json_is_string(jname)
-                          ? strdup(json_string_value(jname))
-                          : strdup("");
-                    addrs[m]   = jaddr && json_is_string(jaddr)
-                          ? strdup(json_string_value(jaddr))
-                          : strdup("");
+                        ? strdup(json_string_value(jname))
+                        : strdup("");
+                    addrs[m] = jaddr && json_is_string(jaddr)
+                        ? strdup(json_string_value(jaddr))
+                        : strdup("");
                     ++m;
                 }
                 if (m > 0) {
@@ -877,7 +889,7 @@ int neu_modbus_simulator_init(const char *config_dir)
             if (buf && sz > 0) {
                 fread(buf, 1, (size_t) sz, fp);
                 json_error_t jerr;
-                json_t      *root = json_loads(buf, 0, &jerr);
+                json_t *     root = json_loads(buf, 0, &jerr);
                 if (root) {
                     json_t *sim = json_object_get(root, "modbus_simulator");
                     if (!sim) {
@@ -921,6 +933,7 @@ static void sim_do_stop(bool persist)
     if (!g_sim.running) {
         return;
     }
+    clean_client();
     atomic_store(&g_sim.gen_stop, true);
     if (g_sim.gen_thread) {
         pthread_join(g_sim.gen_thread, NULL);
@@ -957,9 +970,9 @@ int neu_modbus_simulator_start(const char *ip, uint16_t port)
         return 0;
     }
     if (ip && *ip) {
-        if ((void*)g_sim.ip != (void*)ip) {
-             memset(g_sim.ip, 0, sizeof(g_sim.ip));
-             strncpy(g_sim.ip, ip, sizeof(g_sim.ip) - 1);
+        if ((void *) g_sim.ip != (void *) ip) {
+            memset(g_sim.ip, 0, sizeof(g_sim.ip));
+            strncpy(g_sim.ip, ip, sizeof(g_sim.ip) - 1);
         }
     } else {
         memset(g_sim.ip, 0, sizeof(g_sim.ip));
@@ -1011,7 +1024,7 @@ void neu_modbus_simulator_stop(void)
 }
 
 int neu_modbus_simulator_config_tags(const char **names, const char **addr_strs,
-                                     const uint16_t           *addresses,
+                                     const uint16_t *          addresses,
                                      const neu_sim_tag_type_e *types,
                                      int                       tag_count)
 {
@@ -1058,7 +1071,7 @@ char *neu_modbus_simulator_export_drivers_json(void)
     json_array_append_new(nodes, node);
     json_object_set_new(node, "plugin", json_string("Modbus TCP"));
     json_object_set_new(node, "name", json_string("ModbusTCP_Simulator"));
-    json_t     *params = json_object();
+    json_t *    params = json_object();
     const char *host = (strcmp(g_sim.ip, "0.0.0.0") == 0 || g_sim.ip[0] == '\0')
         ? "127.0.0.1"
         : g_sim.ip;

src/adapter/driver/cache.c

@@ -219,14 +219,15 @@ void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group)
     return trace;
 }
 
-void neu_driver_cache_update_change(neu_driver_cache_t *cache,
+bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
                                     const char *group, const char *tag,
                                     int64_t timestamp, neu_dvalue_t value,
                                     neu_tag_meta_t *metas, int n_meta,
                                     bool change)
 {
-    struct elem *elem = NULL;
-    tkey_t       key  = to_key(group, tag);
+    struct elem *elem        = NULL;
+    tkey_t       key         = to_key(group, tag);
+    bool         tag_changed = false;
 
     pthread_mutex_lock(&cache->mtx);
     HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
@@ -241,6 +242,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
             (sub_filter_err && elem->value.type != value.type &&
              elem->value.type != NEU_TYPE_ERROR)) {
             elem->changed = true;
+            tag_changed   = true;
         } else if (sub_filter_err && elem->value.type != value.type &&
                    elem->value.type == NEU_TYPE_ERROR) {
             switch (value.type) {
@@ -264,17 +266,20 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (memcmp(&elem->value_old.value, &value.value,
                            sizeof(value.value)) != 0) {
                     elem->changed = true;
+                    tag_changed   = true;
                 }
                 break;
             case NEU_TYPE_BYTES:
                 if (elem->value_old.value.bytes.length !=
                     value.value.bytes.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcpy(elem->value_old.value.bytes.bytes,
                                value.value.bytes.bytes,
                                value.value.bytes.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -282,11 +287,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.bools.length !=
                     value.value.bools.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.bools.bools,
                                value.value.bools.bools,
                                value.value.bools.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -294,11 +301,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.i8s.length !=
                     value.value.i8s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.i8s.i8s,
                                value.value.i8s.i8s,
                                value.value.i8s.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -306,11 +315,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.u8s.length !=
                     value.value.u8s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.u8s.u8s,
                                value.value.u8s.u8s,
                                value.value.u8s.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -318,12 +329,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.i16s.length !=
                     value.value.i16s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.i16s.i16s,
                                value.value.i16s.i16s,
                                value.value.i16s.length * sizeof(int16_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -331,12 +344,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.u16s.length !=
                     value.value.u16s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.u16s.u16s,
                                value.value.u16s.u16s,
                                value.value.u16s.length * sizeof(uint16_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -344,12 +359,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.i32s.length !=
                     value.value.i32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.i32s.i32s,
                                value.value.i32s.i32s,
                                value.value.i32s.length * sizeof(int32_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -357,12 +374,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.u32s.length !=
                     value.value.u32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.u32s.u32s,
                                value.value.u32s.u32s,
                                value.value.u32s.length * sizeof(uint32_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -370,12 +389,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.i64s.length !=
                     value.value.i64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.i64s.i64s,
                                value.value.i64s.i64s,
                                value.value.i64s.length * sizeof(int64_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -383,12 +404,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.u64s.length !=
                     value.value.u64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.u64s.u64s,
                                value.value.u64s.u64s,
                                value.value.u64s.length * sizeof(uint64_t)) !=
                         0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -396,11 +419,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.f32s.length !=
                     value.value.f32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.f32s.f32s,
                                value.value.f32s.f32s,
                                value.value.f32s.length * sizeof(float)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -408,11 +433,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.f64s.length !=
                     value.value.f64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.f64s.f64s,
                                value.value.f64s.f64s,
                                value.value.f64s.length * sizeof(double)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -420,11 +447,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.strs.length !=
                     value.value.strs.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.strs.strs,
                                value.value.strs.strs,
                                value.value.strs.length * sizeof(char *)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -432,6 +461,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (json_equal(elem->value_old.value.json, value.value.json) !=
                     0) {
                     elem->changed = true;
+                    tag_changed   = true;
                 }
                 break;
             }
@@ -439,11 +469,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.value.ptr.length !=
                     value.value.ptr.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value_old.value.ptr.ptr,
                                value.value.ptr.ptr,
                                value.value.ptr.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
 
@@ -453,10 +485,12 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.precision == 0) {
                     elem->changed =
                         elem->value_old.value.f32 != value.value.f32;
+                    tag_changed = elem->changed;
                 } else {
                     if (fabs(elem->value_old.value.f32 - value.value.f32) >
                         pow(0.1, elem->value_old.precision)) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -464,10 +498,12 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value_old.precision == 0) {
                     elem->changed =
                         elem->value_old.value.d64 != value.value.d64;
+                    tag_changed = elem->changed;
                 } else {
                     if (fabs(elem->value_old.value.d64 - value.value.d64) >
                         pow(0.1, elem->value_old.precision)) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
 
@@ -497,17 +533,20 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (memcmp(&elem->value.value, &value.value,
                            sizeof(value.value)) != 0) {
                     elem->changed = true;
+                    tag_changed   = true;
                 }
                 break;
             case NEU_TYPE_BYTES:
                 if (elem->value.value.bytes.length !=
                     value.value.bytes.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.bytes.bytes,
                                value.value.bytes.bytes,
                                value.value.bytes.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
@@ -515,140 +554,166 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 if (elem->value.value.bools.length !=
                     value.value.bools.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.bools.bools,
                                value.value.bools.bools,
                                value.value.bools.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_INT8:
                 if (elem->value.value.i8s.length != value.value.i8s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.i8s.i8s, value.value.i8s.i8s,
                                value.value.i8s.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_UINT8:
                 if (elem->value.value.u8s.length != value.value.u8s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.u8s.u8s, value.value.u8s.u8s,
                                value.value.u8s.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_INT16:
                 if (elem->value.value.i16s.length != value.value.i16s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.i16s.i16s, value.value.i16s.i16s,
                             value.value.i16s.length * sizeof(int16_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_UINT16:
                 if (elem->value.value.u16s.length != value.value.u16s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.u16s.u16s, value.value.u16s.u16s,
                             value.value.u16s.length * sizeof(uint16_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_INT32:
                 if (elem->value.value.i32s.length != value.value.i32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.i32s.i32s, value.value.i32s.i32s,
                             value.value.i32s.length * sizeof(int32_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_UINT32:
                 if (elem->value.value.u32s.length != value.value.u32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.u32s.u32s, value.value.u32s.u32s,
                             value.value.u32s.length * sizeof(uint32_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_INT64:
                 if (elem->value.value.i64s.length != value.value.i64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.i64s.i64s, value.value.i64s.i64s,
                             value.value.i64s.length * sizeof(int64_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_UINT64:
                 if (elem->value.value.u64s.length != value.value.u64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(
                             elem->value.value.u64s.u64s, value.value.u64s.u64s,
                             value.value.u64s.length * sizeof(uint64_t)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_FLOAT:
                 if (elem->value.value.f32s.length != value.value.f32s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.f32s.f32s,
                                value.value.f32s.f32s,
                                value.value.f32s.length * sizeof(float)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_DOUBLE:
                 if (elem->value.value.f64s.length != value.value.f64s.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.f64s.f64s,
                                value.value.f64s.f64s,
                                value.value.f64s.length * sizeof(double)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_ARRAY_STRING:
                 if (elem->value.value.strs.length != value.value.strs.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.strs.strs,
                                value.value.strs.strs,
                                value.value.strs.length * sizeof(char *)) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_PTR: {
                 if (elem->value.value.ptr.length != value.value.ptr.length) {
                     elem->changed = true;
+                    tag_changed   = true;
                 } else {
                     if (memcmp(elem->value.value.ptr.ptr, value.value.ptr.ptr,
                                value.value.ptr.length) != 0) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
 
@@ -657,32 +722,38 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
             case NEU_TYPE_CUSTOM: {
                 if (json_equal(elem->value.value.json, value.value.json) != 0) {
                     elem->changed = true;
+                    tag_changed   = true;
                 }
                 break;
             }
             case NEU_TYPE_FLOAT:
                 if (elem->value.precision == 0) {
                     elem->changed = elem->value.value.f32 != value.value.f32;
+                    tag_changed   = elem->changed;
                 } else {
                     if (fabs(elem->value.value.f32 - value.value.f32) >
                         pow(0.1, elem->value.precision)) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
                 break;
             case NEU_TYPE_DOUBLE:
                 if (elem->value.precision == 0) {
                     elem->changed = elem->value.value.d64 != value.value.d64;
+                    tag_changed   = elem->changed;
                 } else {
                     if (fabs(elem->value.value.d64 - value.value.d64) >
                         pow(0.1, elem->value.precision)) {
                         elem->changed = true;
+                        tag_changed   = true;
                     }
                 }
 
                 break;
             case NEU_TYPE_ERROR:
                 elem->changed = true;
+                tag_changed   = true;
                 break;
             }
         }
@@ -786,6 +857,7 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
     }
 
     pthread_mutex_unlock(&cache->mtx);
+    return tag_changed;
 }
 
 void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,

src/adapter/driver/cache.h

@@ -35,7 +35,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
                              const char *tag, int64_t timestamp,
                              neu_dvalue_t value, neu_tag_meta_t *metas,
                              int n_meta);
-void neu_driver_cache_update_change(neu_driver_cache_t *cache,
+bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
                                     const char *group, const char *tag,
                                     int64_t timestamp, neu_dvalue_t value,
                                     neu_tag_meta_t *metas, int n_meta,

src/adapter/driver/driver.c

@@ -113,6 +113,9 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
 static void update_im(neu_adapter_t *adapter, const char *group,
                       const char *tag, neu_dvalue_t value,
                       neu_tag_meta_t *metas, int n_meta);
+static void update_im_f_m(neu_adapter_t *adapter, const char *group,
+                          const char *tag, neu_dvalue_t value,
+                          neu_tag_meta_t *metas, int n_meta);
 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);
@@ -439,6 +442,107 @@ static void update_with_trace(neu_adapter_t *adapter, const char *group,
     }
 }
 
+static void update_im_f_m(neu_adapter_t *adapter, const char *group,
+                          const char *tag, neu_dvalue_t value,
+                          neu_tag_meta_t *metas, int n_meta)
+{
+    neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
+
+    if (tag == NULL) {
+        nlog_warn("update_im_f_m tag is null");
+        return;
+    }
+
+    group_t *find = NULL;
+
+    HASH_FIND_STR(driver->groups, group, find);
+    if (find == NULL) {
+        nlog_error(
+            "update_im_f_m group not found, driver: %s, group: %s, tag: %s",
+            driver->adapter.name, group, tag);
+        return;
+    }
+
+    bool changed = neu_driver_cache_update_change(driver->cache, group, tag,
+                                                  global_timestamp, value,
+                                                  metas, n_meta, false);
+    driver->adapter.cb_funs.update_metric(&driver->adapter,
+                                          NEU_METRIC_TAG_READS_TOTAL, 1, NULL);
+    if (value.type == NEU_TYPE_ERROR) {
+        return;
+    }
+
+    if (!changed) {
+        return;
+    }
+
+    UT_array *tags = neu_adapter_driver_get_ptag(driver, group, tag);
+    if (tags == NULL) {
+        return;
+    }
+    neu_datatag_t *first = utarray_front(tags);
+
+    if (first == NULL) {
+        utarray_free(tags);
+        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_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,
+    };
+    neu_reqresp_trans_data_t *data =
+        calloc(1, sizeof(neu_reqresp_trans_data_t));
+
+    data->driver = strdup(driver->adapter.name);
+    data->group  = strdup(group);
+    utarray_new(data->tags, neu_resp_tag_value_meta_icd());
+
+    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) {
+
+        data->ctx        = calloc(1, sizeof(neu_reqresp_trans_data_ctx_t));
+        data->ctx->index = utarray_len(find->apps);
+        pthread_mutex_init(&data->ctx->mtx, NULL);
+
+        if (utarray_len(find->apps) > 0) {
+            pthread_mutex_lock(&find->apps_mtx);
+
+            utarray_foreach(find->apps, sub_app_t *, app)
+            {
+                if (driver->adapter.cb_funs.responseto(
+                        &driver->adapter, &header, data, app->addr) != 0) {
+                    neu_trans_data_free(data);
+                }
+            }
+
+            pthread_mutex_unlock(&find->apps_mtx);
+        } else {
+            neu_trans_data_free(data);
+        }
+    } else {
+        utarray_free(data->tags);
+        free(data->group);
+        free(data->driver);
+    }
+
+    utarray_free(tags);
+    free(data);
+}
+
 static void update_im(neu_adapter_t *adapter, const char *group,
                       const char *tag, neu_dvalue_t value,
                       neu_tag_meta_t *metas, int n_meta)
@@ -580,6 +684,7 @@ neu_adapter_driver_t *neu_adapter_driver_create()
     driver->adapter.cb_funs.driver.fdown_open_response = fdown_open_response;
     driver->adapter.cb_funs.driver.fup_data_response   = fup_data_response;
     driver->adapter.cb_funs.driver.update_im           = update_im;
+    driver->adapter.cb_funs.driver.update_im_f_m       = update_im_f_m;
     driver->adapter.cb_funs.driver.update_with_trace   = update_with_trace;
     driver->adapter.cb_funs.driver.update_with_meta    = update_with_meta;
     driver->adapter.cb_funs.driver.scan_tags_response  = scan_tags_response;
@@ -1397,6 +1502,8 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     group_t *g = find_group(driver, cmd->group);
 
     if (g == NULL) {
+        nlog_warn("driver write tags group not exist, driver: %s, group: %s",
+                  driver->adapter.name, cmd->group);
         driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                       NEU_ERR_GROUP_NOT_EXIST);
         return NEU_ERR_GROUP_NOT_EXIST;
@@ -1417,6 +1524,9 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
                                   cmd->tags[i].value.value.d64,
                                   cmd->tags[i].value.type, tag->decimal);
         } else {
+            nlog_warn("driver write tags not exist, driver: %s, group: %s, "
+                      "tag: %s",
+                      driver->adapter.name, cmd->group, cmd->tags[i].tag);
             value_check = NEU_ERR_TAG_NOT_EXIST;
         }
 
@@ -1450,6 +1560,8 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     }
 
     if (value_err != NEU_ERR_SUCCESS) {
+        nlog_warn("driver write tags value error(%d), driver: %s, group: %s",
+                  value_err, driver->adapter.name, cmd->group);
         driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                       value_err);
         utarray_foreach(tags, neu_plugin_tag_value_t *, tv)
@@ -1463,6 +1575,9 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     }
 
     if (utarray_len(tags) != (unsigned int) cmd->n_tag) {
+        nlog_warn("driver write tags some tag not exist, driver: %s, group: "
+                  "%s",
+                  driver->adapter.name, cmd->group);
         driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                       NEU_ERR_TAG_NOT_EXIST);
         utarray_foreach(tags, neu_plugin_tag_value_t *, tv)
@@ -1662,6 +1777,8 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
     group_t *            g   = find_group(driver, cmd->group);
 
     if (g == NULL) {
+        nlog_warn("driver write tag group not exist, driver: %s, group: %s",
+                  driver->adapter.name, cmd->group);
         driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                       NEU_ERR_GROUP_NOT_EXIST);
         return NEU_ERR_GROUP_NOT_EXIST;
@@ -1669,11 +1786,16 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
     neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tag);
 
     if (tag == NULL) {
+        nlog_warn("driver write tag not exist, driver: %s, group: %s, tag: %s",
+                  driver->adapter.name, cmd->group, cmd->tag);
         driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                       NEU_ERR_TAG_NOT_EXIST);
         return NEU_ERR_TAG_NOT_EXIST;
     } else {
         if ((tag->attribute & NEU_ATTRIBUTE_WRITE) != NEU_ATTRIBUTE_WRITE) {
+            nlog_warn("driver write tag not allow write, driver: %s, group: "
+                      "%s, tag: %s",
+                      driver->adapter.name, cmd->group, cmd->tag);
             driver->adapter.cb_funs.driver.write_response(
                 &driver->adapter, req, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE);
             neu_tag_free(tag);
@@ -1685,6 +1807,9 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                                             cmd->value.type, tag->decimal);
 
         if (value_check != NEU_ERR_SUCCESS) {
+            nlog_warn("driver write tag value error(%d), driver: %s, group: "
+                      "%s, tag: %s",
+                      value_check, driver->adapter.name, cmd->group, cmd->tag);
             driver->adapter.cb_funs.driver.write_response(&driver->adapter, req,
                                                           value_check);
             neu_tag_free(tag);

src/base/msg_internal.h

@@ -57,6 +57,7 @@ extern "C" {
     XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t)                           \
     XX(NEU_REQ_ADD_NODE, neu_req_add_node_t)                                   \
     XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t)                             \
+    XX(NEU_REQ_UPDATE_NODE_TAG, neu_req_update_node_tag_t)                     \
     XX(NEU_REQ_DEL_NODE, neu_req_del_node_t)                                   \
     XX(NEU_REQ_GET_NODE, neu_req_get_node_t)                                   \
     XX(NEU_RESP_GET_NODE, neu_resp_get_node_t)                                 \

src/connection/connection.c

@@ -732,6 +732,11 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
     }
 }
 
+bool conn_tcp_server_is_listen(neu_conn_t *conn)
+{
+    return conn->tcp_server.is_listen;
+}
+
 static void conn_tcp_server_listen(neu_conn_t *conn)
 {
     if (conn->param.type == NEU_CONN_TCP_SERVER &&
@@ -746,6 +751,7 @@ static void conn_tcp_server_listen(neu_conn_t *conn)
             };
 
             fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, IPPROTO_TCP);
+            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &(int) { 1 }, sizeof(int));
 
             ret = bind(fd, (struct sockaddr *) &local, sizeof(local));
         } else if (is_ipv6(conn->param.params.tcp_server.ip)) {
@@ -756,7 +762,7 @@ static void conn_tcp_server_listen(neu_conn_t *conn)
                       &local.sin6_addr);
 
             fd = socket(AF_INET6, SOCK_STREAM | SOCK_NONBLOCK, IPPROTO_TCP);
-
+            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &(int) { 1 }, sizeof(int));
             ret = bind(fd, (struct sockaddr *) &local, sizeof(local));
         } else {
             zlog_error(conn->param.log, "invalid ip: %s",

src/core/manager.c

@@ -276,6 +276,26 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_msg_free(msg);
         break;
     }
+    case NEU_REQ_UPDATE_NODE_TAG: {
+        neu_req_update_node_tag_t *cmd =
+            (neu_req_update_node_tag_t *) &header[1];
+        int ret = neu_node_manager_update_tags(manager->node_manager, cmd->node,
+                                               cmd->tags);
+        if (ret == 0) {
+            nlog_info("update node %s tags to %s", cmd->node, cmd->tags);
+            manager_storage_update_node_tags(manager, cmd->node, cmd->tags);
+            neu_resp_error_t e = { .error = NEU_ERR_SUCCESS };
+            header->type       = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(manager, header, &e);
+        } else {
+            neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
+            header->type       = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(manager, header, &e);
+        }
+        break;
+    }
     case NEU_REQ_ADD_PLUGIN: {
         neu_req_add_plugin_t *cmd        = (neu_req_add_plugin_t *) &header[1];
         neu_resp_error_t      e          = { 0 };
@@ -742,12 +762,12 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         nlog_notice("add node name:%s plugin:%s", cmd->node, cmd->plugin);
         int error =
             neu_manager_add_node(manager, cmd->node, cmd->plugin, cmd->setting,
-                                 NEU_NODE_RUNNING_STATE_INIT, false);
+                                 cmd->tags, NEU_NODE_RUNNING_STATE_INIT, false);
 
         neu_resp_error_t e = { .error = error };
 
         if (error == NEU_ERR_SUCCESS) {
-            manager_storage_add_node(manager, cmd->node);
+            manager_storage_add_node(manager, cmd->node, cmd->tags);
             if (cmd->setting) {
                 adapter_storage_setting(cmd->node, cmd->setting);
             }
@@ -1600,7 +1620,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         if (NEU_ERR_SUCCESS == e.error) {
             for (uint16_t i = 0; i < cmd->n_driver; ++i) {
                 neu_req_driver_t *driver = &cmd->drivers[i];
-                manager_storage_add_node(manager, driver->node);
+                manager_storage_add_node(manager, driver->node, driver->tags);
                 adapter_storage_setting(driver->node, driver->setting);
                 for (uint16_t j = 0; j < driver->n_group; j++) {
                     adapter_storage_add_group(driver->node,
@@ -1787,7 +1807,7 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
 
     neu_node_manager_add_single(manager->node_manager, adapter, display);
     if (display) {
-        manager_storage_add_node(manager, name);
+        manager_storage_add_node(manager, name, "");
     }
 
     neu_adapter_init(adapter, false);

src/core/manager_internal.c

@@ -48,7 +48,8 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager)
 
 int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
                          const char *plugin_name, const char *setting,
-                         neu_node_running_state_e state, bool load)
+                         const char *tags, neu_node_running_state_e state,
+                         bool load)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -84,7 +85,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
     if (adapter == NULL) {
         return neu_adapter_error();
     }
-    neu_node_manager_add(manager->node_manager, adapter);
+    neu_node_manager_add(manager->node_manager, adapter, tags);
     neu_adapter_init(adapter, state);
 
     if (NULL != setting &&
@@ -511,7 +512,7 @@ static inline int add_driver(neu_manager_t *manager, neu_req_driver_t *driver)
     }
 
     ret = neu_manager_add_node(manager, driver->node, driver->plugin,
-                               driver->setting, false, false);
+                               driver->setting, driver->tags, false, false);
     if (0 != ret) {
         return ret;
     }

src/core/manager_internal.h

@@ -53,7 +53,8 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager);
 
 int       neu_manager_add_node(neu_manager_t *manager, const char *node_name,
                                const char *plugin_name, const char *setting,
-                               neu_node_running_state_e state, bool load);
+                               const char *tags, neu_node_running_state_e state,
+                               bool load);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, int type,
                                 const char *plugin, const char *node,

src/core/node_manager.c

@@ -35,6 +35,8 @@ typedef struct node_entity {
     bool               is_monitor;
     struct sockaddr_un addr;
 
+    char *tags;
+
     UT_hash_handle hh;
 } node_entity_t;
 
@@ -43,6 +45,24 @@ struct neu_node_manager {
     UT_array *     monitors;
 };
 
+int neu_node_manager_update_tags(neu_node_manager_t *mgr, const char *name,
+                                 const char *tags)
+{
+    node_entity_t *node = NULL;
+
+    HASH_FIND_STR(mgr->nodes, name, node);
+    if (NULL == node) {
+        return NEU_ERR_NODE_NOT_EXIST;
+    }
+
+    if (node->tags) {
+        free(node->tags);
+    }
+    node->tags = strdup(tags);
+
+    return 0;
+}
+
 neu_node_manager_t *neu_node_manager_create()
 {
     neu_node_manager_t *node_manager = calloc(1, sizeof(neu_node_manager_t));
@@ -69,7 +89,8 @@ void neu_node_manager_destroy(neu_node_manager_t *mgr)
     free(mgr);
 }
 
-int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter)
+int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter,
+                         const char *tags)
 {
     node_entity_t *node = calloc(1, sizeof(node_entity_t));
 
@@ -78,6 +99,7 @@ int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter)
     node->display = true;
     node->is_monitor =
         (0 == strcmp(node->adapter->module->module_name, "Monitor"));
+    node->tags = tags ? strdup(tags) : NULL;
 
     HASH_ADD_STR(mgr->nodes, name, node);
 
@@ -166,6 +188,9 @@ void neu_node_manager_del(neu_node_manager_t *mgr, const char *name)
                           1);
         }
         free(node->name);
+        if (node->tags) {
+            free(node->tags);
+        }
         free(node);
     }
 }
@@ -225,6 +250,26 @@ static int neu_resp_node_info_sort(const void *a, const void *b)
     return 0;
 }
 
+static bool find_tag_in_tags(const char *tags, const char *tag)
+{
+    if (strlen(tags) == 0) {
+        return false;
+    }
+
+    char *tags_copy = strdup(tags);
+    char *saveptr   = NULL;
+    char *token     = strtok_r(tags_copy, ",", &saveptr);
+    while (token != NULL) {
+        if (strcmp(token, tag) == 0) {
+            free(tags_copy);
+            return true;
+        }
+        token = strtok_r(NULL, ",", &saveptr);
+    }
+    free(tags_copy);
+    return false;
+}
+
 UT_array *neu_node_manager_filter(neu_node_manager_t *mgr, int type,
                                   const char *plugin, const char *node,
                                   bool sort_delay, bool q_state, int state,
@@ -245,9 +290,38 @@ UT_array *neu_node_manager_filter(neu_node_manager_t *mgr, int type,
                     strcmp(el->adapter->module->module_name, plugin) != 0) {
                     continue;
                 }
-                if (strlen(node) > 0 &&
-                    strstr(el->adapter->name, node) == NULL) {
-                    continue;
+                if (strlen(node) > 0) {
+                    char *name_find = strstr(el->adapter->name, node);
+                    if (name_find == NULL &&
+                        (el->tags == NULL || strlen(el->tags) == 0)) {
+                        continue;
+                    }
+
+                    if (el->tags != NULL) {
+                        char node_tmp[NEU_NODE_NAME_LEN] = { 0 };
+                        strncpy(node_tmp, node, NEU_NODE_NAME_LEN - 1);
+                        size_t index                                  = 0;
+                        char   tag_split_array[10][NEU_NODE_TAGS_LEN] = { 0 };
+                        char * token = strtok(node_tmp, ",");
+                        while (token != NULL && index < 10) {
+                            strcpy(tag_split_array[index], token);
+                            index++;
+                            token = strtok(NULL, ",");
+                        }
+
+                        bool tag_found = true;
+                        for (size_t i = 0; i < index; i++) {
+                            if (find_tag_in_tags(el->tags,
+                                                 tag_split_array[i]) == false) {
+                                tag_found = false;
+                                break;
+                            }
+                        }
+
+                        if (!tag_found && name_find == NULL) {
+                            continue;
+                        }
+                    }
                 }
                 if (q_state) {
                     if (el->adapter->state !=
@@ -280,6 +354,7 @@ UT_array *neu_node_manager_filter(neu_node_manager_t *mgr, int type,
 
                 neu_resp_node_info_t info = { 0 };
                 strcpy(info.node, el->adapter->name);
+                strcpy(info.tags, el->tags ? el->tags : "");
                 strcpy(info.plugin, el->adapter->module->module_name);
                 if (sort_delay) {
                     neu_metric_entry_t *e = NULL;

src/core/node_manager.h

@@ -33,7 +33,10 @@ typedef struct neu_node_manager neu_node_manager_t;
 neu_node_manager_t *neu_node_manager_create();
 void                neu_node_manager_destroy(neu_node_manager_t *mgr);
 
-int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter);
+int neu_node_manager_update_tags(neu_node_manager_t *mgr, const char *name,
+                                 const char *tags);
+int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter,
+                         const char *tags);
 int neu_node_manager_add_static(neu_node_manager_t *mgr,
                                 neu_adapter_t *     adapter);
 int neu_node_manager_add_single(neu_node_manager_t *mgr, neu_adapter_t *adapter,

src/core/storage.c

@@ -38,7 +38,8 @@ void manager_strorage_plugin(neu_manager_t *manager)
     return;
 }
 
-void manager_storage_add_node(neu_manager_t *manager, const char *node)
+void manager_storage_add_node(neu_manager_t *manager, const char *node,
+                              const char *tags)
 {
     int                     rv        = 0;
     neu_persist_node_info_t node_info = {};
@@ -48,6 +49,7 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
         nlog_error("unable to get adapter:%s info", node);
         return;
     }
+    node_info.tags = tags ? strdup(tags) : NULL;
 
     rv = neu_persister_store_node(&node_info);
     if (0 != rv) {
@@ -64,6 +66,13 @@ void manager_storage_update_node(neu_manager_t *manager, const char *node,
     neu_persister_update_node(node, new_name);
 }
 
+void manager_storage_update_node_tags(neu_manager_t *manager, const char *node,
+                                      const char *new_tags)
+{
+    (void) manager;
+    neu_persister_update_node_tags(node, new_tags);
+}
+
 void manager_storage_del_node(neu_manager_t *manager, const char *node)
 {
     (void) manager;
@@ -146,7 +155,7 @@ int manager_load_node(neu_manager_t *manager)
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
         rv                    = neu_manager_add_node(manager, node_info->name,
-                                  node_info->plugin_name, NULL,
+                                  node_info->plugin_name, NULL, node_info->tags,
                                   node_info->state, true);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",

src/core/storage.h

@@ -26,7 +26,10 @@
 
 void manager_strorage_plugin(neu_manager_t *manager);
 void manager_storage_del_node(neu_manager_t *manager, const char *node);
-void manager_storage_add_node(neu_manager_t *manager, const char *node);
+void manager_storage_add_node(neu_manager_t *manager, const char *node,
+                              const char *tags);
+void manager_storage_update_node_tags(neu_manager_t *manager, const char *node,
+                                      const char *new_tags);
 void manager_storage_update_node(neu_manager_t *manager, const char *node,
                                  const char *new_name);
 void manager_storage_subscribe(neu_manager_t *manager, const char *app,

src/main.c

@@ -59,6 +59,7 @@ static void sig_handler(int sig)
 
     if (!sig_trigger) {
         if (sig == SIGINT || sig == SIGTERM) {
+            neu_modbus_simulator_fini();
             neu_manager_destroy(g_manager);
             neu_persister_destroy();
             zlog_fini();

src/parser/neu_json_global_config.c

@@ -589,6 +589,9 @@ static void fini_app_group(neu_json_app_group_t *grp)
     free(grp->driver);
     free(grp->group);
     free(grp->params);
+    if (grp->static_tags) {
+        free(grp->static_tags);
+    }
 }
 
 int neu_json_encode_app(void *node_json, void *param)
@@ -644,6 +647,15 @@ int neu_json_encode_app(void *node_json, void *param)
             }
         }
 
+        if (grp->static_tags && strlen(grp->static_tags) > 0) {
+            if (0 !=
+                neu_json_load_key(grp_json, "static_tags", grp->static_tags,
+                                  false)) {
+                json_decref(grp_json);
+                return -1;
+            }
+        }
+
         if (0 != json_array_append_new(groups_json, grp_json)) {
             return -1;
         }
@@ -715,6 +727,8 @@ int neu_json_decode_app_json(void *node_json, neu_json_app_t *app_p)
         groups[i].driver = grp_elems[0].v.val_str;
         groups[i].group  = grp_elems[1].v.val_str;
         neu_json_dump_key(grp_json, "params", &groups[i].params, false);
+        neu_json_dump_key(grp_json, "static_tags", &groups[i].static_tags,
+                          false);
     }
 
     app_p->subscriptions.n_group = n_group;

src/parser/neu_json_global_config.h

@@ -101,6 +101,7 @@ typedef struct {
     char *driver;
     char *group;
     char *params;
+    char *static_tags;
 } neu_json_app_group_t;
 
 typedef struct {

src/parser/neu_json_node.c

@@ -36,16 +36,24 @@ int neu_json_encode_add_node_req(void *json_object, void *param)
     int                      ret = 0;
     neu_json_add_node_req_t *req = param;
 
-    neu_json_elem_t req_elems[] = { {
-                                        .name      = "plugin",
-                                        .t         = NEU_JSON_STR,
-                                        .v.val_str = req->plugin,
-                                    },
-                                    {
-                                        .name      = "name",
-                                        .t         = NEU_JSON_STR,
-                                        .v.val_str = req->name,
-                                    } };
+    neu_json_elem_t req_elems[] = {
+        {
+            .name      = "plugin",
+            .t         = NEU_JSON_STR,
+            .v.val_str = req->plugin,
+        },
+        {
+            .name      = "name",
+            .t         = NEU_JSON_STR,
+            .v.val_str = req->name,
+        },
+        {
+            .name      = "tags",
+            .t         = NEU_JSON_STR,
+            .v.val_str = req->tags,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+        },
+    };
 
     ret = neu_json_encode_field(json_object, req_elems,
                                 NEU_JSON_ELEM_SIZE(req_elems));
@@ -66,14 +74,22 @@ int neu_json_decode_add_node_req_json(void *                   json_obj,
 {
     int ret = 0;
 
-    neu_json_elem_t req_elems[] = { {
-                                        .name = "plugin",
-                                        .t    = NEU_JSON_STR,
-                                    },
-                                    {
-                                        .name = "name",
-                                        .t    = NEU_JSON_STR,
-                                    } };
+    neu_json_elem_t req_elems[] = {
+        {
+            .name = "plugin",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name = "name",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name      = "tags",
+            .t         = NEU_JSON_STR,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+            .v.val_str = NULL,
+        },
+    };
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
     if (ret != 0) {
@@ -82,6 +98,9 @@ int neu_json_decode_add_node_req_json(void *                   json_obj,
 
     req->plugin = req_elems[0].v.val_str;
     req->name   = req_elems[1].v.val_str;
+    if (req_elems[2].v.val_str != NULL) {
+        req->tags = req_elems[2].v.val_str;
+    }
 
     ret = neu_json_dump_key(json_obj, "params", &req->setting, false);
     if (0 != ret) {
@@ -126,6 +145,8 @@ void neu_json_decode_add_node_req_fini(neu_json_add_node_req_t *req)
 {
     free(req->plugin);
     free(req->name);
+    if (req->tags)
+        free(req->tags);
     free(req->setting);
 }
 
@@ -330,6 +351,11 @@ int neu_json_encode_get_nodes_resp(void *json_object, void *param)
                 .t          = NEU_JSON_BOOL,
                 .v.val_bool = p_node->support_import_tags,
             },
+            {
+                .name      = "tags",
+                .t         = NEU_JSON_STR,
+                .v.val_str = p_node->tags,
+            },
         };
         node_array = neu_json_encode_array(node_array, node_elems,
                                            NEU_JSON_ELEM_SIZE(node_elems));
@@ -663,3 +689,63 @@ int neu_json_encode_get_node_setting_resp(void *json_object, void *param)
 
     return ret;
 }
+
+int neu_json_decode_update_node_tag_req(char *                           buf,
+                                        neu_json_update_node_tag_req_t **result)
+{
+    int                             ret      = 0;
+    void *                          json_obj = NULL;
+    neu_json_update_node_tag_req_t *req =
+        calloc(1, sizeof(neu_json_update_node_tag_req_t));
+    if (req == NULL) {
+        return -1;
+    }
+
+    json_obj = neu_json_decode_new(buf);
+    if (NULL == json_obj) {
+        free(req);
+        return -1;
+    }
+
+    neu_json_elem_t req_elems[] = {
+        {
+            .name = "name",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name      = "tags",
+            .t         = NEU_JSON_STR,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+            .v.val_str = NULL,
+        },
+    };
+
+    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
+                                  req_elems);
+    if (ret != 0) {
+        free(req);
+        neu_json_decode_free(json_obj);
+        return -1;
+    }
+
+    req->name = req_elems[0].v.val_str;
+    if (req_elems[1].v.val_str != NULL) {
+        req->tags = req_elems[1].v.val_str;
+    }
+
+    *result = req;
+
+    neu_json_decode_free(json_obj);
+    return 0;
+}
+
+void neu_json_decode_update_node_tag_req_free(
+    neu_json_update_node_tag_req_t *req)
+{
+    if (req) {
+        free(req->name);
+        if (req->tags)
+            free(req->tags);
+        free(req);
+    }
+}

src/parser/neu_json_node.h

@@ -36,6 +36,7 @@ typedef struct {
     char *name;
     char *plugin;
     char *setting;
+    char *tags;
 } neu_json_add_node_req_t;
 
 int  neu_json_encode_add_node_req(void *json_object, void *param);
@@ -45,6 +46,16 @@ int  neu_json_decode_add_node_req(char *buf, neu_json_add_node_req_t **result);
 void neu_json_decode_add_node_req_fini(neu_json_add_node_req_t *req);
 void neu_json_decode_add_node_req_free(neu_json_add_node_req_t *req);
 
+typedef struct {
+    char *name;
+    char *tags;
+} neu_json_update_node_tag_req_t;
+
+int neu_json_decode_update_node_tag_req(
+    char *buf, neu_json_update_node_tag_req_t **result);
+void neu_json_decode_update_node_tag_req_free(
+    neu_json_update_node_tag_req_t *req);
+
 typedef struct {
     char *name;
 } neu_json_del_node_req_t;
@@ -87,6 +98,7 @@ typedef struct {
     char *name;
     char *plugin;
     bool  support_import_tags;
+    char *tags;
 } neu_json_get_nodes_resp_node_t;
 
 typedef struct {

src/parser/neu_json_rw.c

@@ -495,7 +495,11 @@ void neu_json_decode_write_req_free(neu_json_write_req_t *req)
     if (req->t == NEU_JSON_STR) {
         free(req->value.val_str);
     }
-    if (req->t == NEU_JSON_ARRAY_STR && req->value.val_array_str.length > 0) {
+    if (req->t == NEU_JSON_ARRAY_STR &&
+        req->value.val_array_str.p_strs != NULL) {
+        for (size_t i = 0; i < req->value.val_array_str.length; i++) {
+            free(req->value.val_array_str.p_strs[i]);
+        }
         free(req->value.val_array_str.p_strs);
     }
     if (req->t == NEU_JSON_ARRAY_INT8 && req->value.val_array_int8.length > 0) {
@@ -705,6 +709,13 @@ void neu_json_decode_write_tags_req_free(neu_json_write_tags_req_t *req)
             req->tags[i].value.val_array_bool.length > 0) {
             free(req->tags[i].value.val_array_bool.bools);
         }
+        if (req->tags[i].t == NEU_JSON_ARRAY_STR &&
+            req->tags[i].value.val_array_str.p_strs != NULL) {
+            for (int j = 0; j < req->tags[i].value.val_array_str.length; j++) {
+                free(req->tags[i].value.val_array_str.p_strs[j]);
+            }
+            free(req->tags[i].value.val_array_str.p_strs);
+        }
     }
     free(req->tags);
     free(req);
@@ -1430,6 +1441,15 @@ void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req)
                 req->groups[i].tags[k].value.val_array_bool.length > 0) {
                 free(req->groups[i].tags[k].value.val_array_bool.bools);
             }
+            if (req->groups[i].tags[k].t == NEU_JSON_ARRAY_STR &&
+                req->groups[i].tags[k].value.val_array_str.p_strs != NULL) {
+                for (int j = 0;
+                     j < req->groups[i].tags[k].value.val_array_str.length;
+                     j++) {
+                    free(req->groups[i].tags[k].value.val_array_str.p_strs[j]);
+                }
+                free(req->groups[i].tags[k].value.val_array_str.p_strs);
+            }
         }
         if (req->groups[i].n_tag > 0) {
             free(req->groups[i].tags);

src/persist/persist.c

@@ -346,6 +346,11 @@ int neu_persister_update_node(const char *node_name, const char *new_name)
     return g_impl->vtbl->update_node(g_impl, node_name, new_name);
 }
 
+int neu_persister_update_node_tags(const char *node_name, const char *tags)
+{
+    return g_impl->vtbl->update_node_tags(g_impl, node_name, tags);
+}
+
 int neu_persister_update_node_state(const char *node_name, int state)
 {
     return g_impl->vtbl->update_node_state(g_impl, node_name, state);

src/persist/persist_impl.h

@@ -73,6 +73,8 @@ struct neu_persister_vtbl_s {
      */
     int (*update_node)(neu_persister_t *self, const char *node_name,
                        const char *new_name);
+    int (*update_node_tags)(neu_persister_t *self, const char *node_name,
+                            const char *tags);
 
     /**
      * Update node state.

src/persist/sqlite.c

@@ -458,6 +458,7 @@ static struct neu_persister_vtbl_s g_sqlite_persister_vtbl = {
     .load_auth_user           = neu_sqlite_persister_load_auth_user,
     .load_auth_users_by_app   = neu_sqlite_persister_load_auth_users_by_app,
     .delete_auth_user         = neu_sqlite_persister_delete_auth_user,
+    .update_node_tags         = neu_sqlite_persister_update_node_tags,
 };
 
 neu_persister_t *neu_sqlite_persister_create(const char *schema_dir)
@@ -496,9 +497,10 @@ int neu_sqlite_persister_store_node(neu_persister_t *        self,
 {
     int rv = 0;
     rv     = execute_sql(((neu_sqlite_persister_t *) self)->db,
-                     "INSERT INTO nodes (name, type, state, plugin_name) "
-                     "VALUES (%Q, %i, %i, %Q)",
-                     info->name, info->type, info->state, info->plugin_name);
+                     "INSERT INTO nodes (name, type, state, plugin_name, tags) "
+                     "VALUES (%Q, %i, %i, %Q, %Q)",
+                     info->name, info->type, info->state, info->plugin_name,
+                     info->tags);
     return rv;
 }
 
@@ -514,9 +516,10 @@ int neu_sqlite_persister_load_nodes(neu_persister_t *self,
 {
     neu_sqlite_persister_t *persister = (neu_sqlite_persister_t *) self;
 
-    int           rv    = 0;
-    sqlite3_stmt *stmt  = NULL;
-    const char *  query = "SELECT name, type, state, plugin_name FROM nodes;";
+    int           rv   = 0;
+    sqlite3_stmt *stmt = NULL;
+    const char *  query =
+        "SELECT name, type, state, plugin_name, tags FROM nodes;";
 
     utarray_new(*node_infos, &node_info_icd);
 
@@ -543,10 +546,18 @@ int neu_sqlite_persister_load_nodes(neu_persister_t *self,
             break;
         }
 
+        char *tags = (char *) sqlite3_column_text(stmt, 4);
+
         info.name        = name;
         info.type        = sqlite3_column_int(stmt, 1);
         info.state       = sqlite3_column_int(stmt, 2);
         info.plugin_name = plugin_name;
+        if (tags) {
+            info.tags = strdup(tags);
+        } else {
+            info.tags = NULL;
+        }
+
         utarray_push_back(*node_infos, &info);
 
         step = sqlite3_step(stmt);
@@ -580,6 +591,15 @@ int neu_sqlite_persister_update_node(neu_persister_t *self,
                        node_name);
 }
 
+int neu_sqlite_persister_update_node_tags(neu_persister_t *self,
+                                          const char *     node_name,
+                                          const char *     tags)
+{
+    return execute_sql(((neu_sqlite_persister_t *) self)->db,
+                       "UPDATE nodes SET tags=%Q WHERE name=%Q;", tags,
+                       node_name);
+}
+
 int neu_sqlite_persister_update_node_state(neu_persister_t *self,
                                            const char *node_name, int state)
 {

src/persist/sqlite.h

@@ -45,6 +45,9 @@ int neu_sqlite_persister_delete_node(neu_persister_t *self,
 int neu_sqlite_persister_update_node(neu_persister_t *self,
                                      const char *     node_name,
                                      const char *     new_name);
+int neu_sqlite_persister_update_node_tags(neu_persister_t *self,
+                                          const char *     node_name,
+                                          const char *     tags);
 int neu_sqlite_persister_update_node_state(neu_persister_t *self,
                                            const char *node_name, int state);
 int neu_sqlite_persister_store_tag(neu_persister_t *    self,

src/utils/http.c

@@ -340,6 +340,9 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_USER_NO_PERMISSION:
     case NEU_ERR_INVALID_USER_LEN:
     case NEU_ERR_PORT_IN_USE:
+    case NEU_ERR_NODE_TAGS_TOO_MANY:
+    case NEU_ERR_NODE_TAGS_INVALID:
+    case NEU_ERR_NODE_TAGS_TOO_LONG:
         status = NNG_HTTP_STATUS_BAD_REQUEST;
         break;
     case NEU_ERR_FILE_NOT_EXIST:

src/utils/json.c

@@ -187,7 +187,11 @@ static json_t *encode_object(json_t *object, neu_json_elem_t ele)
         json_object_set_new(ob, ele.name, json_integer(ele.v.val_int));
         break;
     case NEU_JSON_STR:
-        json_object_set_new(ob, ele.name, json_string(ele.v.val_str));
+        if (ele.v.val_str != NULL) {
+            json_object_set_new(ob, ele.name, json_string(ele.v.val_str));
+        } else {
+            json_object_set_new(ob, ele.name, json_string(""));
+        }
         break;
     case NEU_JSON_FLOAT: {
         double t = ele.v.val_float;

tests/ft/http_api/test_http_api.py

@@ -133,6 +133,7 @@ global_config_del_modbus = {
 driver_1 = {
     "name": "modbus-driver-1",
     "plugin": "Modbus TCP",
+    "tags": "",
     "params": {
         "connection_mode": 0,
         "transport_mode": 0,
@@ -180,6 +181,7 @@ driver_1 = {
 driver_2 = {
     "name": "modbus-driver-2",
     "plugin": "Modbus TCP",
+    "tags": "",
     "params": {
         "connection_mode": 0,
         "transport_mode": 0,

version

@@ -1 +1 @@
-2.14.0-alpha
+2.14.0

zlog.conf

@@ -9,5 +9,4 @@ syslog = "<P>%d(%Y-%m-%dT%T%z) %H neuron[%p]: level=%v msg="[%c] %m" file="%f:%L
 
 [rules]
 
-*.*     "./logs/%c.log", 25MB * 1 ~ "./logs/%c.#2r.log"; simple
-*.*     $remote_syslog,"%V"; syslog
+*.*     "./logs/%c.log", 50MB * 2 ~ "./logs/%c.#2r.log"; simple