Merge pull request #1644 from fengzeroz/v2.6

version: 2.6.2

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.6.1
-  UI_VERSION: 2.6.1
+  VERSION: 2.6.2
+  UI_VERSION: 2.6.2
   BRANCH: v2.6
 
 jobs:

CMakeLists.txt

@@ -41,7 +41,7 @@ if(USE_GCOV)
 endif(USE_GCOV)
 
 set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wall -Wextra -g")
-set(CMAKE_C_FLAGS_RELEASE "${CMAKE_C_FLAGS} -O1")
+set(CMAKE_C_FLAGS_RELEASE "${CMAKE_C_FLAGS} -O0")
 
 if(NOT DISABLE_ASAN)
   set(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS} -fsanitize=address")

ft/modbus_tcp_test.robot

@@ -154,6 +154,22 @@ Read all type tag with same slave id in input register and input, it should be s
 	${input_bit_4}    input_bit_4    ${modbus_tcp_node}    Compare Tag Value As Int    0
 	${input_bit_5}    input_bit_5    ${modbus_tcp_node}    Compare Tag Value As Int    0	
 
+Read in sync mode, it should fail.
+	${res}=	Add Group  ${modbus_tcp_node}  group  100
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	${res}=	Add Tags  ${modbus_tcp_node}  group  ${input_register_int16},${input_register_uint16},${input_register_int32},${input_register_uint32},${input_register_float},${input_register_string},${input_register_bit}
+	Check Response Status  ${res}  200
+	Check Error Code  ${res}  ${NEU_ERR_SUCCESS}
+
+	Sleep 	290ms
+	${res}=	Read Tags  ${modbus_tcp_node}  group  true
+	Check Response Status  ${res}  200
+	Check Response Status  ${res}  ${NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC}
+
+	[Teardown]	Del Group  ${modbus_tcp_node}  group
+
 Read some tags at time, it will be success.
 	[Template] 	Read some ${tags} from ${node}, it will be success.
 	${input_register_int16},${input_register_uint16},${input_register_int32},${input_register_uint32},${input_register_float},${input_register_string},${input_register_bit} 	${modbus_tcp_node}

ft/resource/api.resource

@@ -197,8 +197,8 @@ Del Tags
   [RETURN]            ${res}[0]
 
 Read Tags
-  [Arguments]         ${node}              ${group}
-  POST                /api/v2/read            {"node": "${node}", "group": "${group}"}
+  [Arguments]         ${node}              ${group}                      ${sync}=false
+  POST                /api/v2/read            {"node": "${node}", "group": "${group}", "sync": ${sync}}
   ${res} =            Object  response body
   [RETURN]            ${res}[0]
 

ft/resource/error.resource

@@ -95,5 +95,6 @@ ${NEU_ERR_PLUGIN_NOT_FOUND}                     3014
 ${NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE}           3015
 ${NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE}          3016
 ${NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS}        3017
+${NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC}         3018
 
 ${NEU_ERR_MQTT_SUBSCRIBE_FAILURE}               4010

include/neuron/adapter.h

