Merge pull request #2240 from fengzeroz/v2.10

version: 2.10.0-beta2

.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.10
 
     steps:
       - uses: actions/checkout@v4
@@ -60,25 +60,9 @@ jobs:
           cd build
           ctest --output-on-failure
 
-      - name: create cov report
-        run: |
-            sudo apt-get update
-            sudo apt-get -y install lcov
-            mkdir -p cov_report
-            ./create_cov_report.sh ut
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v4
-        with:
-          verbose: true
-          fail_ci_if_error: true
-          token: ${{ secrets.CODECOV_TOKEN }}
-          directory: ./cov_report/
-          files: ./cov_report/cov-ut.info
-
   pft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.10
     strategy:
       matrix:
         plugin: [core, ekuiper, modbus, mqtt, azure, metrics]
@@ -120,20 +104,4 @@ jobs:
             pytest -s -v tests/ft/app/"test_azure.py"
           else
             pytest -s -v tests/ft/metrics/"test_metrics.py"          
-          fi
-
-      - name: create cov report
-        run: |
-            sudo apt-get update
-            sudo apt-get -y install lcov
-            mkdir -p cov_report
-            ./create_cov_report.sh ${{ matrix.plugin }}
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v4
-        with:
-          verbose: true
-          fail_ci_if_error: true
-          token: ${{ secrets.CODECOV_TOKEN }}
-          directory: ./cov_report/
-          files: ./cov_report/cov-${{ matrix.plugin }}.info
+          fi

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ on:
 env:
   VERSION: 2.10.0
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.10
 
 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.10
 
     steps:
       - uses: actions/checkout@v4
@@ -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.10
 
     steps:
       - uses: actions/checkout@v4

CMakeLists.txt

@@ -175,7 +175,6 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI
 add_subdirectory(plugins/modbus)
 add_subdirectory(plugins/mqtt)
 add_subdirectory(plugins/ekuiper)
-add_subdirectory(plugins/file)
 
 add_subdirectory(simulator)
 

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

default_plugins.json

@@ -5,7 +5,6 @@
 		"libplugin-azure-iot.so",
 		"libplugin-ekuiper.so",
 		"libplugin-modbus-tcp.so",
-		"libplugin-modbus-rtu.so",
-		"libplugin-file.so"
+		"libplugin-modbus-rtu.so"
 	]
 }

include/neuron/adapter.h

