Merge pull request #2442 from fengzeroz/v2.11

version: 2.11.2

.github/workflows/check.yml

@@ -41,7 +41,7 @@ jobs:
 
   ut:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.11
 
     steps:
       - uses: actions/checkout@v4
@@ -78,7 +78,7 @@ jobs:
 
   pft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.11
     strategy:
       matrix:
         plugin: [core, ekuiper, modbus, mqtt, azure, metrics]

.github/workflows/ci.yml

@@ -8,20 +8,20 @@ on:
       - published
 
 env:
-  VERSION: 2.11.0
+  VERSION: 2.11.2
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.11
 
 jobs:
   cross-build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     strategy:
       matrix:
         arch:
           - [aarch64-linux-gnu, aarch64, arm64]
           - [arm-linux-gnueabihf, armv4, armhf]
           - [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.11
 
     steps:
       - uses: actions/checkout@v4
@@ -52,19 +52,19 @@ jobs:
         run: |
           ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-$VERSION
 
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
-          name: neuron-sdk
+          name: neuron-sdk-${{ matrix.arch[2] }}
           path: |
             neuron-sdk-*.tar.gz
 
   build-debug-sdk:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     strategy:
       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.11
 
     steps:
       - uses: actions/checkout@v4
@@ -83,9 +83,9 @@ jobs:
         run: |
           ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-debug-$VERSION
 
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
-          name: neuron-sdk
+          name: neuron-sdk-debug
           path: |
             neuron-sdk-debug-*.tar.gz
   
@@ -102,9 +102,19 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
-      - uses: actions/download-artifact@v3
+      - uses: actions/download-artifact@v4
         with:
-          name: neuron-sdk
+          name: neuron-sdk-debug
+
+      - uses: actions/download-artifact@v4
+        with:
+          name: neuron-sdk-arm64
+      - uses: actions/download-artifact@v4
+        with:
+          name: neuron-sdk-armhf
+      - uses: actions/download-artifact@v4
+        with:
+          name: neuron-sdk-amd64
       
       - uses: softprops/action-gh-release@v1
         env:

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.11/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.11/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.11/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.11/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 11
-#define NEU_VERSION_FIX 0
+#define NEU_VERSION_FIX 2
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
@@ -161,4 +161,8 @@ typedef enum {
 // forward declaration for neu_adapter_t
 typedef struct neu_adapter neu_adapter_t;
 
+typedef struct neu_manager neu_manager_t;
+
+extern neu_manager_t *g_manager;
+
 #endif

include/neuron/json/json.h

@@ -185,6 +185,7 @@ int   neu_json_decode_value(void *object, neu_json_elem_t *ele);
 void *neu_json_encode_new();
 void  neu_json_encode_free(void *json_object);
 void *neu_json_encode_array(void *array, neu_json_elem_t *t, int n);
+void *neu_json_encode_array_ecp(void *array, neu_json_elem_t *t, int n);
 void *neu_json_encode_array_value(void *array, neu_json_elem_t *t, int n);
 int   neu_json_encode_field(void *json_object, neu_json_elem_t *elem, int n);
 int   neu_json_encode(void *json_object, char **str);

include/neuron/otel/otel_manager.h

@@ -47,6 +47,8 @@ neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
 
 void neu_otel_free_trace(neu_otel_trace_ctx ctx);
 
+void neu_otel_set_internal_parent_span(neu_otel_trace_ctx ctx);
+
 neu_otel_scope_ctx neu_otel_add_span(neu_otel_trace_ctx ctx);
 neu_otel_scope_ctx neu_otel_add_span2(neu_otel_trace_ctx ctx,
                                       const char *       span_name,

include/neuron/type.h

@@ -155,7 +155,8 @@ typedef struct {
     uint8_t *  ptr;
 } neu_value_ptr_t;
 
-#define NEU_VALUE_SIZE 128
+#define NEU_VALUE_SIZE 256
+#define NEU_ARRAY_SIZE 128
 
 typedef struct {
     uint8_t bytes[NEU_VALUE_SIZE];

persistence/0110_2.11.0_tag.sql

@@ -16,21 +16,9 @@
 * along with this program; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 **/
-
 BEGIN TRANSACTION;
 
-ALTER TABLE groups RENAME TO temp_groups;
+alter TABLE groups add column context TEXT NULL check(length(context) <= 512);
+alter TABLE tags add column format TEXT NULL;
 
-CREATE TABLE IF NOT EXISTS
-  groups (
-    driver_name TEXT NOT NULL,
-    name TEXT NULL check(length(name) <= 32),
-    interval INTEGER NOT NULL check(interval >= 100),
-	context TEXT NULL check(length(context) <= 512),
-    UNIQUE (driver_name, name),
-    FOREIGN KEY (driver_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE
-  );
-INSERT INTO groups SELECT driver_name, name, interval, NULL FROM temp_groups;
-
-DROP TABLE temp_groups;
-COMMIT;
+COMMIT;

persistence/0111_2.11.0_sub.sql

@@ -0,0 +1,22 @@
+BEGIN TRANSACTION;
+
+DROP TABLE IF EXISTS temp_subscriptions;
+ALTER TABLE subscriptions RENAME TO temp_subscriptions;
+
+CREATE TABLE
+  IF NOT EXISTS subscriptions (
+    app_name TEXT NOT NULL,
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    params TEXT DEFAULT NULL,
+    static_tags TEXT DEFAULT NULL,
+    CHECK (app_name != driver_name),
+    UNIQUE (app_name, driver_name, group_name),
+    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO subscriptions SELECT app_name, driver_name, group_name, params, NULL FROM temp_subscriptions;
+
+DROP TABLE temp_subscriptions;
+
+COMMIT;

persistence/0112_2.11.0_tag.sql

@@ -1,76 +0,0 @@
-/**
-* NEURON IIoT System for Industry 4.0
-* Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
-*
-* This program is free software; you can redistribute it and/or
-* modify it under the terms of the GNU Lesser General Public
-* License as published by the Free Software Foundation; either
-* version 3 of the License, or (at your option) any later version.
-*
-* This program is distributed in the hope that it will be useful,
-* but WITHOUT ANY WARRANTY; without even the implied warranty of
-* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-* Lesser General Public License for more details.
-*
-* You should have received a copy of the GNU Lesser General Public License
-* along with this program; if not, write to the Free Software Foundation,
-* Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
-**/
-BEGIN TRANSACTION;
-
-ALTER TABLE tags RENAME TO temp_tags;
-
-CREATE TABLE
-  IF NOT EXISTS tags (
-    driver_name TEXT NOT NULL,
-    group_name TEXT NOT NULL,
-    name TEXT NULL check (length (name) <= 128),
-    address TEXT NULL check (length (address) <= 128),
-    attribute INTEGER NOT NULL check (attribute BETWEEN 0 AND 15),
-    precision INTEGER NOT NULL check (precision BETWEEN 0 AND 17),
-    decimal REAL NOT NULL,
-    bias REAL NOT NULL check (bias BETWEEN -1000 AND 1000),
-    type INTEGER NOT NULL check (type BETWEEN 0 AND 40),
-    description TEXT NULL check (length (description) <= 512),
-    value TEXT,
-    format TEXT NULL,
-    UNIQUE (driver_name, group_name, name),
-    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
-  );
-
-INSERT INTO tags 
-SELECT
-  driver_name,
-  group_name,
-  name,
-  address,
-  attribute,
-  precision,
-  decimal,
-  bias,
-  type,
-  description,
-  NULL,
-  ""
-FROM temp_tags;
-
-DROP TABLE temp_tags;
-
-ALTER TABLE subscriptions RENAME TO temp_subscriptions;
-CREATE TABLE
-  IF NOT EXISTS subscriptions (
-    app_name TEXT NOT NULL,
-    driver_name TEXT NOT NULL,
-    group_name TEXT NOT NULL,
-    params TEXT DEFAULT NULL,
-    static_tags TEXT DEFAULT NULL,
-    CHECK (app_name != driver_name),
-    UNIQUE (app_name, driver_name, group_name),
-    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
-    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
-  );
-INSERT INTO subscriptions SELECT app_name, driver_name, group_name, params, NULL FROM temp_subscriptions;
-
-DROP TABLE temp_subscriptions;
-
-COMMIT;

persistence/0112_2.12.0_array.sql

@@ -20,6 +20,7 @@ BEGIN TRANSACTION;
 
 ALTER TABLE tags RENAME TO temp_tags;
 
+-- update type integer
 CREATE TABLE
   IF NOT EXISTS tags (
     driver_name TEXT NOT NULL,
@@ -30,31 +31,32 @@ CREATE TABLE
     precision INTEGER NOT NULL check (precision BETWEEN 0 AND 17),
     decimal REAL NOT NULL,
     bias REAL NOT NULL check (bias BETWEEN -1000 AND 1000),
-    type INTEGER NOT NULL check (type BETWEEN 0 AND 40),
+    type INTEGER NOT NULL check (type BETWEEN 0 AND 34 OR type = 40),
     description TEXT NULL check (length (description) <= 512),
     value TEXT,
+    format TEXT,
     UNIQUE (driver_name, group_name, name),
     FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
   );
 
-INSERT INTO tags SELECT * FROM temp_tags;
+INSERT INTO
+  tags
+SELECT
+  driver_name,
+  group_name,
+  name,
+  address,
+  attribute,
+  PRECISION,
+  decimal,
+  bias,
+  TYPE,
+  description,
+  NULL,
+  format
+FROM
+  temp_tags;
 
 DROP TABLE temp_tags;
 
-ALTER TABLE subscriptions RENAME TO temp_subscriptions;
-CREATE TABLE
-  IF NOT EXISTS subscriptions (
-    app_name TEXT NOT NULL,
-    driver_name TEXT NOT NULL,
-    group_name TEXT NOT NULL,
-    params TEXT DEFAULT NULL,
-    CHECK (app_name != driver_name),
-    UNIQUE (app_name, driver_name, group_name),
-    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
-    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
-  );
-INSERT INTO subscriptions SELECT * FROM temp_subscriptions;
-
-DROP TABLE temp_subscriptions;
-
-COMMIT;
+COMMIT;

plugins/ekuiper/json_rw.c

@@ -195,10 +195,21 @@ static int decode_write_req_json(void *json_obj, neu_json_write_req_t *req)
             .name = "value",
             .t    = NEU_JSON_VALUE,
         },
+        {
+            .name      = "precision",
+            .t         = NEU_JSON_INT,
+            .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+        },
     };
-    ret       = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
+    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
-    req->node = req_elems[0].v.val_str;
+
+    if (req_elems[4].v.val_int > 0) {
+        req_elems[3].t            = NEU_JSON_DOUBLE;
+        req_elems[3].v.val_double = (double_t) req_elems[3].v.val_int;
+    }
+
+    req->node  = req_elems[0].v.val_str;
     req->group = req_elems[1].v.val_str;
     req->tag   = req_elems[2].v.val_str;
     req->t     = req_elems[3].t;
@@ -259,10 +270,21 @@ static int decode_write_tags_req_json(void *                     json_obj,
                 .name = "value",
                 .t    = NEU_JSON_VALUE,
             },
+            {
+                .name      = "precision",
+                .t         = NEU_JSON_INT,
+                .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
+            },
         };
 
         ret = neu_json_decode_array_by_json(
             json_obj, "tags", i, NEU_JSON_ELEM_SIZE(v_elems), v_elems);
+
+        if (v_elems[2].v.val_int > 0) {
+            v_elems[1].t            = NEU_JSON_DOUBLE;
+            v_elems[1].v.val_double = (double_t) v_elems[1].v.val_int;
+        }
+
         req->tags[i].tag   = v_elems[0].v.val_str;
         req->tags[i].t     = v_elems[1].t;
         req->tags[i].value = v_elems[1].v;

plugins/ekuiper/plugin_ekuiper.c

@@ -316,7 +316,7 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/ekuiper/read_write.c

@@ -120,7 +120,8 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
         nng_msg *msg              = NULL;
         size_t   json_len         = strlen(json_str);
         size_t   trace_header_len = 0;
-        if (neu_otel_data_is_started() && trans_data->trace_ctx) {
+        if (neu_otel_data_is_started() && trans_data->trace_ctx &&
+            trans_trace) {
             trace_header_len = 26;
         }
 

plugins/mqtt/mqtt.json

@@ -56,8 +56,8 @@
 	"format": {
 		"name": "Upload Format",
 		"name_zh": "上报数据格式",
-		"description": "JSON format of the data reported. In Values-format mode, data are split into `values` and `errors` sub objects. In Tags-format mode, tag data are put in a single array. ECP-format is the format for connecting to ECP data storage. Custom format supports user-defined data format. For variable definition specifications, please refer to [https://docs.emqx.com/en/neuronex/latest/configuration/north-apps/mqtt/api.html].",
-		"description_zh": "上报数据的 JSON 格式。在 Values-format 格式下，数据被分为 `values` 和 `errors` 两个子对象。在 Tags-format 格式下，数据被放在一个数组中。 ECP-format 为对接 ECP 数据存储的格式。自定义格式，支持用户自定义上报数据格式，变量定义规范请参考【https://docs.emqx.com/zh/neuronex/latest/configuration/north-apps/mqtt/api.html】。",
+		"description": "JSON format of the data reported. In Values-format mode, data are split into `values` and `errors` sub objects. In Tags-format mode, tag data are put in a single array. ECP-format is the format for connecting to ECP data storage. Custom format supports user-defined data format. For variable definition specifications, please refer to 'https://docs.emqx.com/en/neuronex/latest/configuration/north-apps/mqtt/api.html'.",
+		"description_zh": "上报数据的 JSON 格式。在 Values-format 格式下，数据被分为 `values` 和 `errors` 两个子对象。在 Tags-format 格式下，数据被放在一个数组中。 ECP-format 为对接 ECP 数据存储的格式。自定义格式，支持用户自定义上报数据格式，变量定义规范请参考'https://docs.emqx.com/zh/neuronex/latest/configuration/north-apps/mqtt/api.html'。",
 		"attribute": "required",
 		"type": "map",
 		"default": 0,
@@ -129,30 +129,6 @@
 			"length": 255
 		}
 	},
-	"driver-action-req-topic": {
-		"name": "Driver Action Request Topic",
-		"name_zh": "驱动执行动作请求主题",
-		"description": "MQTT topic to receive driver action request messages.",
-		"description_zh": "驱动执行动作请求的 MQTT 主题。",
-		"attribute": "required",
-		"type": "string",
-		"default": "/neuron/${random_str}/action/req",
-		"valid": {
-			"length": 255
-		}
-	},
-	"driver-action-resp-topic": {
-		"name": "Driver Action Response Topic",
-		"name_zh": "驱动执行动作响应主题",
-		"description": "MQTT topic to send driver action response messages.",
-		"description_zh": "驱动执行动作响应的 MQTT 主题。",
-		"attribute": "required",
-		"type": "string",
-		"default": "/neuron/${random_str}/action/resp",
-		"valid": {
-			"length": 255
-		}
-	},
 	"upload_drv_state": {
 		"name": "Driver Status Report",
 		"name_zh": "驱动状态上报",

plugins/mqtt/mqtt_plugin_intf.c

@@ -474,7 +474,7 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     neu_otel_trace_ctx trace           = NULL;
     neu_otel_scope_ctx scope           = NULL;
     char               new_span_id[36] = { 0 };
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && head->ctx) {
         trace = neu_otel_find_trace(head->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/mqtt/schema.c

@@ -2,6 +2,80 @@
 
 #include "schema.h"
 
+static int mqtt_schema_parse(json_t *root, mqtt_schema_vt_t **vts,
+                             size_t *vts_len, int deep)
+{
+    *vts_len = 0;
+    *vts     = NULL;
+
+    const char *key   = NULL;
+    json_t *    value = NULL;
+
+    json_object_foreach(root, key, value)
+    {
+        if (json_is_string(value)) {
+            const char *str_val = json_string_value(value);
+            if (str_val == NULL) {
+                return -1;
+            }
+
+            *vts_len += 1;
+            *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t));
+            mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1];
+
+            memset(vt, 0, sizeof(mqtt_schema_vt_t));
+            strcpy(vt->name, key);
+
+            if (strcmp(str_val, "${timestamp}") == 0) {
+                vt->vt = MQTT_SCHEMA_TIMESTAMP;
+            } else if (strcmp(str_val, "${node}") == 0) {
+                vt->vt = MQTT_SCHEMA_NODE_NAME;
+            } else if (strcmp(str_val, "${group}") == 0) {
+                vt->vt = MQTT_SCHEMA_GROUP_NAME;
+            } else if (strcmp(str_val, "${tags}") == 0) {
+                vt->vt = MQTT_SCHEMA_TAGS;
+            } else if (strcmp(str_val, "${static_tags}") == 0) {
+                vt->vt = MQTT_SCHEMA_STATIC_TAGS;
+            } else if (strcmp(str_val, "${tag_values}") == 0) {
+                vt->vt = MQTT_SCHEMA_TAGVALUES;
+            } else if (strcmp(str_val, "${static_tag_values}") == 0) {
+                vt->vt = MQTT_SCHEMA_STATIC_TAGVALUES;
+            } else if (strcmp(str_val, "${tag_error_values}") == 0) {
+                vt->vt = MQTT_SCHEMA_TAG_ERROR_VALUES;
+            } else if (strcmp(str_val, "${tag_errors}") == 0) {
+                vt->vt = MQTT_SCHEMA_TAG_ERRORS;
+            } else {
+                if (str_val[0] == '$' && str_val[1] == '{' &&
+                    str_val[strlen(str_val) - 1] == '}') {
+                    return -1;
+                } else {
+                    vt->vt = MQTT_SCHEMA_UD;
+                    strcpy(vt->ud, str_val);
+                }
+            }
+        } else if (json_is_object(value)) {
+            if (deep >= 3) {
+                return -1;
+            }
+
+            *vts_len += 1;
+            *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t));
+            mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1];
+
+            memset(vt, 0, sizeof(mqtt_schema_vt_t));
+            strcpy(vt->name, key);
+
+            vt->vt = MQTT_SCHEMA_OBJECT;
+            return mqtt_schema_parse(value, &vt->sub_vts, &vt->n_sub_vts,
+                                     deep + 1);
+        } else {
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
 int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts,
                          size_t *vts_len)
 {
@@ -10,71 +84,24 @@ int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts,
         return -1;
     }
 
-    *vts_len = 0;
-    *vts     = NULL;
-
-    const char *key   = NULL;
-    json_t *    value = NULL;
-
-    json_object_foreach(root, key, value)
-    {
-        if (!json_is_string(value)) {
-            json_decref(root);
-            if (*vts_len > 0) {
-                free(*vts);
-            }
-            return -1;
-        }
-
-        const char *str_val = json_string_value(value);
-        if (str_val == NULL) {
-            json_decref(root);
-            if (*vts_len > 0) {
-                free(*vts);
-            }
-            return -1;
-        }
-
-        *vts_len += 1;
-        *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t));
-        mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1];
-
-        memset(vt, 0, sizeof(mqtt_schema_vt_t));
-        strcpy(vt->name, key);
-
-        if (strcmp(str_val, "${timestamp}") == 0) {
-            vt->vt = MQTT_SCHEMA_TIMESTAMP;
-        } else if (strcmp(str_val, "${node}") == 0) {
-            vt->vt = MQTT_SCHEMA_NODE_NAME;
-        } else if (strcmp(str_val, "${group}") == 0) {
-            vt->vt = MQTT_SCHEMA_GROUP_NAME;
-        } else if (strcmp(str_val, "${tag_values}") == 0) {
-            vt->vt = MQTT_SCHEMA_TAG_VALUES;
-        } else if (strcmp(str_val, "${static_tags}") == 0) {
-            vt->vt = MQTT_SCHEMA_STATIC_TAG_VALUES;
-        } else if (strcmp(str_val, "${tag_errors}") == 0) {
-            vt->vt = MQTT_SCHEMA_TAG_ERRORS;
-        } else {
-            if (str_val[0] == '$' && str_val[1] == '{' &&
-                str_val[strlen(str_val) - 1] == '}') {
-                json_decref(root);
-                free(*vts);
-                return -1;
-            } else {
-                vt->vt = MQTT_SCHEMA_UD;
-                strcpy(vt->ud, str_val);
-            }
+    int ret = mqtt_schema_parse(root, vts, vts_len, 1);
+    if (ret != 0) {
+        if (*vts != NULL) {
+            free(*vts);
+            *vts = NULL;
         }
+        json_decref(root);
+        return -1;
     }
 
     json_decref(root);
     return 0;
 }
 