@@ -47,6 +47,7 @@ typedef enum neu_reqresp_type {
     NEU_RESP_READ_GROUP,
     NEU_REQ_WRITE_TAG,
     NEU_REQ_WRITE_TAGS,
+    NEU_REQ_WRITE_GTAGS,
 
     NEU_REQ_SUBSCRIBE_GROUP,
     NEU_REQ_UNSUBSCRIBE_GROUP,
@@ -153,6 +154,7 @@ static const char *neu_reqresp_type_string_t[] = {
     [NEU_RESP_READ_GROUP] = "NEU_RESP_READ_GROUP",
     [NEU_REQ_WRITE_TAG]   = "NEU_REQ_WRITE_TAG",
     [NEU_REQ_WRITE_TAGS]  = "NEU_REQ_WRITE_TAGS",
+    [NEU_REQ_WRITE_GTAGS] = "NEU_REQ_WRITE_GTAGS",
 
     [NEU_REQ_SUBSCRIBE_GROUP]        = "NEU_REQ_SUBSCRIBE_GROUP",
     [NEU_REQ_UNSUBSCRIBE_GROUP]      = "NEU_REQ_UNSUBSCRIBE_GROUP",
@@ -600,6 +602,7 @@ typedef struct neu_req_get_node_state {
 typedef struct neu_resp_get_node_state {
     neu_node_state_t state;
     uint16_t         rtt; // round trip time in milliseconds
+    char             core_level[NEU_LOG_LEVEL_LEN];
 } neu_resp_get_node_state_t;
 
 typedef struct neu_req_get_nodes_state {
@@ -619,6 +622,7 @@ inline static UT_icd neu_nodes_state_t_icd()
 }
 typedef struct {
     UT_array *states; // array of neu_nodes_state_t
+    char      core_level[NEU_LOG_LEVEL_LEN];
 } neu_resp_get_nodes_state_t, neu_reqresp_nodes_state_t;
 
 typedef struct {
@@ -769,6 +773,7 @@ static inline void neu_req_inst_templates_fini(neu_req_inst_templates_t *req)
 typedef struct neu_req_read_group {
     char driver[NEU_NODE_NAME_LEN];
     char group[NEU_GROUP_NAME_LEN];
+    bool sync;
 } neu_req_read_group_t;
 
 typedef struct neu_req_write_tag {
@@ -803,6 +808,20 @@ typedef struct neu_req_write_tags {
     neu_resp_tag_value_t *tags;
 } neu_req_write_tags_t;
 
+typedef struct {
+    char group[NEU_GROUP_NAME_LEN];
+
+    int                   n_tag;
+    neu_resp_tag_value_t *tags;
+} neu_req_gtag_group_t;
+
+typedef struct {
+    char driver[NEU_NODE_NAME_LEN];
+
+    int                   n_group;
+    neu_req_gtag_group_t *groups;
+} neu_req_write_gtags_t;
+
 typedef struct {
     char driver[NEU_NODE_NAME_LEN];
     char group[NEU_GROUP_NAME_LEN];
@@ -1046,6 +1065,7 @@ typedef struct {
 typedef struct neu_req_update_log_level {
     char node[NEU_NODE_NAME_LEN];
     char log_level[NEU_LOG_LEVEL_LEN];
+    bool core;
 } neu_req_update_log_level_t;
 
 void  neu_msg_gen(neu_reqresp_head_t *header, void *data);

include/neuron/errcodes.h

@@ -124,6 +124,7 @@ typedef enum {
     NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE     = 3015,
     NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE    = 3016,
     NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS  = 3017,
+    NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC   = 3018,
 
     NEU_ERR_MQTT_FAILURE                        = 4000,
     NEU_ERR_MQTT_NO_CERTFILESET                 = 4001,

include/neuron/json/json.h

@@ -78,6 +78,8 @@ void *neu_json_array();
 int   neu_json_decode_by_json(void *json, int size, neu_json_elem_t *ele);
 int   neu_json_decode(char *buf, int size, neu_json_elem_t *ele);
 int   neu_json_decode_array_size_by_json(void *json, char *child);
+int   neu_json_decode_array_elem(void *json, int index, int size,
+                                 neu_json_elem_t *ele);
 int   neu_json_decode_array_by_json(void *json, char *name, int index, int size,
                                     neu_json_elem_t *ele);
 int   neu_json_decode_array_size(char *buf, char *child);

include/neuron/json/neu_json_rw.h

@@ -94,6 +94,24 @@ int  neu_json_decode_write_tags_req_json(void *                      json_obj,
                                          neu_json_write_tags_req_t **result);
 void neu_json_decode_write_tags_req_free(neu_json_write_tags_req_t *req);
 
+typedef struct {
+    char *                      group;
+    int                         n_tag;
+    neu_json_write_tags_elem_t *tags;
+} neu_json_write_gtags_elem_t;
+typedef struct {
+    char *node;
+
+    int                          n_group;
+    neu_json_write_gtags_elem_t *groups;
+} neu_json_write_gtags_req_t;
+
+int  neu_json_decode_write_gtags_req(char *                       buf,
+                                     neu_json_write_gtags_req_t **result);
+int  neu_json_decode_write_gtags_req_json(void *                       json_obj,
+                                          neu_json_write_gtags_req_t **result);
+void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req);
+
 typedef struct {
     union {
         neu_json_write_req_t      single;
@@ -108,6 +126,7 @@ void neu_json_decode_write_free(neu_json_write_t *req);
 typedef struct {
     char *group;
     char *node;
+    bool  sync;
 } neu_json_read_req_t;
 
 int  neu_json_decode_read_req(char *buf, neu_json_read_req_t **result);

include/neuron/plugin.h

@@ -85,6 +85,7 @@ typedef struct neu_plugin_intf_funs {
         struct {
             int (*validate_tag)(neu_plugin_t *plugin, neu_datatag_t *tag);
             int (*group_timer)(neu_plugin_t *plugin, neu_plugin_group_t *group);
+            int (*group_sync)(neu_plugin_t *plugin, neu_plugin_group_t *group);
             int (*write_tag)(neu_plugin_t *plugin, void *req,
                              neu_datatag_t *tag, neu_value_u value);
             int (*write_tags)(

include/neuron/utils/utextend.h

@@ -33,8 +33,12 @@ static inline UT_array *utarray_clone(UT_array *array)
     result->i   = array->i;
     result->n   = array->n;
     result->icd = array->icd;
-    result->d   = (char *) calloc(1, array->n * array->icd.sz);
-    memcpy(result->d, array->d, array->n * array->icd.sz);
+    if (array->n > 0) {
+        result->d = (char *) calloc(1, array->n * array->icd.sz);
+        memcpy(result->d, array->d, array->n * array->icd.sz);
+    } else {
+        result->d = NULL;
+    }
 
     return result;
 }
@@ -75,4 +79,15 @@ static inline void *utarray_steal(UT_array *array)
     for ((el) = (head); (el) != NULL; (el) = DECLTYPE(el)((el)->hh.next))
 #endif
 
+static inline UT_array *utarray_deep_clone(UT_array *array)
+{
+    UT_array *result = NULL;
+
+    utarray_new(result, &array->icd);
+
+    utarray_foreach(array, void *, elem) { utarray_push_back(result, elem); }
+
+    return result;
+}
+
 #endif

plugins/mqtt/mqtt_handle.c

@@ -140,6 +140,7 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     neu_req_read_group_t cmd  = { 0 };
     strcpy(cmd.driver, req->node);
     strcpy(cmd.group, req->group);
+    cmd.sync = req->sync;
     if (0 != neu_plugin_op(plugin, header, &cmd)) {
         plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
         return -1;

plugins/restful/adapter_handle.c

@@ -334,6 +334,7 @@ void handle_get_node_state_resp(nng_aio *aio, neu_resp_get_node_state_t *state)
     res.running   = state->state.running;
     res.rtt       = state->rtt;
     res.log_level = state->state.log_level;
+    strcpy(res.core_level, state->core_level);
 
     neu_json_encode_by_fn(&res, neu_json_encode_get_node_state_resp, &result);
 
@@ -361,6 +362,8 @@ void handle_get_nodes_state_resp(nng_aio *                   aio,
         states_res.nodes[index].log_level = state->state.log_level;
     }
 
+    strcpy(states_res.core_level, states->core_level);
+
     neu_json_encode_by_fn(&states_res, neu_json_encode_get_nodes_state_resp,
                           &result);
 

plugins/restful/handle.c

@@ -78,6 +78,9 @@ static struct neu_http_handler cors_handler[] = {
     {
         .url = "/api/v2/write/tags",
     },
+    {
+        .url = "/api/v2/write/gtags",
+    },
     {
         .url = "/api/v2/subscribe",
     },
@@ -290,6 +293,12 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/write/tags",
         .value.handler = handle_write_tags,
     },
+    {
+        .method        = NEU_HTTP_METHOD_POST,
+        .type          = NEU_HTTP_HANDLER_FUNCTION,
+        .url           = "/api/v2/write/gtags",
+        .value.handler = handle_write_gtags,
+    },
     {
         .method        = NEU_HTTP_METHOD_POST,
         .type          = NEU_HTTP_HANDLER_FUNCTION,

plugins/restful/log_handle.c

@@ -116,10 +116,7 @@ void handle_log_level(nng_aio *aio)
             int                        ret    = 0;
             neu_reqresp_head_t         header = { 0 };
             neu_req_update_log_level_t cmd    = { 0 };
-            zlog_category_t *          ct = zlog_get_category(req->node_name);
-            zlog_category_t *          neuron = zlog_get_category("neuron");
             int                        log_level;
-            static int                 log_level_neuron = 60;
 
             if (0 == strcmp(req->log_level, NEU_LOG_LEVEL_DEBUG)) {
                 log_level = ZLOG_LEVEL_DEBUG;
@@ -149,36 +146,39 @@ void handle_log_level(nng_aio *aio)
                 });
                 goto end;
             }
-
-            ret = zlog_level_switch(ct, log_level);
-            if (ret != 0) {
-                nlog_error(
-                    "Failed to modify log_level of the node, node_name:%s, "
-                    "ret: %d,",
-                    req->node_name, ret);
-                NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-                    neu_http_response(aio, error_code.error, result_error);
-                });
+            if (req->core) {
+                zlog_category_t *neuron = zlog_get_category("neuron");
+                ret                     = zlog_level_switch(neuron, log_level);
+                if (ret != 0) {
+                    nlog_error(
+                        "Failed to modify log_level of the core, ret: %d", ret);
+                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
+                        neu_http_response(aio, error_code.error, result_error);
+                    });
+                    goto end;
+                }
             }
 
-            if (log_level > log_level_neuron) {
-                log_level_neuron = log_level;
-            }
-
-            ret = zlog_level_switch(neuron, log_level_neuron);
-            if (ret != 0) {
-                nlog_error("Failed to modify log_level of the node, ret: %d",
-                           ret);
-                NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-                    neu_http_response(aio, error_code.error, result_error);
-                });
+            if (req->node_name) {
+                zlog_category_t *ct = zlog_get_category(req->node_name);
+                strcpy(cmd.node, req->node_name);
+                ret = zlog_level_switch(ct, log_level);
+                if (ret != 0) {
+                    nlog_error(
+                        "Failed to modify log_level of the node, node_name:%s, "
+                        "ret: %d,",
+                        req->node_name, ret);
+                    NEU_JSON_RESPONSE_ERROR(0, {
+                        neu_http_response(aio, error_code.error, result_error);
+                    });
+                    goto end;
+                }
             }
 
             header.ctx  = aio;
             header.type = NEU_REQ_UPDATE_LOG_LEVEL;
-            strcpy(cmd.node, req->node_name);
-
-            ret = neu_plugin_op(plugin, header, &cmd);
+            cmd.core    = req->core;
+            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/rw_handle.c

@@ -43,7 +43,8 @@ void handle_read(nng_aio *aio)
 
             strcpy(cmd.driver, req->node);
             strcpy(cmd.group, req->group);
-            ret = neu_plugin_op(plugin, header, &cmd);
+            cmd.sync = req->sync;
+            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);
@@ -203,6 +204,83 @@ void handle_write_tags(nng_aio *aio)
         })
 }
 
+static void trans(neu_json_write_gtags_req_t *req, neu_req_write_gtags_t *cmd)
+{
+    strcpy(cmd->driver, req->node);
+    cmd->n_group = req->n_group;
+    cmd->groups  = calloc(cmd->n_group, sizeof(neu_req_gtag_group_t));
+    for (int i = 0; i < cmd->n_group; i++) {
+        strcpy(cmd->groups[i].group, req->groups[i].group);
+        cmd->groups[i].n_tag = req->groups[i].n_tag;
+        cmd->groups[i].tags =
+            calloc(cmd->groups[i].n_tag, sizeof(neu_resp_tag_value_t));
+
+        for (int k = 0; k < cmd->groups[i].n_tag; k++) {
+            strcpy(cmd->groups[i].tags[k].tag, req->groups[i].tags[k].tag);
+
+            switch (req->groups[i].tags[k].t) {
+            case NEU_JSON_INT:
+                cmd->groups[i].tags[k].value.type = NEU_TYPE_INT64;
+                cmd->groups[i].tags[k].value.value.u64 =
+                    req->groups[i].tags[k].value.val_int;
+                break;
+            case NEU_JSON_STR:
+                cmd->groups[i].tags[k].value.type = NEU_TYPE_STRING;
+                strcpy(cmd->groups[i].tags[k].value.value.str,
+                       req->groups[i].tags[k].value.val_str);
+                break;
+            case NEU_JSON_DOUBLE:
+                cmd->groups[i].tags[k].value.type = NEU_TYPE_DOUBLE;
+                cmd->groups[i].tags[k].value.value.d64 =
+                    req->groups[i].tags[k].value.val_double;
+                break;
+            case NEU_JSON_BOOL:
+                cmd->groups[i].tags[k].value.type = NEU_TYPE_BOOL;
+                cmd->groups[i].tags[k].value.value.boolean =
+                    req->groups[i].tags[k].value.val_bool;
+                break;
+            case NEU_JSON_BYTES:
+                cmd->groups[i].tags[k].value.type = NEU_TYPE_BYTES;
+                cmd->groups[i].tags[k].value.value.bytes.length =
+                    req->groups[i].tags[k].value.val_bytes.length;
+                memcpy(cmd->groups[i].tags[k].value.value.bytes.bytes,
+                       req->groups[i].tags[k].value.val_bytes.bytes,
+                       req->groups[i].tags[k].value.val_bytes.length);
+                break;
+            default:
+                assert(false);
+                break;
+            }
+        }
+    }
+}
+
+void handle_write_gtags(nng_aio *aio)
+{
+    neu_plugin_t *plugin = neu_rest_get_plugin();
+
+    NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
+        aio, neu_json_write_gtags_req_t, neu_json_decode_write_gtags_req, {
+            neu_reqresp_head_t    header = { 0 };
+            neu_req_write_gtags_t cmd    = { 0 };
+
+            nng_http_req *nng_req = nng_aio_get_input(aio, 0);
+            nlog_notice("<%p> req %s %s", aio, nng_http_req_get_method(nng_req),
+                        nng_http_req_get_uri(nng_req));
+            header.ctx  = aio;
+            header.type = NEU_REQ_WRITE_GTAGS;
+
+            trans(req, &cmd);
+
+            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);
+                });
+            }
+        })
+}
+
 void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp)
 {
     neu_json_read_resp_t api_res = { 0 };

plugins/restful/rw_handle.h

@@ -29,6 +29,7 @@ void handle_rw_uninit();
 void handle_read(nng_aio *aio);
 void handle_write(nng_aio *aio);
 void handle_write_tags(nng_aio *aio);
+void handle_write_gtags(nng_aio *aio);
 void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp);
 
 #endif