@@ -57,6 +57,10 @@ typedef struct adapter_callbacks {
         struct {
             void (*update)(neu_adapter_t *adapter, const char *group,
                            const char *tag, neu_dvalue_t value);
+            void (*update_with_trace)(neu_adapter_t *adapter, const char *group,
+                                      const char *tag, neu_dvalue_t value,
+                                      neu_tag_meta_t *metas, int n_meta,
+                                      void *trace_ctx);
             void (*update_with_meta)(neu_adapter_t *adapter, const char *group,
                                      const char *tag, neu_dvalue_t value,
                                      neu_tag_meta_t *metas, int n_meta);

include/neuron/connection/mqtt_client.h

@@ -134,6 +134,13 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
                             char *topic, uint8_t *payload, uint32_t len,
                             void *data, neu_mqtt_client_publish_cb_t cb);
 
+int neu_mqtt_client_publish_with_trace(neu_mqtt_client_t *client,
+                                       neu_mqtt_qos_e qos, char *topic,
+                                       uint8_t *payload, uint32_t len,
+                                       void *                       data,
+                                       neu_mqtt_client_publish_cb_t cb,
+                                       const char *traceparent);
+
 /** Subscribe to `topic` with service quality `qos`.
  *
  * This function tries to send a `SUBSCRIBE` packet with the given `qos` and

include/neuron/define.h

@@ -53,7 +53,7 @@
 #define NEU_MSG_MAX_SIZE 2048
 
 #define NEU_TAG_META_LENGTH 20
-#define NEU_TAG_META_SIZE 3
+#define NEU_TAG_META_SIZE 8
 
 #define NEU_LOG_LEVEL_DEBUG "debug"
 #define NEU_LOG_LEVEL_INFO "info"

include/neuron/msg.h

@@ -675,6 +675,8 @@ typedef struct neu_resp_tag_value {
     neu_dvalue_t value;
 } neu_resp_tag_value_t;
 
+typedef neu_resp_tag_value_t neu_tag_value_t;
+
 typedef struct neu_resp_tag_value_meta {
     char           tag[NEU_TAG_NAME_LEN];
     neu_dvalue_t   value;
@@ -799,6 +801,7 @@ typedef struct {
 typedef struct {
     char *driver;
     char *group;
+    void *trace_ctx;
 
     neu_reqresp_trans_data_ctx_t *ctx;
     UT_array *                    tags; // neu_resp_tag_value_meta_t

include/neuron/otel/otel_manager.h

@@ -34,6 +34,9 @@ neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx);
 
 void neu_otel_trace_set_final(neu_otel_trace_ctx ctx);
 
+void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num);
+uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx);
+
 neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
 
 void neu_otel_free_trace(neu_otel_trace_ctx ctx);
@@ -73,6 +76,7 @@ int neu_otel_trace_pack_size(neu_otel_trace_ctx ctx);
 int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out);
 
 void neu_otel_new_span_id(char *id);
+void neu_otel_new_trace_id(char *id);
 
 void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id,
                                 uint32_t *flags);

plugins/ekuiper/plugin_ekuiper.c

@@ -314,8 +314,8 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
         NEU_NODE_LINK_STATE_DISCONNECTED == plugin->common.link_state;
     nng_mtx_unlock(plugin->mtx);
 
-    neu_otel_trace_ctx *trace = NULL;
-    neu_otel_scope_ctx  scope = NULL;
+    neu_otel_trace_ctx trace = NULL;
+    neu_otel_scope_ctx scope = NULL;
     if (neu_otel_control_is_started()) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
@@ -361,6 +361,35 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
 
         if (disconnected) {
             plog_debug(plugin, "not connected");
+
+            neu_otel_trace_ctx trans_trace = NULL;
+            neu_otel_scope_ctx trans_scope = NULL;
+            if (neu_otel_data_is_started() && trans_data->trace_ctx) {
+                trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
+                if (trans_trace) {
+                    trans_scope = neu_otel_add_span(trans_trace);
+                    neu_otel_scope_set_span_name(trans_scope, "ekuiper send");
+                    char new_span_id[36] = { 0 };
+                    neu_otel_new_span_id(new_span_id);
+                    neu_otel_scope_set_span_id(trans_scope, new_span_id);
+                    uint8_t *p_sp_id =
+                        neu_otel_scope_get_pre_span_id(trans_scope);
+                    if (p_sp_id) {
+                        neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id,
+                                                           8);
+                    }
+                    neu_otel_scope_add_span_attr_int(trans_scope, "thread id",
+                                                     (int64_t)(pthread_self()));
+                    neu_otel_scope_add_span_attr_int(
+                        trans_scope, "error", NEU_ERR_PLUGIN_DISCONNECTED);
+                    neu_otel_scope_set_span_start_time(trans_scope,
+                                                       neu_time_ms());
+                    neu_otel_scope_set_span_end_time(trans_scope,
+                                                     neu_time_ms());
+                    neu_otel_trace_set_final(trans_trace);
+                }
+            }
+
         } else {
             send_data(plugin, trans_data);
         }

plugins/ekuiper/read_write.c

@@ -20,6 +20,7 @@
 #include <nng/nng.h>
 
 #include "neuron.h"
+#include "otel/otel_manager.h"
 #include "json/neu_json_fn.h"
 #include "json/neu_json_rw.h"
 
@@ -34,6 +35,40 @@ static int send_write_tags_req(neu_plugin_t *             plugin,
                                bool trace_flag, uint8_t *trace_id,
                                uint8_t *span_id);
 
+static int hex_char_to_int(char c)
+{
+    if (c >= '0' && c <= '9')
+        return c - '0';
+    if (c >= 'a' && c <= 'f')
+        return c - 'a' + 10;
+    if (c >= 'A' && c <= 'F')
+        return c - 'A' + 10;
+    return -1;
+}
+
+static int hex_string_to_binary(const char *   hex_string,
+                                unsigned char *binary_array, int max_length)
+{
+    int length = strlen(hex_string);
+    if (length % 2 != 0 || length <= 0)
+        return -1;
+
+    int binary_length = length / 2;
+    if (binary_length > max_length)
+        return -1;
+
+    for (int i = 0; i < binary_length; i++) {
+        int high_nibble = hex_char_to_int(hex_string[2 * i]);
+        int low_nibble  = hex_char_to_int(hex_string[2 * i + 1]);
+        if (high_nibble == -1 || low_nibble == -1)
+            return -1;
+
+        binary_array[i] = (high_nibble << 4) | low_nibble;
+    }
+
+    return binary_length;
+}
+
 static void byte_to_hex(unsigned char byte, char *hex)
 {
     static const char hex_chars[] = "0123456789ABCDEF";
@@ -57,39 +92,89 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
         .plugin     = plugin,
         .trans_data = trans_data,
     };
-    rv = neu_json_encode_by_fn(&resp, json_encode_read_resp, &json_str);
-    if (0 != rv) {
-        plog_error(plugin, "fail encode trans data to json");
-        return;
+
+    neu_otel_trace_ctx trans_trace     = NULL;
+    neu_otel_scope_ctx trans_scope     = NULL;
+    uint8_t *          trace_id        = NULL;
+    char               new_span_id[36] = { 0 };
+    if (neu_otel_data_is_started() && trans_data->trace_ctx) {
+        trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
+        if (trans_trace) {
+            trans_scope = neu_otel_add_span(trans_trace);
+            neu_otel_scope_set_span_name(trans_scope, "ekuiper send");
+            neu_otel_new_span_id(new_span_id);
+            neu_otel_scope_set_span_id(trans_scope, new_span_id);
+            uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
+            if (p_sp_id) {
+                neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
+            }
+            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_ms());
+        }
     }
 
-    nng_msg *msg      = NULL;
-    size_t   json_len = strlen(json_str);
-    rv                = nng_msg_alloc(&msg, json_len);
-    if (0 != rv) {
-        plog_error(plugin, "nng cannot allocate msg");
+    do {
+
+        rv = neu_json_encode_by_fn(&resp, json_encode_read_resp, &json_str);
+        if (0 != rv) {
+            plog_error(plugin, "fail encode trans data to json");
+            break;
+        }
+
+        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) {
+            trace_header_len = 26;
+        }
+
+        rv = nng_msg_alloc(&msg, json_len + trace_header_len);
+        if (0 != rv) {
+            plog_error(plugin, "nng cannot allocate msg");
+            free(json_str);
+            break;
+        }
+
+        if (trans_trace) {
+            neu_otel_scope_add_span_attr_string(trans_scope, "playload",
+                                                json_str);
+            trace_id           = neu_otel_get_trace_id(trans_trace);
+            uint8_t span_id[8] = { 0 };
+            hex_string_to_binary(new_span_id, span_id, 8);
+            uint16_t tarce_header_magic = 0xCE0A;
+            memcpy(nng_msg_body(msg), &tarce_header_magic, 2);
+            memcpy(nng_msg_body(msg) + 2, trace_id, 16);
+            memcpy(nng_msg_body(msg) + 2 + 16, span_id, 8);
+        }
+
+        memcpy(nng_msg_body(msg) + trace_header_len, json_str,
+               json_len); // no null byte
+        plog_debug(plugin, ">> %s", json_str);
         free(json_str);
-        return;
-    }
+        rv = nng_sendmsg(plugin->sock, msg,
+                         NNG_FLAG_NONBLOCK); // TODO: use aio to send message
+        if (0 == rv) {
+            NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1,
+                                     NULL);
+            NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
+                                     NULL);
+            NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S,
+                                     json_len, NULL);
+            NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S,
+                                     json_len, NULL);
+        } else {
+            plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
+            nng_msg_free(msg);
+            NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
+                                     1, NULL);
+        }
+    } while (0);
 
-    memcpy(nng_msg_body(msg), json_str, json_len); // no null byte
-    plog_debug(plugin, ">> %s", json_str);
-    free(json_str);
-    rv = nng_sendmsg(plugin->sock, msg,
-                     NNG_FLAG_NONBLOCK); // TODO: use aio to send message
-    if (0 == rv) {
-        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
-        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
-                                 NULL);
-        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, json_len,
-                                 NULL);
-        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, json_len,
-                                 NULL);
-    } else {
-        plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
-        nng_msg_free(msg);
-        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
-                                 NULL);
+    if (trans_trace) {
+        neu_otel_scope_add_span_attr_int(trans_scope, "error", rv);
+        neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
+        neu_otel_trace_set_final(trans_trace);
     }
 }
 

plugins/file/CMakeLists.txt

@@ -1,11 +0,0 @@
-set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
-
-set(CMAKE_BUILD_RPATH ./)
-file(COPY ${CMAKE_SOURCE_DIR}/plugins/file/file.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
-
-set(PLUGIN_NAME plugin-file)
-set(PLUGIN_SOURCES file_plugin.c)
-add_library(${PLUGIN_NAME} SHARED)
-target_include_directories(${PLUGIN_NAME} PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron)
-target_sources(${PLUGIN_NAME} PRIVATE ${PLUGIN_SOURCES})
-target_link_libraries(${PLUGIN_NAME} neuron-base)

plugins/file/file.json

@@ -1,9 +0,0 @@
-{
-	"tag_regex": [
-		{
-			"type": 13,
-			"regex": ""
-		}
-	],
-	"group_interval": 1000
-}

plugins/file/file_plugin.c

@@ -1,252 +0,0 @@
-/**
- * NEURON IIoT System for Industry 4.0
- * Copyright (C) 2020-2022 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.
- **/
-#include <errno.h>
-#include <stdlib.h>
-
-#include <neuron.h>
-
-#include "errcodes.h"
-
-struct neu_plugin {
-    neu_plugin_common_t common;
-
-    // neu_events_t *events;
-
-    // int file_length;
-};
-
-static neu_plugin_t *driver_open(void);
-
-static int driver_close(neu_plugin_t *plugin);
-static int driver_init(neu_plugin_t *plugin, bool load);
-static int driver_uninit(neu_plugin_t *plugin);
-static int driver_start(neu_plugin_t *plugin);
-static int driver_stop(neu_plugin_t *plugin);
-static int driver_config(neu_plugin_t *plugin, const char *config);
-static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
-                          void *data);
-
-static int driver_tag_validator(const neu_datatag_t *tag);
-static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
-static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
-static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
-                        neu_value_u value);
-
-static const neu_plugin_intf_funs_t plugin_intf_funs = {
-    .open    = driver_open,
-    .close   = driver_close,
-    .init    = driver_init,
-    .uninit  = driver_uninit,
-    .start   = driver_start,
-    .stop    = driver_stop,
-    .setting = driver_config,
-    .request = driver_request,
-
-    .driver.validate_tag  = driver_validate_tag,
-    .driver.group_timer   = driver_group_timer,
-    .driver.group_sync    = NULL,
-    .driver.write_tag     = driver_write,
-    .driver.tag_validator = driver_tag_validator,
-    .driver.write_tags    = NULL,
-    .driver.test_read_tag = NULL,
-    .driver.add_tags      = NULL,
-    .driver.load_tags     = NULL,
-    .driver.del_tags      = NULL,
-};
-
-const neu_plugin_module_t neu_plugin_module = {
-    .version         = NEURON_PLUGIN_VER_1_0,
-    .schema          = "file",
-    .module_name     = "File",
-    .module_descr    = "The plugin is used to read and write files.",
-    .module_descr_zh = "该插件用于读写文件。",
-    .intf_funs       = &plugin_intf_funs,
-    .kind            = NEU_PLUGIN_KIND_SYSTEM,
-    .type            = NEU_NA_TYPE_DRIVER,
-    .display         = true,
-    .single          = false,
-};
-
-static neu_plugin_t *driver_open(void)
-{
-    neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
-
-    neu_plugin_common_init(&plugin->common);
-
-    return plugin;
-}
-
-static int driver_close(neu_plugin_t *plugin)
-{
-    free(plugin);
-
-    return 0;
-}
-
-static int driver_init(neu_plugin_t *plugin, bool load)
-{
-    (void) load;
-    plog_notice(plugin, "%s init success", plugin->common.name);
-    return 0;
-}
-
-static int driver_uninit(neu_plugin_t *plugin)
-{
-    plog_notice(plugin, "%s uninit success", plugin->common.name);
-
-    return 0;
-}
-
-static int driver_start(neu_plugin_t *plugin)
-{
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
-
-    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
-    plog_notice(plugin, "start `%s` plugin success", plugin->common.name);
-
-    return 0;
-}
-
-static int driver_stop(neu_plugin_t *plugin)
-{
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 9999, NULL);
-
-    plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
-    plog_notice(plugin, "stop `%s` plugin success", plugin->common.name);
-
-    return 0;
-}
-static int driver_config(neu_plugin_t *plugin, const char *config)
-{
-    (void) plugin;
-    (void) config;
-
-    return 0;
-}
-
-static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
-                          void *data)
-{
-    (void) plugin;
-    (void) head;
-    (void) data;
-    return 0;
-}
-
-static int driver_tag_validator(const neu_datatag_t *tag)
-{
-    (void) tag;
-    return 0;
-}
-
-static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
-{
-    if (tag->type != NEU_TYPE_STRING) {
-        return NEU_ERR_TAG_TYPE_NOT_SUPPORT;
-    }
-
-    plog_notice(plugin, "validate tag success, name:%s, address:%s, type:%d ",
-                tag->name, tag->address, tag->type);
-
-    return 0;
-}
-
-static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
-{
-    neu_adapter_update_metric_cb_t update_metric =
-        plugin->common.adapter_callbacks->update_metric;
-
-    utarray_foreach(group->tags, neu_datatag_t *, tag)
-    {
-        neu_dvalue_t dvalue = { 0 };
-
-        FILE *fp = fopen(tag->address, "r");
-        if (fp == NULL) {
-            dvalue.type = NEU_TYPE_ERROR;
-            if (errno == ENOENT) {
-                dvalue.value.i32 = NEU_ERR_FILE_NOT_EXIST;
-            } else {
-                dvalue.value.i32 = NEU_ERR_FILE_OPEN_FAILURE;
-            }
-        } else {
-            char *buf = calloc(1, 1024);
-            int   ret = fread(buf, 1024, 1, fp);
-            if (ret < 0) {
-                dvalue.type      = NEU_TYPE_ERROR;
-                dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
-            } else if (ret < NEU_VALUE_SIZE) {
-                dvalue.type = NEU_TYPE_STRING;
-                strcpy(dvalue.value.str, buf);
-            } else {
-                dvalue.type             = NEU_TYPE_PTR;
-                dvalue.value.ptr.type   = NEU_TYPE_STRING;
-                dvalue.value.ptr.length = ret;
-                dvalue.value.ptr.ptr    = (uint8_t *) strdup(buf);
-            }
-
-            fclose(fp);
-            free(buf);
-        }
-
-        plugin->common.adapter_callbacks->driver.update(
-            plugin->common.adapter, group->group_name, tag->name, dvalue);
-    }
-
-    update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, 0, NULL);
-    update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, 0, NULL);
-    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
-    update_metric(plugin->common.adapter, NEU_METRIC_GROUP_LAST_SEND_MSGS, 1,
-                  group->group_name);
-    return 0;
-}
-
-static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
-                        neu_value_u value)
-{
-    FILE *fp = fopen(tag->address, "w");
-    if (fp == NULL) {
-        if (errno == ENOENT) {
-            plugin->common.adapter_callbacks->driver.write_response(
-                plugin->common.adapter, req, NEU_ERR_FILE_NOT_EXIST);
-        } else {
-            plugin->common.adapter_callbacks->driver.write_response(
-                plugin->common.adapter, req, NEU_ERR_FILE_OPEN_FAILURE);
-        }
-    } else {
-        if (tag->type == NEU_TYPE_STRING) {
-            size_t ret = fwrite(value.str, 1, strlen(value.str), fp);
-            if (ret != strlen(value.str)) {
-                plugin->common.adapter_callbacks->driver.write_response(
-                    plugin->common.adapter, req, NEU_ERR_FILE_WRITE_FAILURE);
-            }
-        } else {
-            plugin->common.adapter_callbacks->driver.write_response(
-                plugin->common.adapter, req, NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH);
-        }
-
-        fclose(fp);
-        plugin->common.adapter_callbacks->driver.write_response(
-            plugin->common.adapter, req, 0);
-    }
-    return 0;
-}