-int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
-                       mqtt_schema_vt_t *vts, size_t n_vts,
-                       mqtt_static_vt_t *s_tags, size_t n_s_tags,
-                       char **result_str)
+static void *schema_encode(char *driver, char *group,
+                           neu_json_read_resp_t *tags, mqtt_schema_vt_t *vts,
+                           size_t n_vts, mqtt_static_vt_t *s_tags,
+                           size_t n_s_tags)
 {
     void *root = neu_json_encode_new();
 
@@ -95,7 +122,7 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
             elem.t         = NEU_JSON_STR;
             elem.v.val_str = group;
             break;
-        case MQTT_SCHEMA_TAG_VALUES: {
+        case MQTT_SCHEMA_TAGS: {
             void *values_array = neu_json_array();
 
             neu_json_read_resp_tag_t *p_tag = tags->tags;
@@ -130,7 +157,37 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
             elem.v.val_object = values_array;
             break;
         }
-        case MQTT_SCHEMA_STATIC_TAG_VALUES: {
+        case MQTT_SCHEMA_TAGVALUES: {
+            void *values_array = neu_json_encode_new();
+
+            neu_json_read_resp_tag_t *p_tag = tags->tags;
+
+            for (int j = 0; j < tags->n_tag; j++) {
+                neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
+
+                if (p_tag->error == 0) {
+                    tag_elems[0].name = p_tag->name;
+                    tag_elems[0].t    = p_tag->t;
+                    tag_elems[0].v    = p_tag->value;
+
+                    for (int k = 0; k < p_tag->n_meta; k++) {
+                        tag_elems[1 + k].name = p_tag->metas[k].name;
+                        tag_elems[1 + k].t    = p_tag->metas[k].t;
+                        tag_elems[1 + k].v    = p_tag->metas[k].value;
+                    }
+
+                    neu_json_encode_field(values_array, tag_elems,
+                                          1 + p_tag->n_meta);
+                }
+
+                p_tag++;
+            }
+
+            elem.t            = NEU_JSON_OBJECT;
+            elem.v.val_object = values_array;
+            break;
+        }
+        case MQTT_SCHEMA_STATIC_TAGS: {
             void *static_array = neu_json_array();
 
             for (size_t k = 0; k < n_s_tags; k++) {
@@ -153,6 +210,24 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
 
             break;
         }
+        case MQTT_SCHEMA_STATIC_TAGVALUES: {
+            void *static_array = neu_json_encode_new();
+
+            for (size_t k = 0; k < n_s_tags; k++) {
+                neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
+
+                tag_elems[0].name = s_tags[k].name;
+                tag_elems[0].t    = s_tags[k].jtype;
+                tag_elems[0].v    = s_tags[k].jvalue;
+
+                neu_json_encode_field(static_array, tag_elems, 1);
+            }
+
+            elem.t            = NEU_JSON_OBJECT;
+            elem.v.val_object = static_array;
+
+            break;
+        }
         case MQTT_SCHEMA_TAG_ERRORS: {
             void *errors_array = neu_json_array();
 
@@ -181,15 +256,56 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
             elem.v.val_object = errors_array;
             break;
         }
+        case MQTT_SCHEMA_TAG_ERROR_VALUES: {
+            void *errors_array = neu_json_encode_new();
+
+            neu_json_read_resp_tag_t *p_tag = tags->tags;
+
+            for (int k = 0; k < tags->n_tag; k++) {
+                neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
+
+                if (p_tag->error != 0) {
+                    tag_elems[0].name      = p_tag->name;
+                    tag_elems[0].t         = NEU_JSON_INT;
+                    tag_elems[0].v.val_int = p_tag->error;
+
+                    neu_json_encode_field(errors_array, tag_elems, 1);
+                }
+
+                p_tag++;
+            }
+
+            elem.t            = NEU_JSON_OBJECT;
+            elem.v.val_object = errors_array;
+            break;
+        }
         case MQTT_SCHEMA_UD:
             elem.t         = NEU_JSON_STR;
             elem.v.val_str = vts[i].ud;
             break;
+        case MQTT_SCHEMA_OBJECT: {
+            void *sub_root = schema_encode(driver, group, tags, vts[i].sub_vts,
+                                           vts[i].n_sub_vts, s_tags, n_s_tags);
+            elem.t         = NEU_JSON_OBJECT;
+            elem.v.val_object = sub_root;
+            break;
+        }
         }
 
         neu_json_encode_field(root, &elem, 1);
     }
 
+    return root;
+}
+
+int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
+                       mqtt_schema_vt_t *vts, size_t n_vts,
+                       mqtt_static_vt_t *s_tags, size_t n_s_tags,
+                       char **result_str)
+{
+    void *root =
+        schema_encode(driver, group, tags, vts, n_vts, s_tags, n_s_tags);
+
     int ret = neu_json_encode(root, result_str);
     neu_json_decode_free(root);
     return ret;

plugins/mqtt/schema.h

@@ -32,20 +32,29 @@ extern "C" {
 #include "plugin.h"
 
 typedef enum {
-    MQTT_SCHEMA_TIMESTAMP         = 0,
-    MQTT_SCHEMA_NODE_NAME         = 1,
-    MQTT_SCHEMA_GROUP_NAME        = 2,
-    MQTT_SCHEMA_TAG_VALUES        = 3,
-    MQTT_SCHEMA_STATIC_TAG_VALUES = 4,
-    MQTT_SCHEMA_TAG_ERRORS        = 5,
-    MQTT_SCHEMA_UD                = 6,
+    MQTT_SCHEMA_TIMESTAMP        = 0,
+    MQTT_SCHEMA_NODE_NAME        = 1,
+    MQTT_SCHEMA_GROUP_NAME       = 2,
+    MQTT_SCHEMA_TAGS             = 3,
+    MQTT_SCHEMA_STATIC_TAGS      = 4,
+    MQTT_SCHEMA_TAGVALUES        = 5,
+    MQTT_SCHEMA_STATIC_TAGVALUES = 6,
+    MQTT_SCHEMA_TAG_ERRORS       = 7,
+    MQTT_SCHEMA_TAG_ERROR_VALUES = 8,
+    MQTT_SCHEMA_UD               = 9,
+    MQTT_SCHEMA_OBJECT           = 10,
 } mqtt_schema_vt_e;
 
-typedef struct {
+typedef struct mqtt_schema_vt mqtt_schema_vt_t;
+typedef struct mqtt_schema_vt {
     char name[128];
 
     mqtt_schema_vt_e vt;
-    char             ud[128];
+
+    char ud[128];
+
+    mqtt_schema_vt_t *sub_vts;
+    size_t            n_sub_vts;
 } mqtt_schema_vt_t;
 
 int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts,

plugins/restful/normal_handle.c

@@ -558,8 +558,8 @@ void handle_get_plugin_schema_resp(nng_aio *aio, neu_resp_check_schema_t *resp)
         size_t len         = 0;
         char * buf         = NULL;
 
-        if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
-                             resp->schema)) {
+        if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
+                             g_plugin_dir, resp->schema)) {
             NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
                 neu_http_response(aio, error_code.error, result_error);
             });