src/adapter/adapter.c

@@ -416,6 +416,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
     switch (pheader->type) {
     case NEU_REQ_READ_GROUP:
     case NEU_REQ_WRITE_TAGS:
+    case NEU_REQ_WRITE_GTAGS:
     case NEU_REQ_WRITE_TAG: {
         neu_req_read_group_t *cmd = (neu_req_read_group_t *) data;
         strcpy(pheader->receiver, cmd->driver);
@@ -470,7 +471,9 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
     }
     case NEU_REQ_UPDATE_LOG_LEVEL: {
         neu_req_update_log_level_t *cmd = (neu_req_update_log_level_t *) data;
-        strcpy(pheader->receiver, cmd->node);
+        if (cmd->node != NULL) {
+            strcpy(pheader->receiver, cmd->node);
+        }
         break;
     }
     default:
@@ -705,6 +708,22 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         break;
     }
+    case NEU_REQ_WRITE_GTAGS: {
+        neu_resp_error_t error = { 0 };
+
+        if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
+            error.error  = NEU_ERR_GROUP_NOT_ALLOW;
+            header->type = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(adapter, header, &error);
+        } else {
+            void *msg_dump = calloc(header->len, 1);
+            memcpy(msg_dump, header, header->len);
+            neu_adapter_driver_write_gtags((neu_adapter_driver_t *) adapter,
+                                           msg_dump);
+        }
+        break;
+    }
     case NEU_REQ_NODE_SETTING: {
         neu_req_node_setting_t *cmd   = (neu_req_node_setting_t *) &header[1];
         neu_resp_error_t        error = { 0 };
@@ -741,17 +760,18 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         break;
     }
     case NEU_REQ_GET_NODE_STATE: {
-        neu_resp_get_node_state_t resp = { 0 };
+        neu_resp_get_node_state_t *resp =
+            (neu_resp_get_node_state_t *) &header[1];
 
         neu_metric_entry_t *e = NULL;
         if (NULL != adapter->metrics) {
             HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e);
         }