plugins/mqtt/mqtt_handle.c

@@ -19,6 +19,7 @@
 
 #include "connection/mqtt_client.h"
 #include "errcodes.h"
+#include "otel/otel_manager.h"
 #include "utils/asprintf.h"
 #include "version.h"
 #include "json/neu_json_mqtt.h"
@@ -27,6 +28,17 @@
 #include "mqtt_handle.h"
 #include "mqtt_plugin.h"
 
+static void to_traceparent(uint8_t *trace_id, char *span_id, char *out)
+{
+    int size = 0;
+    size     = sprintf(out, "00-");
+    for (size_t i = 0; i < 16; i++) {
+        size += sprintf(out + size, "%02x", trace_id[i]);
+    }
+
+    sprintf(out + size, "-%s-00", span_id);
+}
+
 static int tag_values_to_json(UT_array *tags, neu_json_read_resp_t *json)
 {
     int index = 0;
@@ -364,6 +376,24 @@ int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
     return rv;
 }
 
+int publish_with_trace(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
+                       char *payload, size_t payload_len,
+                       const char *traceparent)
+{
+    int rv = neu_mqtt_client_publish_with_trace(
+        plugin->client, qos, topic, (uint8_t *) payload, (uint32_t) payload_len,
+        plugin, publish_cb, traceparent);
+    if (0 != rv) {
+        plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos);
+        NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
+                                 NULL);
+        free(payload);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+    }
+
+    return rv;
+}
+
 void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
                       const uint8_t *payload, uint32_t len, void *data,
                       trace_w3c_t *trace_w3c)