@@ -581,7 +581,7 @@ void handle_get_plugin_schema_resp(nng_aio *aio, neu_resp_check_schema_t *resp)
 
         if (NULL == buf) {
             free(schema_path);
-            if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
+            if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json",
                                  g_plugin_dir, resp->schema)) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
                     neu_http_response(aio, error_code.error, result_error);

plugins/restful/rest.c

@@ -177,7 +177,7 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/restful/rw_handle.c

@@ -284,7 +284,7 @@ void handle_write(nng_aio *aio)
             }
 
             if (req->t == NEU_JSON_ARRAY_INT64 &&
-                req->value.val_array_int64.length > NEU_VALUE_SIZE) {
+                req->value.val_array_int64.length > NEU_ARRAY_SIZE) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
                     neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
                                       result_error);

src/adapter/driver/cache.c

@@ -726,6 +726,13 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                 }
             }
 
+            if (elem->value.type == NEU_TYPE_PTR) {
+                if (elem->value.value.ptr.ptr != NULL) {
+                    free(elem->value.value.ptr.ptr);
+                    elem->value.value.ptr.ptr = NULL;
+                }
+            }
+
             if (elem->value.type == NEU_TYPE_ARRAY_STRING) {
                 for (int i = 0; i < elem->value.value.strs.length; i++) {
                     free(elem->value.value.strs.strs[i]);

src/adapter/driver/driver.c

@@ -40,6 +40,7 @@
 #include "errcodes.h"
 #include "tag.h"
 
+#include "core/node_manager.h"
 #include "otel/otel_manager.h"
 
 typedef struct to_be_write_tag {
@@ -1113,21 +1114,19 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (result < (int64_t) -FLT_MAX ||
-                       result > (int64_t) FLT_MAX) {
+            }
+            if (result < (int64_t) -FLT_MAX || result > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
             }
-            return check_value(write_type, result, INT16_MIN, INT16_MAX);
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
         case NEU_TYPE_BOOL:
         case NEU_TYPE_BIT:
@@ -1171,20 +1170,19 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (value < (int64_t) -FLT_MAX ||
-                       value > (int64_t) FLT_MAX) {
+            }
+            if (value < (int64_t) -FLT_MAX || value > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
+            if (write_type != NEU_TYPE_DOUBLE && write_type != NEU_TYPE_INT64) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
             }
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
             if (write_type != NEU_TYPE_STRING) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
@@ -2102,19 +2100,17 @@ static int report_callback(void *usr_data)
 
     neu_otel_trace_ctx trans_trace = NULL;
     neu_otel_scope_ctx trans_scope = NULL;
-    if (trace_ctx) {
-        data->trace_ctx = trace_ctx;
-        if (neu_otel_data_is_started() && data->trace_ctx) {
-            trans_trace = neu_otel_find_trace(data->trace_ctx);
-            if (trans_trace) {
-                char new_span_id[36] = { 0 };
-                neu_otel_new_span_id(new_span_id);
-                trans_scope =
-                    neu_otel_add_span2(trans_trace, "report cb", new_span_id);
-                neu_otel_scope_add_span_attr_int(trans_scope, "thread id",
-                                                 (int64_t)(pthread_self()));
-                neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns());
-            }
+    if (neu_otel_data_is_started() && trace_ctx) {
+        trans_trace = neu_otel_find_trace(trace_ctx);
+        if (trans_trace) {
+            data->trace_ctx      = trace_ctx;
+            char new_span_id[36] = { 0 };
+            neu_otel_new_span_id(new_span_id);
+            trans_scope =
+                neu_otel_add_span2(trans_trace, "report cb", new_span_id);
+            neu_otel_scope_add_span_attr_int(trans_scope, "thread id",
+                                             (int64_t)(pthread_self()));
+            neu_otel_scope_set_span_start_time(trans_scope, neu_time_ns());
         }
     }
 
@@ -2143,9 +2139,26 @@ static int report_callback(void *usr_data)
                 }
             }
 
+            if (trans_trace) {
+                neu_otel_set_internal_parent_span(trans_trace);
+            }
+
             utarray_foreach(group->apps, sub_app_t *, app)
             {
 
+                neu_adapter_t *sub_app =
+                    neu_node_manager_find(g_manager->node_manager, app->app);
+                if (sub_app != NULL) {
+                    if (sub_app->state != NEU_NODE_RUNNING_STATE_RUNNING) {
+                        neu_trans_data_free(data);
+                        continue;
+                    }
+                } else {
+                    nlog_error("sub app %s not exist", app->app);
+                    neu_trans_data_free(data);
+                    continue;
+                }
+
                 if (group->driver->adapter.cb_funs.responseto(
                         &group->driver->adapter, &header, data, app->addr) !=
                     0) {

src/core/manager_internal.h

@@ -30,7 +30,7 @@
 #include "subscribe.h"
 #include "utils/log.h"
 
-typedef struct neu_manager {
+struct neu_manager {
     int server_fd;
 
     neu_events_t *  events;
@@ -45,7 +45,7 @@ typedef struct neu_manager {
     int64_t timestamp_lev_manager;
 
     int log_level;
-} neu_manager_t;
+};
 
 int       neu_manager_add_plugin(neu_manager_t *manager, const char *library);
 int       neu_manager_del_plugin(neu_manager_t *manager, const char *plugin);

src/main.c

@@ -35,15 +35,15 @@
 #include "daemon.h"
 #include "version.h"
 
-static bool           exit_flag         = false;
-static neu_manager_t *g_manager         = NULL;
-zlog_category_t *     neuron            = NULL;
-bool                  disable_jwt       = false;
-bool                  sub_filter_err    = false;
-int                   default_log_level = ZLOG_LEVEL_NOTICE;
-char                  host_port[32]     = { 0 };
-char                  g_status[32]      = { 0 };
-static bool           sig_trigger       = false;
+static bool      exit_flag         = false;
+neu_manager_t *  g_manager         = NULL;
+zlog_category_t *neuron            = NULL;
+bool             disable_jwt       = false;
+bool             sub_filter_err    = false;
+int              default_log_level = ZLOG_LEVEL_NOTICE;
+char             host_port[32]     = { 0 };
+char             g_status[32]      = { 0 };
+static bool      sig_trigger       = false;
 
 int64_t global_timestamp = 0;
 

src/otel/otel_manager.c

@@ -56,6 +56,7 @@ typedef struct {
     int64_t                                     ts;
     UT_array *                                  scopes;
     pthread_mutex_t                             mutex;
+    ProtobufCBinaryData                         internal_parent_span;
 } trace_ctx_t;
 
 typedef struct {
@@ -158,6 +159,18 @@ static int parse_tracestate(const char *tracestate, trace_kv_t *kvs,
     return 1;
 }
 
+void neu_otel_set_internal_parent_span(neu_otel_trace_ctx ctx)
+{
+    trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
+    pthread_mutex_lock(&trace_ctx->mutex);
+    if (utarray_len(trace_ctx->scopes) > 0) {
+        trace_scope_t *scope =
+            *(trace_scope_t **) utarray_back(trace_ctx->scopes);
+        trace_ctx->internal_parent_span = scope->span->span_id;
+    }
+    pthread_mutex_unlock(&trace_ctx->mutex);
+}
+
 void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span)
 {
     for (size_t i = 0; i < span->n_attributes; i++) {
@@ -186,6 +199,20 @@ void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span)
 neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
                                          uint32_t flags, const char *tracestate)
 {
+    trace_ctx_table_ele_t *find = NULL;
+
+    pthread_mutex_lock(&table_mutex);
+
+    HASH_FIND(hh, traces_table, &req_ctx, sizeof(req_ctx), find);
+
+    if (find) {
+        HASH_DEL(traces_table, find);
+        neu_otel_free_trace(find->ctx);
+        free(find);
+    }
+
+    pthread_mutex_unlock(&table_mutex);
+
     trace_ctx_t *ctx = calloc(1, sizeof(trace_ctx_t));
     opentelemetry__proto__trace__v1__traces_data__init(&ctx->trace_data);
     strncpy((char *) ctx->trace_id, trace_id, 64);
@@ -458,7 +485,16 @@ neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx)
     pthread_mutex_unlock(&table_mutex);
 
     if (find) {
-        return (void *) find->ctx;
+        if (find->ctx->final &&
+            find->ctx->span_num ==
+                find->ctx->trace_data.resource_spans[0]
+                    ->scope_spans[0]
+                    ->n_spans &&
+            find->ctx->expected_span_num <= 0) {
+            return NULL;
+        } else {
+            return (void *) find->ctx;
+        }
     } else {
         return NULL;
     }
@@ -670,7 +706,11 @@ neu_otel_scope_ctx neu_otel_add_span2(neu_otel_trace_ctx ctx,
     scope->span->flags         = trace_ctx->flags;
     neu_otel_scope_set_span_name(scope, span_name);
     neu_otel_scope_set_span_id(scope, span_id);
-    if (scope->span_index != 0) {
+    if (trace_ctx->internal_parent_span.data != NULL) {
+        neu_otel_scope_set_parent_span_id2(scope,
+                                           trace_ctx->internal_parent_span.data,
+                                           trace_ctx->internal_parent_span.len);
+    } else if (scope->span_index != 0) {
         neu_otel_scope_set_parent_span_id2(
             scope,
             trace_ctx->trace_data.resource_spans[0]
@@ -932,8 +972,10 @@ int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out)
 
 void neu_otel_new_span_id(char *id)
 {
+    int64_t p_id = (int64_t) pthread_self();
     for (int i = SPAN_ID_LENGTH - 1; i >= 0; i--) {
-        id[i] = ID_CHARSET[rand() % 16];
+        int64_t rand_id = (int64_t) rand() + neu_time_ns() + p_id;
+        id[i]           = ID_CHARSET[rand_id % 16];
     }
     id[SPAN_ID_LENGTH] = '\0';
 }
@@ -984,19 +1026,25 @@ void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id,
 static int otel_timer_cb(void *data)
 {
     (void) data;
-
+    int                    batch_size = 50;
+    int                    index      = 0;
     trace_ctx_table_ele_t *el = NULL, *tmp = NULL;
 
     pthread_mutex_lock(&table_mutex);
 
     HASH_ITER(hh, traces_table, el, tmp)
     {
-        if (el->ctx->final &&
-            el->ctx->span_num ==
-                el->ctx->trace_data.resource_spans[0]
-                    ->scope_spans[0]
-                    ->n_spans &&
-            el->ctx->expected_span_num <= 0) {
+        if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) {
+            nlog_debug("trace:%s time out", (char *) el->ctx->trace_id);
+            HASH_DEL(traces_table, el);
+            neu_otel_free_trace(el->ctx);
+            free(el);
+        } else if (el->ctx->final &&
+                   el->ctx->span_num ==
+                       el->ctx->trace_data.resource_spans[0]
+                           ->scope_spans[0]
+                           ->n_spans &&
+                   el->ctx->expected_span_num <= 0) {
             int data_size = neu_otel_trace_pack_size(el->ctx);
 
             uint8_t *data_buf = calloc(1, data_size);
@@ -1009,12 +1057,13 @@ static int otel_timer_cb(void *data)
                 HASH_DEL(traces_table, el);
                 neu_otel_free_trace(el->ctx);
                 free(el);
+                index += 1;
+                if (index > batch_size) {
+                    break;
+                }
+            } else {
+                break;
             }
-        } else if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) {
-            nlog_debug("trace:%s time out", (char *) el->ctx->trace_id);
-            HASH_DEL(traces_table, el);
-            neu_otel_free_trace(el->ctx);
-            free(el);
         }
     }
 
@@ -1035,7 +1084,7 @@ void neu_otel_start()
         param.usr_data = NULL;
 
         param.second      = 0;
-        param.millisecond = 100;
+        param.millisecond = 80;
         param.cb          = otel_timer_cb;
         param.type        = NEU_EVENT_TIMER_BLOCK;
 

src/parser/neu_json_rw.c

@@ -358,31 +358,30 @@ int neu_json_encode_read_resp_ecp(void *json_object, void *param)
 
         tag_elems[0].name      = "name";
         tag_elems[0].t         = NEU_JSON_STR;
-        tag_elems[0].v.val_str = p_tag->name;
+        tag_elems[0].v.val_str = p_tag[i].name;
 
-        if (p_tag->error != 0) {
+        if (p_tag[i].error != 0) {
             continue;
         } else {
             tag_elems[1].name      = "value";
-            tag_elems[1].t         = p_tag->t;
-            tag_elems[1].v         = p_tag->value;
-            tag_elems[1].precision = p_tag->precision;
+            tag_elems[1].t         = p_tag[i].t;
+            tag_elems[1].v         = p_tag[i].value;
+            tag_elems[1].precision = p_tag[i].precision;
 
             tag_elems[2].name      = "type";
             tag_elems[2].t         = NEU_JSON_INT;
-            tag_elems[2].v.val_int = neu_json_type_transfer(p_tag->t);
+            tag_elems[2].v.val_int = neu_json_type_transfer(p_tag[i].t);
         }
 
-        for (int k = 0; k < p_tag->n_meta; k++) {
-            tag_elems[3 + k].name = p_tag->metas[k].name;
-            tag_elems[3 + k].t    = p_tag->metas[k].t;
-            tag_elems[3 + k].v    = p_tag->metas[k].value;
+        for (int k = 0; k < p_tag[i].n_meta; k++) {
+            tag_elems[3 + k].name = p_tag[i].metas[k].name;
+            tag_elems[3 + k].t    = p_tag[i].metas[k].t;
+            tag_elems[3 + k].v    = p_tag[i].metas[k].value;
         }
 
-        tag_array =
-            neu_json_encode_array(tag_array, tag_elems, 3 + p_tag->n_meta);
-        has_data = 1;
-        p_tag++;
+        tag_array = neu_json_encode_array_ecp(tag_array, tag_elems,
+                                              3 + p_tag[i].n_meta);
+        has_data  = 1;
     }
 
     neu_json_elem_t resp_elems[] = { {

src/utils/json.c

@@ -346,6 +346,136 @@ static json_t *encode_object(json_t *object, neu_json_elem_t ele)
     return ob;
 }
 
+static json_t *encode_object_ecp(json_t *object, neu_json_elem_t ele)
+{
+    json_t *ob = object;
+
+    switch (ele.t) {
+    case NEU_JSON_BIT:
+        json_object_set_new(ob, ele.name, json_integer(ele.v.val_bit));
+        break;
+    case NEU_JSON_INT:
+        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));
+        break;
+    case NEU_JSON_FLOAT: {
+        double t = ele.v.val_float;
+        if (ele.precision == 0 && ele.bias == 0) {
+            t = format_tag_value(ele.v.val_float);
+        }
+        json_object_set_new(ob, ele.name, json_realp(t, ele.precision));
+        break;
+    }
+    case NEU_JSON_DOUBLE:
+        json_object_set_new(ob, ele.name,
+                            json_realp(ele.v.val_double, ele.precision));
+        break;
+    case NEU_JSON_BOOL:
+        json_object_set_new(ob, ele.name, json_boolean(ele.v.val_bool));
+        break;
+
+#define ENCODE_ARRAY_TO_STRING(TYPE, FIELD, DATA, FORMAT)                      \
+    case TYPE: {                                                               \
+        size_t buffer_size = 2;                                                \
+        for (int i = 0; i < ele.v.FIELD.length; i++) {                         \
+            buffer_size += snprintf(NULL, 0, FORMAT, ele.v.FIELD.DATA[i]) + 2; \
+        }                                                                      \
+        char *array_str = (char *) malloc(buffer_size);                        \
+        if (array_str == NULL) {                                               \
+            return NULL;                                                       \
+        }                                                                      \
+        char *ptr = array_str;                                                 \
+        *ptr++    = '[';                                                       \
+        for (int i = 0; i < ele.v.FIELD.length; i++) {                         \
+            int len = sprintf(ptr, FORMAT, ele.v.FIELD.DATA[i]);               \
+            ptr += len;                                                        \
+            if (i < ele.v.FIELD.length - 1) {                                  \
+                *ptr++ = ',';                                                  \
+                *ptr++ = ' ';                                                  \
+            }                                                                  \
+        }                                                                      \
+        *ptr++ = ']';                                                          \
+        *ptr   = '\0';                                                         \
+        json_object_set_new(ob, ele.name, json_string(array_str));             \
+        free(array_str);                                                       \
+        break;                                                                 \
+    }
+
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_BOOL, val_array_bool, bools, "%d")
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_INT8, val_array_int8, i8s,
+                               "%" PRId8)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_UINT8, val_array_uint8, u8s,
+                               "%" PRIu8)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_INT16, val_array_int16, i16s,
+                               "%" PRId16)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_UINT16, val_array_uint16, u16s,
+                               "%" PRIu16)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_INT32, val_array_int32, i32s,
+                               "%" PRId32)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_UINT32, val_array_uint32, u32s,
+                               "%" PRIu32)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_INT64, val_array_int64, i64s,
+                               "%" PRId64)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_UINT64, val_array_uint64, u64s,
+                               "%" PRIu64)
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_FLOAT, val_array_float, f32s,
+                               "%.6f")
+        ENCODE_ARRAY_TO_STRING(NEU_JSON_ARRAY_DOUBLE, val_array_double, f64s,
+                               "%.6f")
+
+#undef ENCODE_ARRAY_TO_STRING
+
+    case NEU_JSON_ARRAY_STR: {
+        size_t buffer_size = 2;
+        for (int i = 0; i < ele.v.val_array_str.length; i++) {
+            buffer_size +=
+                snprintf(NULL, 0, "\"%s\"", ele.v.val_array_str.p_strs[i]) + 2;
+        }
+
+        char *array_str = (char *) malloc(buffer_size);
+        if (array_str == NULL) {
+            return NULL;
+        }
+
+        char *ptr = array_str;
+        *ptr++    = '[';
+
+        for (int i = 0; i < ele.v.val_array_str.length; i++) {
+            int len = sprintf(ptr, "\"%s\"", ele.v.val_array_str.p_strs[i]);
+            ptr += len;
+
+            if (i < ele.v.val_array_str.length - 1) {
+                *ptr++ = ',';
+                *ptr++ = ' ';
+            }
+        }
+
+        *ptr++ = ']';
+        *ptr   = '\0';
+
+        json_object_set_new(ob, ele.name, json_string(array_str));
+        free(array_str);
+        break;
+    }
+    case NEU_JSON_OBJECT: {
+        char *json_str = json_dumps(ele.v.val_object, JSON_ENCODE_ANY);
+        if (json_str == NULL) {
+            return NULL;
+        }
+
+        json_object_set_new(ob, ele.name, json_string(json_str));
+        free(json_str);
+        break;
+    }
+    default:
+        break;
+    }
+
+    return ob;
+}
+
 static int decode_object(json_t *root, neu_json_elem_t *ele)
 {
     json_t *ob = NULL;
@@ -910,6 +1040,21 @@ void *neu_json_encode_array(void *array, neu_json_elem_t *t, int n)
     return array;
 }
 