-        resp.rtt     = NULL != e ? e->value : 0;
-        resp.state   = neu_adapter_get_state(adapter);
+        resp->rtt    = NULL != e ? e->value : 0;
+        resp->state  = neu_adapter_get_state(adapter);
         header->type = NEU_RESP_GET_NODE_STATE;
         neu_msg_exchange(header);
-        reply(adapter, header, &resp);
+        reply(adapter, header, resp);
         break;
     }
     case NEU_REQ_GET_GROUP: {
@@ -1682,6 +1702,9 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data)
     case NEU_REQ_WRITE_TAGS:
         data_size = sizeof(neu_req_write_tags_t);
         break;
+    case NEU_REQ_WRITE_GTAGS:
+        data_size = sizeof(neu_req_write_gtags_t);
+        break;
     case NEU_RESP_READ_GROUP:
         data_size = sizeof(neu_resp_read_group_t);
         break;

src/adapter/driver/driver.c

@@ -405,6 +405,30 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
 
             utarray_push_back(resp.tags, &tag_value);
         }
+    } else if (cmd->sync) {
+        if (NULL == driver->adapter.module->intf_funs->driver.group_sync) {
+            // plugin does not support sync read
+            utarray_foreach(tags, neu_datatag_t *, tag)
+            {
+                neu_resp_tag_value_meta_t tag_value = { 0 };
+                strcpy(tag_value.tag, tag->name);
+                tag_value.value.type = NEU_TYPE_ERROR;
+                tag_value.value.value.i32 =
+                    NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC;
+
+                utarray_push_back(resp.tags, &tag_value);
+            }
+        } else {
+            // sync read to update cache
+            driver->adapter.module->intf_funs->driver.group_sync(
+                driver->adapter.plugin, &g->grp);
+            // fetch updated data from cache
+            read_group(global_timestamp,
+                       neu_group_get_interval(group) *
+                           NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
+                       neu_adapter_get_tag_cache_type(&driver->adapter),
+                       driver->cache, cmd->group, tags, resp.tags);
+        }
     } else {
         read_group(global_timestamp,
                    neu_group_get_interval(group) *
@@ -599,7 +623,7 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
 
             if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
                 if (cmd->tags[i].value.type == NEU_TYPE_INT64) {
-                    tv.value.d64 = (double) tv.value.u64;
+                    tv.value.d64 = (double) tv.value.i64;
                 }
             }
             if (tag->decimal != 0) {
@@ -639,6 +663,127 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     free_tags(cmd);
 }
 
+void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
+                                    neu_reqresp_head_t *  req)
+{
+    neu_req_write_gtags_t *cmd     = (neu_req_write_gtags_t *) &req[1];
+    group_t *              first_g = NULL;
+
+    if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) {
+        for (int i = 0; i < cmd->n_group; i++) {
+            free(cmd->groups[i].tags);
+        }
+        free(cmd->groups);
+        driver->adapter.cb_funs.driver.write_response(
+            &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING);
+        return;
+    }
+
+    if (driver->adapter.module->intf_funs->driver.write_tags == NULL) {
+        for (int i = 0; i < cmd->n_group; i++) {
+            free(cmd->groups[i].tags);
+        }
+        free(cmd->groups);
+        driver->adapter.cb_funs.driver.write_response(
+            &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS);
+        return;
+    }
+
+    for (int i = 0; i < cmd->n_group; i++) {
+        group_t *g = find_group(driver, cmd->groups[i].group);
+
+        if (g == NULL) {
+            neu_resp_error_t error = { .error = NEU_ERR_GROUP_NOT_EXIST };
+            req->type              = NEU_RESP_ERROR;
+            for (int x = 0; x < cmd->n_group; x++) {
+                free(cmd->groups[x].tags);
+            }
+            free(cmd->groups);
+            driver->adapter.cb_funs.response(&driver->adapter, req, &error);
+            free(req);
+            return;
+        }
+
+        if (first_g == NULL) {
+            first_g = g;
+        }
+    }
+
+    UT_array *tags = NULL;
+    UT_icd    icd  = { sizeof(neu_plugin_tag_value_t), NULL, NULL, NULL };
+    utarray_new(tags, &icd);
+
+    for (int i = 0; i < cmd->n_group; i++) {
+        group_t *g = find_group(driver, cmd->groups[i].group);
+
+        for (int k = 0; k < cmd->groups[i].n_tag; k++) {
+            neu_plugin_tag_value_t tv = { 0 };
+
+            neu_datatag_t *tag =
+                neu_group_find_tag(g->group, cmd->groups[i].tags[k].tag);
+
+            if (tag != NULL &&
+                neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) &&
+                neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) {
+                tv.tag = neu_tag_dup(tag);
+
+                if (tag->type == NEU_TYPE_FLOAT ||
+                    tag->type == NEU_TYPE_DOUBLE) {
+                    if (cmd->groups[i].tags[k].value.type == NEU_TYPE_INT64) {
+                        tv.value.d64 = (double) tv.value.u64;
+                    }
+                }
+                if (tag->decimal != 0) {
+                    cal_decimal(tag->type, cmd->groups[i].tags[k].value.type,
+                                &cmd->groups[i].tags[k].value.value,
+                                tag->decimal);
+                }
+                fix_value(tag, cmd->groups[i].tags[k].value.type,
+                          &cmd->groups[i].tags[k].value);
+
+                tv.value = cmd->groups[i].tags[k].value.value;
+                utarray_push_back(tags, &tv);
+            }
+            if (tag != NULL) {
+                neu_tag_free(tag);
+            }
+        }
+    }
+
+    uint32_t n_tag = 0;
+    for (int i = 0; i < cmd->n_group; i++) {
+        n_tag += cmd->groups[i].n_tag;
+    }
+
+    if (utarray_len(tags) != (unsigned int) n_tag) {
+        neu_resp_error_t error = { .error = NEU_ERR_TAG_NOT_EXIST };
+        req->type              = NEU_RESP_ERROR;
+        for (int i = 0; i < cmd->n_group; i++) {
+            free(cmd->groups[i].tags);
+        }
+        free(cmd->groups);
+        driver->adapter.cb_funs.response(&driver->adapter, req, &error);
+        free(req);
+        utarray_foreach(tags, neu_plugin_tag_value_t *, tv)
+        {
+            neu_tag_free(tv->tag);
+        }
+        utarray_free(tags);
+        return;
+    }
+
+    to_be_write_tag_t wtag = { 0 };
+    wtag.single            = false;
+    wtag.req               = (void *) req;
+    wtag.tvs               = tags;
+
+    store_write_tag(first_g, &wtag);
+    for (int i = 0; i < cmd->n_group; i++) {
+        free(cmd->groups[i].tags);
+    }
+    free(cmd->groups);
+}
+
 void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                                   neu_reqresp_head_t *  req)
 {
@@ -675,7 +820,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
         }
         if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
             if (cmd->value.type == NEU_TYPE_INT64) {
-                cmd->value.value.d64 = (double) cmd->value.value.u64;
+                cmd->value.value.d64 = (double) cmd->value.value.i64;
             }
         }
 