@@ -566,35 +596,87 @@ int handle_trans_data(neu_plugin_t *            plugin,
 {
     int rv = 0;
 
-    if (NULL == plugin->client) {
-        return NEU_ERR_MQTT_IS_NULL;
+    neu_otel_trace_ctx trans_trace       = NULL;
+    neu_otel_scope_ctx trans_scope       = NULL;
+    uint8_t *          trace_id          = NULL;
+    char               trace_parent[128] = { 0 };
+    if (neu_otel_data_is_started() && trans_data->trace_ctx) {
+        trans_trace = neu_otel_find_trace(trans_data->trace_ctx);
+        if (trans_trace) {
+            trans_scope = neu_otel_add_span(trans_trace);
+            neu_otel_scope_set_span_name(trans_scope, "mqtt publish");
+            char new_span_id[36] = { 0 };
+            neu_otel_new_span_id(new_span_id);
+            neu_otel_scope_set_span_id(trans_scope, new_span_id);
+            uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
+            if (p_sp_id) {
+                neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
+            }
+            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_ms());
+
+            trace_id = neu_otel_get_trace_id(trans_trace);
+
+            to_traceparent(trace_id, new_span_id, trace_parent);
+        }
     }
 
-    if (0 == plugin->config.cache &&
-        !neu_mqtt_client_is_connected(plugin->client)) {
-        // cache disable and we are disconnected
-        return NEU_ERR_MQTT_FAILURE;
-    }
+    do {
 
-    const route_entry_t *route = route_tbl_get(
-        &plugin->route_tbl, trans_data->driver, trans_data->group);
-    if (NULL == route) {
-        plog_error(plugin, "no route for driver:%s group:%s",
-                   trans_data->driver, trans_data->group);
-        return NEU_ERR_GROUP_NOT_SUBSCRIBE;
-    }
+        if (NULL == plugin->client) {
+            rv = NEU_ERR_MQTT_IS_NULL;
+            break;
+        }
 
-    char *json_str =
-        generate_upload_json(plugin, trans_data, plugin->config.format);
-    if (NULL == json_str) {
-        plog_error(plugin, "generate upload json fail");
-        return NEU_ERR_EINTERNAL;
-    }
+        if (0 == plugin->config.cache &&
+            !neu_mqtt_client_is_connected(plugin->client)) {
+            // cache disable and we are disconnected
+            rv = NEU_ERR_MQTT_FAILURE;
+            break;
+        }
 
-    char *         topic = route->topic;
-    neu_mqtt_qos_e qos   = plugin->config.qos;
-    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
-    json_str = NULL;
+        const route_entry_t *route = route_tbl_get(
+            &plugin->route_tbl, trans_data->driver, trans_data->group);
+        if (NULL == route) {
+            plog_error(plugin, "no route for driver:%s group:%s",
+                       trans_data->driver, trans_data->group);
+            rv = NEU_ERR_GROUP_NOT_SUBSCRIBE;
+            break;
+        }
+
+        char *json_str =
+            generate_upload_json(plugin, trans_data, plugin->config.format);
+        if (NULL == json_str) {
+            plog_error(plugin, "generate upload json fail");
+            rv = NEU_ERR_EINTERNAL;
+            break;
+        }
+
+        char *         topic = route->topic;
+        neu_mqtt_qos_e qos   = plugin->config.qos;
+
+        if (plugin->config.version == NEU_MQTT_VERSION_V5 && trans_trace) {
+            neu_otel_scope_add_span_attr_string(trans_scope, "playload",
+                                                json_str);
+            rv = publish_with_trace(plugin, qos, topic, json_str,
+                                    strlen(json_str), trace_parent);
+        } else {
+            if (trans_trace) {
+                neu_otel_scope_add_span_attr_string(trans_scope, "playload",
+                                                    json_str);
+            }
+            rv = publish(plugin, qos, topic, json_str, strlen(json_str));
+        }
+
+        json_str = NULL;
+    } while (0);
+
+    if (trans_trace) {
+        neu_otel_scope_add_span_attr_int(trans_scope, "error", rv);
+        neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
+        neu_otel_trace_set_final(trans_trace);
+    }
 
     return rv;
 }

plugins/mqtt/mqtt_handle.h

@@ -31,6 +31,10 @@ extern "C" {
 int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
             char *payload, size_t payload_len);
 
+int publish_with_trace(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
+                       char *payload, size_t payload_len,
+                       const char *traceparent);
+
 void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
                       const uint8_t *payload, uint32_t len, void *data,
                       trace_w3c_t *trace_w3c);