+void *neu_json_encode_array_ecp(void *array, neu_json_elem_t *t, int n)
+{
+    if (array == NULL) {
+        array = json_array();
+    }
+
+    json_t *ob = json_object();
+    for (int j = 0; j < n; j++) {
+        encode_object_ecp(ob, t[j]);
+    }
+
+    json_array_append_new(array, ob);
+    return array;
+}
+
 void *neu_json_encode_new()
 {
     return json_object();

tests/ft/app/test_ekuiper.py

@@ -25,7 +25,7 @@ PLUGIN = "eKuiper"
 DRIVER = "modbus"
 DRIVER_PORT = random_port()
 GROUP = "group"
-INTERVAL = 100
+INTERVAL = 200
 TAGS = [
     {
         "name": "tag0",
@@ -126,7 +126,7 @@ def conf_bad_port(conf_base):
 
 @pytest.fixture(scope="function")
 def mocker():
-    mock = Mocker(timeout=1000)
+    mock = Mocker(timeout=2000)
     yield mock
     mock.disconnect()
 
@@ -159,7 +159,8 @@ class Mocker:
 
     def recv(self):
         msg = self.sock.recv()
-        msg = msg[26:]
+        if msg[0:2] == b'\x0A\xCE':
+            msg = msg[26:]
         return json.loads(msg.decode())
 
     def send(self, data):

tests/ft/driver/test_modbus.py

@@ -582,12 +582,6 @@ class TestModbus:
         assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()[
             'error']
 
-        response = api.write_tag(
-            node=param[0], group='group', tag=hold_float[0]['name'], value=1)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()[
-            'error']
-
         response = api.write_tags(node=param[0], group='group', tag_values=[
             {"tag": hold_int16[0]['name'], "value": 32768},
             {"tag": hold_uint16[0]['name'], "value": 2},

tests/ut/mqtt_schema_test.cc

@@ -29,10 +29,10 @@ TEST(validate_success, schema)
     EXPECT_EQ(vts[2].vt, MQTT_SCHEMA_GROUP_NAME);
 
     EXPECT_STREQ(vts[3].name, "tags");
-    EXPECT_EQ(vts[3].vt, MQTT_SCHEMA_TAG_VALUES);
+    EXPECT_EQ(vts[3].vt, MQTT_SCHEMA_TAGVALUES);
 
     EXPECT_STREQ(vts[4].name, "static");
-    EXPECT_EQ(vts[4].vt, MQTT_SCHEMA_STATIC_TAG_VALUES);
+    EXPECT_EQ(vts[4].vt, MQTT_SCHEMA_STATIC_TAGS);
 
     EXPECT_STREQ(vts[5].name, "errors");
     EXPECT_EQ(vts[5].vt, MQTT_SCHEMA_TAG_ERRORS);
@@ -44,6 +44,51 @@ TEST(validate_success, schema)
     free(vts);
 }
 
+TEST(validate_deep_success, schema)
+{
+    const char *schema = "{\"timestamp\": \"${timestamp}\", \"object\": "
+                         "{\"group\":\"${group}\", \"sub_object\": "
+                         "{\"tags\":\"${tags}\", \"cus\":\"aabb\"}}}";
+    mqtt_schema_vt_t *vts;
+    size_t            n_vts;
+    int               ret = mqtt_schema_validate(schema, &vts, &n_vts);
+    EXPECT_EQ(ret, 0);
+    EXPECT_EQ(n_vts, 2);
+
+    EXPECT_STREQ(vts[0].name, "timestamp");
+    EXPECT_EQ(vts[0].vt, MQTT_SCHEMA_TIMESTAMP);
+
+    EXPECT_STREQ(vts[1].name, "object");
+    EXPECT_EQ(vts[1].vt, MQTT_SCHEMA_OBJECT);
+    EXPECT_EQ(vts[1].n_sub_vts, 2);
+
+    EXPECT_STREQ(vts[1].sub_vts[0].name, "group");
+    EXPECT_EQ(vts[1].sub_vts[0].vt, MQTT_SCHEMA_GROUP_NAME);
+
+    EXPECT_STREQ(vts[1].sub_vts[1].name, "sub_object");
+    EXPECT_EQ(vts[1].sub_vts[1].vt, MQTT_SCHEMA_OBJECT);
+    EXPECT_EQ(vts[1].sub_vts[1].n_sub_vts, 2);
+
+    EXPECT_STREQ(vts[1].sub_vts[1].sub_vts[0].name, "tags");
+    EXPECT_EQ(vts[1].sub_vts[1].sub_vts[0].vt, MQTT_SCHEMA_TAGS);
+
+    EXPECT_STREQ(vts[1].sub_vts[1].sub_vts[1].name, "cus");
+    EXPECT_EQ(vts[1].sub_vts[1].sub_vts[1].vt, MQTT_SCHEMA_UD);
+
+    for (size_t i = 0; i < n_vts; i++) {
+        if (vts[i].vt == MQTT_SCHEMA_OBJECT) {
+            for (size_t k = 0; k < vts[i].n_sub_vts; k++) {
+                if (vts[i].sub_vts[k].vt == MQTT_SCHEMA_OBJECT) {
+                    free(vts[i].sub_vts[k].sub_vts);
+                }
+            }
+            free(vts[i].sub_vts);
+        }
+    }
+
+    free(vts);
+}
+
 TEST(validate_failed, schema)
 {
     const char *schema =
@@ -100,6 +145,41 @@ TEST(validate_success, schema_static)
     mqtt_static_free(vts, n_vts);
 }
 
+TEST(schema_encode_deep, schema_encode)
+{
+    const char *schema = "{\"timestamp\": \"${timestamp}\", \"object\": "
+                         "{\"group\":\"${group}\", \"sub_object\": "
+                         "{\"node\":\"${node}\", \"cus\":\"aabb\"}}}";
+    mqtt_schema_vt_t *vts;
+    size_t            n_vts;
+    int               ret = mqtt_schema_validate(schema, &vts, &n_vts);
+    EXPECT_EQ(ret, 0);
+    EXPECT_EQ(n_vts, 2);
+
+    char *               result = NULL;
+    neu_json_read_resp_t tags   = { 0 };
+    ret = mqtt_schema_encode((char *) "driver", (char *) "group", &tags, vts,
+                             n_vts, NULL, 0, &result);
+    EXPECT_EQ(ret, 0);
+    EXPECT_STREQ(result,
+                 "{\"timestamp\": 0, \"object\": {\"group\": \"group\", "
+                 "\"sub_object\": {\"node\": \"driver\", \"cus\": \"aabb\"}}}");
+    free(result);
+
+    for (size_t i = 0; i < n_vts; i++) {
+        if (vts[i].vt == MQTT_SCHEMA_OBJECT) {
+            for (size_t k = 0; k < vts[i].n_sub_vts; k++) {
+                if (vts[i].sub_vts[k].vt == MQTT_SCHEMA_OBJECT) {
+                    free(vts[i].sub_vts[k].sub_vts);
+                }
+            }
+            free(vts[i].sub_vts);
+        }
+    }
+
+    free(vts);
+}
+
 TEST(schema_encode, schema_static)
 {
     const char *schema =
@@ -134,7 +214,7 @@ TEST(schema_encode, schema_static)
     EXPECT_STREQ(
         result,
         "{\"timestamp\": 0, \"node\": \"driver\", \"group\": \"group\", "
-        "\"tags\": [], \"static\": [{\"name\": \"static_tag1\", \"value\": "
+        "\"tags\": {}, \"static\": [{\"name\": \"static_tag1\", \"value\": "
         "\"string\"}, {\"name\": \"static_tag2\", \"value\": 1234}, {\"name\": "
         "\"static_tag3\", \"value\": 11.44}, {\"name\": \"static_tag4\", "
         "\"value\": true}, {\"name\": \"static_tag5\", \"value\": \"[1, 3, "

version

@@ -1 +1 @@
-2.11.0-alpha
+2.11.2