@@ -1061,8 +1206,8 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
 
     HASH_FIND_STR(driver->groups, group, find);
     if (find == NULL) {
-        neu_adapter_driver_add_group(driver, group, 3000);
-        adapter_storage_add_group(driver->adapter.name, group, 3000);
+        neu_adapter_driver_add_group(driver, group, 100);
+        adapter_storage_add_group(driver->adapter.name, group, 100);
     }
     HASH_FIND_STR(driver->groups, group, find);
     assert(find != NULL);

src/adapter/driver/driver_internal.h

@@ -40,6 +40,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                                   neu_reqresp_head_t *  req);
 void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
                                    neu_reqresp_head_t *  req);
+void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
+                                    neu_reqresp_head_t *  req);
 
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval);

src/connection/connection.c

@@ -348,14 +348,19 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                 }
             } else {
                 if (rc == -1 && errno == EAGAIN) {
-                    if (retry > 5) {
+                    if (retry > 50) {
                         zlog_error(conn->param.log,
                                    "conn fd: %d, send buf len: %zd, ret: %zd, "
                                    "errno: %s(%d)",
                                    conn->fd, len, ret, strerror(errno), errno);
                         break;
                     } else {
-                        sleep(1);
+                        struct timespec t1 = {
+                            .tv_sec  = 0,
+                            .tv_nsec = 1000 * 1000 * 50,
+                        };
+                        struct timespec t2 = { 0 };
+                        nanosleep(&t1, &t2);
                         retry++;
                         zlog_warn(conn->param.log,
                                   "not all data send, retry: %d, ret: "
@@ -442,11 +447,21 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                 conn->fd, len, ret, strerror(errno), errno);
             conn_disconnect(conn);
         }
+    } else if (conn->param.type == NEU_CONN_UDP ||
+               conn->param.type == NEU_CONN_UDP_TO) {
+        if (ret <= 0) {
+            zlog_error(
+                conn->param.log,
+                "udp conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                conn->fd, len, ret, strerror(errno), errno);
+            conn_disconnect(conn);
+        }
     } else {
         if (ret <= 0) {
-            zlog_error(conn->param.log,
-                       "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
-                       conn->fd, len, ret, strerror(errno), errno);
+            zlog_error(
+                conn->param.log,
+                "tcp conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                conn->fd, len, ret, strerror(errno), errno);
             if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
                 conn_disconnect(conn);
             }

src/core/manager.c

@@ -155,7 +155,6 @@ void neu_manager_destroy(neu_manager_t *manager)
 
     UT_array *addrs = neu_node_manager_get_addrs_all(manager->node_manager);
 
-    neu_event_del_timer(manager->events, manager->timer_lev);
     neu_event_del_timer(manager->events, manager->timer_timestamp);
 
     header->type = NEU_REQ_NODE_UNINIT;
@@ -788,6 +787,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         neu_resp_get_nodes_state_t resp = { 0 };
         UT_array *states = neu_node_manager_get_state(manager->node_manager);
 
+        neu_adapter_t *adapter_core =
+            neu_node_manager_find(manager->node_manager, "default-dashboard");
+        strcpy(resp.core_level,
+               neu_plugin_to_plugin_common(adapter_core->plugin)->log_level);
         resp.states = utarray_clone(states);
 
         strcpy(header->receiver, header->sender);
@@ -813,10 +816,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
     case NEU_REQ_GET_GROUP:
     case NEU_REQ_GET_NODE_SETTING:
-    case NEU_REQ_GET_NODE_STATE:
     case NEU_REQ_READ_GROUP:
     case NEU_REQ_WRITE_TAG:
     case NEU_REQ_WRITE_TAGS:
+    case NEU_REQ_WRITE_GTAGS:
     case NEU_REQ_GET_TAG:
     case NEU_REQ_GET_NDRIVER_TAGS:
     case NEU_REQ_NODE_CTL:
@@ -834,6 +837,27 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         break;
     }
 
+    case NEU_REQ_GET_NODE_STATE: {
+        if (neu_node_manager_find(manager->node_manager, header->receiver) ==
+            NULL) {
+            neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
+            header->type       = NEU_RESP_ERROR;
+            neu_msg_exchange(header);
+            reply(manager, header, &e);
+        } else {
+            neu_resp_get_node_state_t *resp =
+                (neu_resp_get_node_state_t *) &header[1];
+            neu_adapter_t *adapter_core = neu_node_manager_find(
+                manager->node_manager, "default-dashboard");
+            strcpy(
+                resp->core_level,
+                neu_plugin_to_plugin_common(adapter_core->plugin)->log_level);
+            forward_msg(manager, header, header->receiver);
+        }
+
+        break;
+    }
+
     case NEU_RESP_NODE_RENAME: {
         neu_resp_node_rename_t *resp = (neu_resp_node_rename_t *) &header[1];
         if (0 == resp->error) {
@@ -1088,25 +1112,68 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         break;
     }
-    case NEU_REQ_UPDATE_LOG_LEVEL:
-        if (neu_node_manager_find(manager->node_manager, header->receiver) ==
-            NULL) {
-            neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
-            header->type       = NEU_RESP_ERROR;
-            neu_msg_exchange(header);
-            reply(manager, header, &e);
-        } else {
-            neu_req_update_log_level_t *cmd =
-                (neu_req_update_log_level_t *) &header[1];
-            neu_adapter_t *adapter_log =
-                neu_node_manager_find(manager->node_manager, header->receiver);
-            strcpy(neu_plugin_to_plugin_common(adapter_log->plugin)->log_level,
-                   cmd->log_level);
+    case NEU_REQ_UPDATE_LOG_LEVEL: {
+        neu_req_update_log_level_t *cmd =
+            (neu_req_update_log_level_t *) &header[1];
 
-            forward_msg(manager, header, header->receiver);
+        if (cmd->core) {
+            if (strcmp(header->receiver, "") == 0) {
+                neu_adapter_t *adapter_core = neu_node_manager_find(
+                    manager->node_manager, "default-dashboard");
+                strcpy(neu_plugin_to_plugin_common(adapter_core->plugin)
+                           ->log_level,
+                       cmd->log_level);
+                neu_resp_error_t error = { 0 };
+                neu_msg_exchange(header);
+                header->type = NEU_RESP_ERROR;
+                reply(manager, header, &error);
+            } else {
+                if (neu_node_manager_find(manager->node_manager,
+                                          header->receiver) == NULL) {
+                    neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
+                    header->type       = NEU_RESP_ERROR;
+                    neu_msg_exchange(header);
+                    reply(manager, header, &e);
+                } else {
+                    neu_adapter_t *adapter_core = neu_node_manager_find(
+                        manager->node_manager, "default-dashboard");
+                    strcpy(neu_plugin_to_plugin_common(adapter_core->plugin)
+                               ->log_level,
+                           cmd->log_level);
+                    neu_adapter_t *adapter_log = neu_node_manager_find(
+                        manager->node_manager, header->receiver);
+                    strcpy(neu_plugin_to_plugin_common(adapter_log->plugin)
+                               ->log_level,
+                           cmd->log_level);
+                    forward_msg(manager, header, header->receiver);
+                }
+            }
+        } else {
+            if (strcmp(header->receiver, "") == 0) {
+                neu_resp_error_t e = { .error = NEU_ERR_PARAM_IS_WRONG };
+                header->type       = NEU_RESP_ERROR;
+                neu_msg_exchange(header);
+                reply(manager, header, &e);
+            } else {
+                if (neu_node_manager_find(manager->node_manager,
+                                          header->receiver) == NULL) {
+                    neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
+                    header->type       = NEU_RESP_ERROR;
+                    neu_msg_exchange(header);
+                    reply(manager, header, &e);
+                } else {
+                    neu_adapter_t *adapter_log = neu_node_manager_find(
+                        manager->node_manager, header->receiver);
+                    strcpy(neu_plugin_to_plugin_common(adapter_log->plugin)
+                               ->log_level,
+                           cmd->log_level);
+                    forward_msg(manager, header, header->receiver);
+                }
+            }
         }
 
         break;
+    }
     default:
         assert(false);
         break;

src/core/manager_internal.h

@@ -42,8 +42,7 @@ typedef struct neu_manager {
 
     neu_event_timer_t *timer_timestamp;
 
-    neu_event_timer_t *timer_lev;
-    int64_t            timestamp_lev_manager;
+    int64_t timestamp_lev_manager;
 } neu_manager_t;
 
 int       neu_manager_add_plugin(neu_manager_t *manager, const char *library);

src/event/event_linux.c

@@ -73,14 +73,14 @@ struct neu_events {
 
     pthread_mutex_t   mtx;
     int               n_event;
-    struct event_data event_datas[256];
+    struct event_data event_datas[512];
 };
 
 static int get_free_event(neu_events_t *events)
 {
     int ret = -1;
     pthread_mutex_lock(&events->mtx);
-    for (int i = 0; i < 256; i++) {
+    for (int i = 0; i < 512; i++) {
         if (events->event_datas[i].use == false) {
             events->event_datas[i].use   = true;
             events->event_datas[i].index = i;
@@ -213,6 +213,9 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
         .it_interval.tv_nsec = timer.millisecond * 1000 * 1000,
     };
     int index = get_free_event(events);
+    if (index < 0) {
+        zlog_notice(neuron, "no free event: %d", events->epoll_fd);
+    }
     assert(index >= 0);
 
     neu_event_timer_t *timer_ctx = &events->event_datas[index].ctx.timer;

src/parser/neu_json_log.c

@@ -42,14 +42,24 @@ int neu_json_decode_update_log_level_req(
         return -1;
     }
 
-    neu_json_elem_t req_elems[] = { {
-                                        .name = "node",
-                                        .t    = NEU_JSON_STR,
-                                    },
-                                    {
-                                        .name = "level",
-                                        .t    = NEU_JSON_STR,
-                                    } };
+    req->node_name = NULL;
+    req->core      = true;
+
+    neu_json_elem_t req_elems[] = {
+        {
+            .name      = "node",
+            .t         = NEU_JSON_STR,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+        },
+        {
+            .name = "level",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name      = "core",
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+        }
+    };
 
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
@@ -57,8 +67,13 @@ int neu_json_decode_update_log_level_req(
         goto decode_fail;
     }
 
-    req->node_name = req_elems[0].v.val_str;
+    if (req_elems[0].v.val_str != NULL) {
+        req->node_name = req_elems[0].v.val_str;
+    }
     req->log_level = req_elems[1].v.val_str;
+    if (req_elems[2].t == NEU_JSON_BOOL) {
+        req->core = req_elems[2].v.val_bool;
+    }
 
     *result = req;
     goto decode_exit;

src/parser/neu_json_log.h

@@ -36,6 +36,7 @@ typedef char *neu_json_get_log_resp_row_t;
 typedef struct {
     char *node_name;
     char *log_level;
+    bool  core;
 } neu_json_update_log_level_req_t;
 
 typedef struct {

src/parser/neu_json_node.c

@@ -189,6 +189,11 @@ int neu_json_encode_get_node_state_resp(void *json_object, void *param)
                                          .name      = "log_level",
                                          .t         = NEU_JSON_STR,
                                          .v.val_str = resp->log_level,
+                                     },
+                                     {
+                                         .name      = "neuron_core",
+                                         .t         = NEU_JSON_STR,
+                                         .v.val_str = resp->core_level,
                                      } };
     ret = neu_json_encode_field(json_object, resp_elems,
                                 NEU_JSON_ELEM_SIZE(resp_elems));
@@ -236,10 +241,15 @@ int neu_json_encode_get_nodes_state_resp(void *json_object, void *param)
     }
 
     neu_json_elem_t resp_elems[] = { {
-        .name         = "states",
-        .t            = NEU_JSON_OBJECT,
-        .v.val_object = node_array,
-    } };
+                                         .name         = "states",
+                                         .t            = NEU_JSON_OBJECT,
+                                         .v.val_object = node_array,
+                                     },
+                                     {
+                                         .name      = "neuron_core",
+                                         .t         = NEU_JSON_STR,
+                                         .v.val_str = resp->core_level,
+                                     } };
     ret = neu_json_encode_field(json_object, resp_elems,
                                 NEU_JSON_ELEM_SIZE(resp_elems));
 

src/parser/neu_json_node.h

@@ -54,6 +54,7 @@ typedef struct {
     int64_t link;
     int64_t rtt;
     char *  log_level;
+    char    core_level[NEU_LOG_LEVEL_LEN];
 } neu_json_get_node_state_resp_t;
 
 int neu_json_encode_get_node_state_resp(void *json_object, void *param);
@@ -69,6 +70,7 @@ typedef struct {
 typedef struct {
     uint16_t                    n_node;
     neu_json_get_nodes_state_t *nodes;
+    char                        core_level[NEU_LOG_LEVEL_LEN];
 } neu_json_get_nodes_state_resp_t;
 
 int neu_json_encode_get_nodes_state_resp(void *json_object, void *param);

src/parser/neu_json_rw.c

@@ -413,14 +413,21 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
 
     json_obj = neu_json_decode_new(buf);
 
-    neu_json_elem_t req_elems[] = { {
-                                        .name = "node",
-                                        .t    = NEU_JSON_STR,
-                                    },
-                                    {
-                                        .name = "group",
-                                        .t    = NEU_JSON_STR,
-                                    } };
+    neu_json_elem_t req_elems[] = {
+        {
+            .name = "node",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name = "group",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name      = "sync",
+            .t         = NEU_JSON_BOOL,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+        },
+    };
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
     if (ret != 0) {
@@ -429,6 +436,7 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
 
     req->node  = req_elems[0].v.val_str;
     req->group = req_elems[1].v.val_str;
+    req->sync  = req_elems[2].v.val_bool;
 
     *result = req;
     goto decode_exit;
@@ -548,4 +556,149 @@ void neu_json_metas_to_json(neu_tag_meta_t *metas, int n_meta,
             break;
         }
     }
-}
+}
+
+int neu_json_decode_write_gtags_req(char *                       buf,
+                                    neu_json_write_gtags_req_t **result)
+{
+    void *json_obj = neu_json_decode_new(buf);
+    if (NULL == json_obj) {
+        return -1;
+    }
+
+    int ret = neu_json_decode_write_gtags_req_json(json_obj, result);
+    neu_json_decode_free(json_obj);
+    return ret;
+}
+
+#include "utils/log.h"
+
+static int decode_write_gtags_req_json(void *                      json_obj,
+                                       neu_json_write_gtags_req_t *req)
+{
+    int ret = 0;
+
+    neu_json_elem_t req_elems[] = {
+        {
+            .name = "node",
+            .t    = NEU_JSON_STR,
+        },
+        {
+            .name = "groups",
+            .t    = NEU_JSON_OBJECT,
+        },
+    };
+    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
+                                  req_elems);
+    if (ret != 0) {
+        return -1;
+    }
+
+    req->node = req_elems[0].v.val_str;
+
+    req->n_group = neu_json_decode_array_size_by_json(json_obj, "groups");
+    if (req->n_group <= 0) {
+        return -1;
+    }
+
+    req->groups = calloc(req->n_group, sizeof(neu_json_write_gtags_elem_t));
+    for (int i = 0; i < req->n_group; i++) {
+        neu_json_elem_t g_elems[] = {
+            {
+                .name = "group",
+                .t    = NEU_JSON_STR,
+            },
+            {
+                .name = "tags",
+                .t    = NEU_JSON_OBJECT,
+            },
+        };
+
+        ret = neu_json_decode_array_by_json(
+            json_obj, "groups", i, NEU_JSON_ELEM_SIZE(g_elems), g_elems);
+
+        req->groups[i].group = g_elems[0].v.val_str;
+        req->groups[i].n_tag = json_array_size(g_elems[1].v.val_object);
+        if (req->groups[i].n_tag <= 0) {
+            continue;
+        }
+
+        req->groups[i].tags =
+            calloc(req->groups[i].n_tag, sizeof(neu_json_write_tags_elem_t));
+
+        for (int k = 0; k < req->groups[i].n_tag; k++) {
+            neu_json_elem_t v_elems[] = {
+                {
+                    .name = "tag",
+                    .t    = NEU_JSON_STR,
+                },
+                {
+                    .name = "value",
+                    .t    = NEU_JSON_VALUE,
+                },
+            };
+
+            ret = neu_json_decode_array_elem(g_elems[1].v.val_object, k,
+                                             NEU_JSON_ELEM_SIZE(v_elems),
+                                             v_elems);
+            if (ret != 0) {
+                for (int x = i; x >= 0; x--) {
+                    free(req->groups[x].tags);
+                }
+                free(req->groups);
+                return -1;
+            }
+            req->groups[i].tags[k].tag   = v_elems[0].v.val_str;
+            req->groups[i].tags[k].t     = v_elems[1].t;
+            req->groups[i].tags[k].value = v_elems[1].v;
+        }
+    }
+
+    return 0;
+}
+
+int neu_json_decode_write_gtags_req_json(void *                       json_obj,
+                                         neu_json_write_gtags_req_t **result)
+{
+    neu_json_write_gtags_req_t *req =
+        calloc(1, sizeof(neu_json_write_gtags_req_t));
+    if (req == NULL) {
+        return -1;
+    }
+
+    int ret = decode_write_gtags_req_json(json_obj, req);
+    if (0 == ret) {
+        *result = req;
+    } else {
+        free(req);
+    }
+
+    return ret;
+}
+
+void neu_json_decode_write_gtags_req_free(neu_json_write_gtags_req_t *req)
+{
+    free(req->node);
+
+    for (int i = 0; i < req->n_group; i++) {
+        free(req->groups[i].group);
+
+        for (int k = 0; k < req->groups[i].n_tag; k++) {
+            free(req->groups[i].tags[k].tag);
+
+            if (req->groups[i].tags[k].t == NEU_JSON_STR) {
+                free(req->groups[i].tags[k].value.val_str);
+            }
+            if (req->groups[i].tags[k].t == NEU_JSON_BYTES &&
+                req->groups[i].tags[k].value.val_bytes.length > 0) {
+                free(req->groups[i].tags[k].value.val_bytes.bytes);
+            }
+        }
+        if (req->groups[i].n_tag > 0) {
+            free(req->groups[i].tags);
+        }
+    }
+
+    free(req->groups);
+    free(req);
+}

src/utils/json.c

@@ -272,6 +272,25 @@ int neu_json_decode_array(char *buf, char *name, int index, int size,
     return 0;
 }
 
+int neu_json_decode_array_elem(void *json, int index, int size,
+                               neu_json_elem_t *ele)
+{
+    json_t *child = NULL;
+
+    child = json_array_get(json, index);
+    if (child == NULL) {
+        return -1;
+    }
+
+    for (int i = 0; i < size; i++) {
+        if (decode_object(child, &ele[i]) == -1) {
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
 int neu_json_decode_array_by_json(void *json, char *name, int index, int size,
                                   neu_json_elem_t *ele)
 {

version

@@ -1 +1 @@
-2.6.1
+2.6.2