plugins/mqtt/mqtt_plugin_intf.c

@@ -460,8 +460,8 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
         plugin->cache_metric_update_ts = global_timestamp;
     }
 
-    neu_otel_trace_ctx *trace = NULL;
-    neu_otel_scope_ctx  scope = NULL;
+    neu_otel_trace_ctx trace = NULL;
+    neu_otel_scope_ctx scope = NULL;
     if (neu_otel_control_is_started()) {
         trace = neu_otel_find_trace(head->ctx);
         if (trace) {

plugins/restful/rest.c

@@ -174,8 +174,8 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
         return 0;
     }
 
-    neu_otel_trace_ctx *trace = NULL;
-    neu_otel_scope_ctx  scope = NULL;
+    neu_otel_trace_ctx trace = NULL;
+    neu_otel_scope_ctx scope = NULL;
     if (neu_otel_control_is_started()) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {

src/adapter/adapter.c

@@ -593,26 +593,6 @@ static int adapter_responseto(neu_adapter_t *     adapter,
     neu_reqresp_head_t *pheader = neu_msg_get_header(msg);
     strcpy(pheader->sender, adapter->name);
 
-    neu_otel_trace_ctx *trace = NULL;
-    neu_otel_scope_ctx  scope = NULL;
-    if (neu_otel_control_is_started()) {
-        trace = neu_otel_find_trace(header->ctx);
-        if (trace) {
-            scope = neu_otel_add_span(trace);
-            neu_otel_scope_set_span_name(scope, "adapter write tag response");
-            char new_span_id[36] = { 0 };
-            neu_otel_new_span_id(new_span_id);
-            neu_otel_scope_set_span_id(scope, new_span_id);
-            uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(scope);
-            if (p_sp_id) {
-                neu_otel_scope_set_parent_span_id2(scope, p_sp_id, 8);
-            }
-            neu_otel_scope_add_span_attr_int(scope, "thread id",
-                                             (int64_t) pthread_self());
-            neu_otel_scope_set_span_start_time(scope, neu_time_ms());
-        }
-    }
-
     int ret = neu_send_msg_to(adapter->control_fd, &dst, msg);
     if (0 != ret) {
         nlog_error("adapter: %s send responseto %s failed, ret: %d, errno: %d",
@@ -621,10 +601,6 @@ static int adapter_responseto(neu_adapter_t *     adapter,
         neu_msg_free(msg);
     }
 
-    if (neu_otel_control_is_started() && trace) {
-        neu_otel_scope_set_span_end_time(scope, neu_time_ms());
-    }
-
     return ret;
 }
 
@@ -798,8 +774,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_WRITE_TAG: {
         neu_resp_error_t error = { 0 };
 
-        neu_otel_trace_ctx *trace = NULL;
-        neu_otel_scope_ctx  scope = NULL;
+        neu_otel_trace_ctx trace = NULL;
+        neu_otel_scope_ctx scope = NULL;
         if (neu_otel_control_is_started()) {
             trace = neu_otel_find_trace(header->ctx);
             if (trace) {
@@ -854,8 +830,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_WRITE_TAGS: {
         neu_resp_error_t error = { 0 };
 
-        neu_otel_trace_ctx *trace = NULL;
-        neu_otel_scope_ctx  scope = NULL;
+        neu_otel_trace_ctx trace = NULL;
+        neu_otel_scope_ctx scope = NULL;
         if (neu_otel_control_is_started()) {
             trace = neu_otel_find_trace(header->ctx);
             if (trace) {
@@ -911,8 +887,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_WRITE_GTAGS: {
         neu_resp_error_t error = { 0 };
 
-        neu_otel_trace_ctx *trace = NULL;
-        neu_otel_scope_ctx  scope = NULL;
+        neu_otel_trace_ctx trace = NULL;
+        neu_otel_scope_ctx scope = NULL;
         if (neu_otel_control_is_started()) {
             trace = neu_otel_find_trace(header->ctx);
             if (trace) {

src/adapter/driver/cache.c

@@ -49,10 +49,16 @@ struct elem {
     UT_hash_handle hh;
 };
 
+typedef struct {
+    char           key[NEU_GROUP_NAME_LEN];
+    void *         trace_ctx;
+    UT_hash_handle hh;
+} group_trace_t;
+
 struct neu_driver_cache {
     pthread_mutex_t mtx;
-
-    struct elem *table;
+    group_trace_t * trace_table;
+    struct elem *   table;
 };
 
 // static void update_tag_error(neu_driver_cache_t *cache, const char *group,
@@ -94,6 +100,16 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
         }
         free(elem);
     }
+
+    group_trace_t *elem1 = NULL;
+    group_trace_t *tmp1  = NULL;
+
+    HASH_ITER(hh, cache->trace_table, elem1, tmp1)
+    {
+        HASH_DEL(cache->trace_table, elem1);
+        free(elem1);
+    }
+
     pthread_mutex_unlock(&cache->mtx);
 
     pthread_mutex_destroy(&cache->mtx);
@@ -132,6 +148,52 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
     pthread_mutex_unlock(&cache->mtx);
 }
 
+void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
+                                   void *trace_ctx)
+{
+    group_trace_t *elem                    = NULL;
+    char           key[NEU_GROUP_NAME_LEN] = { 0 };
+
+    strcpy(key, group);
+
+    pthread_mutex_lock(&cache->mtx);
+    HASH_FIND(hh, cache->trace_table, &key, sizeof(key), elem);
+
+    if (elem == NULL) {
+        elem = calloc(1, sizeof(group_trace_t));
+
+        strcpy(elem->key, group);
+        elem->trace_ctx = trace_ctx;
+
+        HASH_ADD(hh, cache->trace_table, key, sizeof(key), elem);
+    }
+
+    elem->trace_ctx = trace_ctx;
+
+    pthread_mutex_unlock(&cache->mtx);
+}
+
+void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group)
+{
+    group_trace_t *elem                    = NULL;
+    char           key[NEU_GROUP_NAME_LEN] = { 0 };
+
+    void *trace = NULL;
+
+    strcpy(key, group);
+
+    pthread_mutex_lock(&cache->mtx);
+    HASH_FIND(hh, cache->trace_table, &key, sizeof(key), elem);
+
+    if (elem != NULL) {
+        trace = elem->trace_ctx;
+    }
+
+    pthread_mutex_unlock(&cache->mtx);
+
+    return trace;
+}
+
 void neu_driver_cache_update_change(neu_driver_cache_t *cache,
                                     const char *group, const char *tag,
                                     int64_t timestamp, neu_dvalue_t value,

src/adapter/driver/cache.h

@@ -44,6 +44,11 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
 void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
                           const char *tag);
 
+void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
+                                   void *trace_ctx);
+
+void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group);
+
 typedef struct {
     neu_dvalue_t   value;
     int64_t        timestamp;

src/adapter/driver/driver.c

@@ -104,6 +104,10 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
                               neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values);
+static void update_with_trace(neu_adapter_t *adapter, const char *group,
+                              const char *tag, neu_dvalue_t value,
+                              neu_tag_meta_t *metas, int n_meta,
+                              void *trace_ctx);
 static void update(neu_adapter_t *adapter, const char *group, const char *tag,
                    neu_dvalue_t value);
 static void update_im(neu_adapter_t *adapter, const char *group,
@@ -161,8 +165,8 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
     neu_reqresp_head_t *req    = (neu_reqresp_head_t *) r;
     neu_resp_error_t    nerror = { .error = error };
 
-    neu_otel_trace_ctx *trace = NULL;
-    neu_otel_scope_ctx  scope = NULL;
+    neu_otel_trace_ctx trace = NULL;
+    neu_otel_scope_ctx scope = NULL;
     if (neu_otel_control_is_started()) {
         trace = neu_otel_find_trace(req->ctx);
         if (trace) {
@@ -199,7 +203,7 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 
     req->type = NEU_RESP_ERROR;
 
-    nlog_notice("write tag response <%p>", req->ctx);
+    nlog_notice("write tag response start <%p>", req->ctx);
 
     adapter->cb_funs.response(adapter, req, &nerror);
 
@@ -259,6 +263,18 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
         global_timestamp, n_meta);
 }
 
+static void update_with_trace(neu_adapter_t *adapter, const char *group,
+                              const char *tag, neu_dvalue_t value,
+                              neu_tag_meta_t *metas, int n_meta,
+                              void *trace_ctx)
+{
+    update_with_meta(adapter, group, tag, value, metas, n_meta);
+    if (trace_ctx) {
+        neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
+        neu_driver_cache_update_trace(driver->cache, group, trace_ctx);
+    }
+}
+
 static void update_im(neu_adapter_t *adapter, const char *group,
                       const char *tag, neu_dvalue_t value,
                       neu_tag_meta_t *metas, int n_meta)
@@ -409,6 +425,7 @@ neu_adapter_driver_t *neu_adapter_driver_create()
     driver->adapter.cb_funs.driver.update             = update;
     driver->adapter.cb_funs.driver.write_response     = write_response;
     driver->adapter.cb_funs.driver.update_im          = update_im;
+    driver->adapter.cb_funs.driver.update_with_trace  = update_with_trace;
     driver->adapter.cb_funs.driver.update_with_meta   = update_with_meta;
     driver->adapter.cb_funs.driver.scan_tags_response = scan_tags_response;
     driver->adapter.cb_funs.driver.test_read_tag_response =
@@ -1960,6 +1977,32 @@ static int report_callback(void *usr_data)
     data->group  = strdup(group->name);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
 
+    void *trace_ctx =
+        neu_driver_cache_get_trace(group->driver->cache, group->name);
+
+    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) {
+                trans_scope = neu_otel_add_span(trans_trace);
+                neu_otel_scope_set_span_name(trans_scope, "report cb");
+                char new_span_id[36] = { 0 };
+                neu_otel_new_span_id(new_span_id);
+                neu_otel_scope_set_span_id(trans_scope, new_span_id);
+                uint8_t *p_sp_id = neu_otel_scope_get_pre_span_id(trans_scope);
+                if (p_sp_id) {
+                    neu_otel_scope_set_parent_span_id2(trans_scope, p_sp_id, 8);
+                }
+                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_ms());
+            }
+        }
+    }
+
     read_report_group(global_timestamp,
                       neu_group_get_interval(group->group) *
                           NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
@@ -1970,6 +2013,7 @@ static int report_callback(void *usr_data)
         pthread_mutex_lock(&group->apps_mtx);
 
         if (utarray_len(group->apps) > 0) {
+            int app_num      = 0;
             data->ctx        = calloc(1, sizeof(neu_reqresp_trans_data_ctx_t));
             data->ctx->index = utarray_len(group->apps);
             pthread_mutex_init(&data->ctx->mtx, NULL);
@@ -1980,8 +2024,27 @@ static int report_callback(void *usr_data)
                         &group->driver->adapter, &header, data, app->addr) !=
                     0) {
                     neu_trans_data_free(data);
+                    if (trans_trace) {
+                        neu_otel_scope_add_span_attr_int(trans_scope, app->app,
+                                                         0);
+                    }
+                } else {
+                    app_num += 1;
+                    if (trans_trace) {
+                        neu_otel_scope_add_span_attr_int(trans_scope, app->app,
+                                                         1);
+                    }
                 }
             }
+
+            if (trans_trace) {
+                neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
+                neu_otel_trace_set_expected_span_num(trans_trace, app_num);
+                if (app_num == 0) {
+                    neu_otel_trace_set_final(trans_trace);
+                }
+            }
+
         } else {
             utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tag_value)
             {
@@ -1992,6 +2055,12 @@ static int report_callback(void *usr_data)
             utarray_free(data->tags);
             free(data->group);
             free(data->driver);
+
+            if (trans_trace) {
+                neu_otel_scope_add_span_attr_int(trans_scope, "no sub app", 1);
+                neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
+                neu_otel_trace_set_final(trans_trace);
+            }
         }
 
         pthread_mutex_unlock(&group->apps_mtx);
@@ -1999,6 +2068,11 @@ static int report_callback(void *usr_data)
         utarray_free(data->tags);
         free(data->group);
         free(data->driver);
+        if (trans_trace) {
+            neu_otel_scope_add_span_attr_int(trans_scope, "no tags", 1);
+            neu_otel_scope_set_span_end_time(trans_scope, neu_time_ms());
+            neu_otel_trace_set_final(trans_trace);
+        }
     }
     utarray_free(tags);
     free(data);
