Merge pull request #2375 from fengzeroz/v2.11

version: 2.11.0-alpha2

.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

@@ -10,7 +10,7 @@ on:
 env:
   VERSION: 2.11.0
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.11
 
 jobs:
   cross-build:
@@ -21,7 +21,7 @@ jobs:
           - [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,9 +52,9 @@ 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
 
@@ -64,7 +64,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.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++)

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/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/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);

src/otel/otel_manager.c

@@ -458,7 +458,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;
     }
@@ -932,8 +941,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';
 }
@@ -1009,6 +1020,8 @@ static int otel_timer_cb(void *data)
                 HASH_DEL(traces_table, el);
                 neu_otel_free_trace(el->ctx);
                 free(el);
+            } else {
+                break;
             }
         } else if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) {
             nlog_debug("trace:%s time out", (char *) el->ctx->trace_id);
@@ -1035,7 +1048,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);
+            neu_json_encode_array(tag_array, tag_elems, 3 + p_tag[i].n_meta);
         has_data = 1;
-        p_tag++;
     }
 
     neu_json_elem_t resp_elems[] = { {

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.0-alpha2