@@ -2090,8 +2164,8 @@ static int write_callback(void *usr_data)
         int64_t s_time = 0;
         int64_t e_time = 0;
 
-        neu_otel_trace_ctx *trace = NULL;
-        neu_otel_scope_ctx  scope = NULL;
+        neu_otel_trace_ctx trace = NULL;
+        neu_otel_scope_ctx scope = NULL;
         if (neu_otel_control_is_started()) {
             trace =
                 neu_otel_find_trace(((neu_reqresp_head_t *) wtag->req)->ctx);
@@ -2559,6 +2633,13 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
         tag_value.datatag.option      = tag->option;
         memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH);
 
+        if (tag->attribute == NEU_ATTRIBUTE_WRITE) {
+            tag_value.value.type         = NEU_TYPE_STRING;
+            tag_value.value.value.str[0] = 0;
+            utarray_push_back(tag_values, &tag_value);
+            continue;
+        }
+
         if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
                                       tag_value.metas,
                                       NEU_TAG_META_SIZE) != 0) {

src/base/group.c

@@ -315,8 +315,8 @@ UT_array *neu_group_query_read_tag_paginate(neu_group_t *group,
     };
 
     pthread_mutex_lock(&group->mtx);
-    array = filter_tags_by_page(group->tags, is_readable_and_match_query, &q,
-                                current_page, page_size, total_count);
+    array = filter_tags_by_page(group->tags, match_query, &q, current_page,
+                                page_size, total_count);
     pthread_mutex_unlock(&group->mtx);
 
     return array;

src/base/neu_plugin_common.c

@@ -55,12 +55,11 @@ int neu_plugin_op(neu_plugin_t *plugin, neu_reqresp_head_t head, void *data)
     neu_otel_scope_ctx scope = NULL;
 
     if (head.otel_trace_type == NEU_OTEL_TRACE_TYPE_REST_COMM) {
-        nng_aio *     aio         = (nng_aio *) head.ctx;
-        nng_http_req *nng_req     = nng_aio_get_input(aio, 0);
-        const char *  http_method = nng_http_req_get_method(nng_req);
-
         if (neu_otel_control_is_started()) {
-            const char *trace_parent =
+            nng_aio *     aio         = (nng_aio *) head.ctx;
+            nng_http_req *nng_req     = nng_aio_get_input(aio, 0);
+            const char *  http_method = nng_http_req_get_method(nng_req);
+            const char *  trace_parent =
                 nng_http_req_get_header(nng_req, "traceparent");
             const char *trace_state =
                 nng_http_req_get_header(nng_req, "tracestate");

src/connection/mqtt_client.c

@@ -1601,6 +1601,64 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
     return 0;
 }
 
+int neu_mqtt_client_publish_with_trace(neu_mqtt_client_t *client,
+                                       neu_mqtt_qos_e qos, char *topic,
+                                       uint8_t *payload, uint32_t len,
+                                       void *                       data,
+                                       neu_mqtt_client_publish_cb_t cb,
+                                       const char *                 traceparent)
+{
+    int      rv      = 0;
+    nng_msg *pub_msg = NULL;
+    task_t * task    = NULL;
+
+    if (0 != (rv = nng_mqtt_msg_alloc(&pub_msg, 0))) {
+        log(error, "nng_mqtt_msg_alloc fail: %s", nng_strerror(rv));
+        return -1;
+    }
+
+    if (0 != (rv = nng_mqtt_msg_set_publish_topic(pub_msg, topic))) {
+        nng_msg_free(pub_msg);
+        log(error, "nng_mqtt_msg_set_publish_topic fail: %s", nng_strerror(rv));
+        return -1;
+    }
+
+    nng_mqtt_msg_set_packet_type(pub_msg, NNG_MQTT_PUBLISH);
+    nng_mqtt_msg_set_publish_payload(pub_msg, (uint8_t *) payload, len);
+    nng_mqtt_msg_set_publish_qos(pub_msg, qos);
+
+    if (client->version == MQTT_PROTOCOL_VERSION_v5) {
+        property *plist = mqtt_property_alloc();
+        property *p     = mqtt_property_set_value_strpair(
+            USER_PROPERTY, "traceparent", strlen("traceparent"), traceparent,
+            strlen(traceparent), true);
+        mqtt_property_append(plist, p);
+        nng_mqtt_msg_set_publish_property(pub_msg, plist);
+    }
+
+    nng_mtx_lock(client->mtx);
+    task = client_alloc_task(client);
+    nng_mtx_unlock(client->mtx);
+
+    if (NULL == task) {
+        nng_msg_free(pub_msg);
+        log(error, "client_alloc_task fail");
+        return -1;
+    }
+
+    task->kind        = TASK_PUB;
+    task->pub.cb      = cb;
+    task->pub.qos     = qos;
+    task->pub.topic   = topic;
+    task->pub.payload = payload;
+    task->pub.len     = len;
+    task->pub.data    = data;
+    nng_aio_set_msg(task->aio, pub_msg);
+    nng_send_aio(client->sock, task->aio);
+
+    return 0;
+}
+
 int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
                               const char *topic, void *data,
                               neu_mqtt_client_subscribe_cb_t cb)

src/core/manager.c

@@ -1051,8 +1051,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_GET_TAG:
     case NEU_REQ_NODE_CTL:
     case NEU_REQ_ADD_GROUP: {
-        neu_otel_trace_ctx *trace = NULL;
-        neu_otel_scope_ctx  scope = NULL;
+        neu_otel_trace_ctx trace = NULL;
+        neu_otel_scope_ctx scope = NULL;
         if (neu_otel_control_is_started() &&
             (NEU_REQ_WRITE_TAG == header->type ||
              NEU_REQ_WRITE_TAGS == header->type ||

src/otel/otel_manager.c

@@ -28,6 +28,7 @@
 #include "plugin.h"
 #include "utils/http.h"
 #include "utils/log.h"
+#include "utils/time.h"
 #include "utils/utarray.h"
 #include "utils/uthash.h"
 
@@ -35,7 +36,8 @@
 #include "trace.pb-c.h"
 
 #define SPAN_ID_LENGTH 16
-#define SPAN_ID_CHARSET "0123456789abcdef"
+#define ID_CHARSET "0123456789abcdef"
+#define TRACE_TIME_OUT (3 * 60 * 1000)
 
 bool   otel_flag               = false;
 char   otel_collector_url[128] = { 0 };
@@ -50,6 +52,8 @@ typedef struct {
     uint32_t                                    flags;
     bool                                        final;
     size_t                                      span_num;
+    int32_t                                     expected_span_num;
+    int64_t                                     ts;
     UT_array *                                  scopes;
     pthread_mutex_t                             mutex;
 } trace_ctx_t;
@@ -177,6 +181,7 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
     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);
+    ctx->expected_span_num = 0;
 
     ctx->trace_data.n_resource_spans = 1;
     ctx->trace_data.resource_spans =
@@ -417,6 +422,8 @@ neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
 
     ctx->flags = flags;
 
+    ctx->ts = neu_time_ms();
+
     trace_ctx_table_ele_t *ele = calloc(1, sizeof(trace_ctx_table_ele_t));
     ele->key                   = req_ctx;
     ele->ctx                   = ctx;
@@ -455,6 +462,24 @@ void neu_otel_trace_set_final(neu_otel_trace_ctx ctx)
     trace_ctx->final       = true;
 }
 
+void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num)
+{
+    trace_ctx_t *trace_ctx       = (trace_ctx_t *) ctx;
+    trace_ctx->expected_span_num = num;
+}
+
+uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
+{
+    trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
+    pthread_mutex_lock(&trace_ctx->mutex);
+    uint8_t *id = trace_ctx->trace_data.resource_spans[0]
+                      ->scope_spans[0]
+                      ->spans[0]
+                      ->trace_id.data;
+    pthread_mutex_unlock(&trace_ctx->mutex);
+    return id;
+}
+
 neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id)
 {
     trace_ctx_table_ele_t *el = NULL, *tmp = NULL;
@@ -758,19 +783,24 @@ void neu_otel_scope_set_span_end_time(neu_otel_scope_ctx ctx, int64_t ms)
     scope->span->end_time_unix_nano = ((uint64_t) ms) * 1000000;
     trace_ctx_t *trace_ctx          = (trace_ctx_t *) scope->trace_ctx;
     trace_ctx->span_num += 1;
+    trace_ctx->expected_span_num -= 1;
 }
 
 uint8_t *neu_otel_scope_get_pre_span_id(neu_otel_scope_ctx ctx)
 {
     trace_scope_t *scope     = (trace_scope_t *) ctx;
     trace_ctx_t *  trace_ctx = (trace_ctx_t *) scope->trace_ctx;
+    pthread_mutex_lock(&trace_ctx->mutex);
     if (scope->span_index <= 0) {
+        pthread_mutex_unlock(&trace_ctx->mutex);
         return NULL;
     }
-    return trace_ctx->trace_data.resource_spans[0]
-        ->scope_spans[0]
-        ->spans[scope->span_index - 1]
-        ->span_id.data;
+    uint8_t *id = trace_ctx->trace_data.resource_spans[0]
+                      ->scope_spans[0]
+                      ->spans[scope->span_index - 1]
+                      ->span_id.data;
+    pthread_mutex_unlock(&trace_ctx->mutex);
+    return id;
 }
 
 int neu_otel_trace_pack_size(neu_otel_trace_ctx ctx)
@@ -797,12 +827,18 @@ void neu_otel_new_span_id(char *id)
     counter++;
 
     for (int i = SPAN_ID_LENGTH - 1; i >= 0; i--) {
-        id[i] = SPAN_ID_CHARSET[combined % 16];
+        id[i] = ID_CHARSET[combined % 16];
         combined /= 16;
     }
     id[SPAN_ID_LENGTH] = '\0';
 }
 
+void neu_otel_new_trace_id(char *id)
+{
+    neu_otel_new_span_id(id);
+    neu_otel_new_span_id(id + SPAN_ID_LENGTH);
+}
+
 void neu_otel_split_traceparent(const char *in, char *trace_id, char *span_id,
                                 uint32_t *flags)
 {
@@ -854,7 +890,8 @@ static int otel_timer_cb(void *data)
             el->ctx->span_num ==
                 el->ctx->trace_data.resource_spans[0]
                     ->scope_spans[0]
-                    ->n_spans) {
+                    ->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);
@@ -868,6 +905,11 @@ static int otel_timer_cb(void *data)
                 neu_otel_free_trace(el->ctx);
                 free(el);
             }
+        } 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);
         }
     }
 

version

@@ -1 +1 @@
-2.10.0-alpha
+2.10.0-beta2