Repository: robustmq/robustmq (language: rust)

Issue:
[MINOR] Enable telemetry
### Have you checked the documentation for submitting an Issue?

- [x] Yes.

### What problem does the new feature solve?

Enable telemetry

### What does the feature do?

Tracking code execution and timing

### Implementation challenges

Use opentelemetry lib to develop trace functions

### Are you willing to submit PR?

- [x] Yes. I would be willing to submit a PR with guidance from the RobustMQ community to improve.
- [ ] No. I cannot submit a PR at this time.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 92bd34c13..2baad5dc1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -721,6 +721,9 @@ dependencies = [
  "log4rs",
  "mysql",
  "opendal",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry_sdk",
  "prometheus",
  "prometheus-client",
  "prost",
@@ -2296,6 +2299,7 @@ dependencies = [
  "log",
  "metadata-struct",
  "mysql",
+ "opentelemetry",
  "os_info",
  "paho-mqtt",
  "placement-center",
@@ -2662,6 +2666,83 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "opentelemetry"
+version = "0.27.1"
+source = "git+https://github.com/open-telemetry/opentelemetry-rust.git?rev=b6783a10984146c62ceaa6997fef1385d2ee5ae8#b6783a10984146c62ceaa6997fef1385d2ee5ae8"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "pin-project-lite",
+ "thiserror 2.0.11",
+ "tracing",
+]
+
+[[package]]
+name = "opentelemetry-http"
+version = "0.27.0"
+source = "git+https://github.com/open-telemetry/opentelemetry-rust.git?rev=b6783a10984146c62ceaa6997fef1385d2ee5ae8#b6783a10984146c62ceaa6997fef1385d2ee5ae8"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "http",
+ "opentelemetry",
+ "reqwest",
+ "tracing",
+]
+
+[[package]]
+name = "opentelemetry-otlp"
+version = "0.27.0"
+source = "git+https://github.com/open-telemetry/opentelemetry-rust.git?rev=b6783a10984146c62ceaa6997fef1385d2ee5ae8#b6783a10984146c62ceaa6997fef1385d2ee5ae8"
+dependencies = [
+ "async-trait",
+ "futures-core",
+ "http",
+ "opentelemetry",
+ "opentelemetry-http",
+ "opentelemetry-proto",
+ "opentelemetry_sdk",
+ "prost",
+ "reqwest",
+ "thiserror 2.0.11",
+ "tokio",
+ "tonic",
+ "tracing",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.27.0"
+source = "git+https://github.com/open-telemetry/opentelemetry-rust.git?rev=b6783a10984146c62ceaa6997fef1385d2ee5ae8#b6783a10984146c62ceaa6997fef1385d2ee5ae8"
+dependencies = [
+ "opentelemetry",
+ "opentelemetry_sdk",
+ "prost",
+ "tonic",
+]
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.27.1"
+source = "git+https://github.com/open-telemetry/opentelemetry-rust.git?rev=b6783a10984146c62ceaa6997fef1385d2ee5ae8#b6783a10984146c62ceaa6997fef1385d2ee5ae8"
+dependencies = [
+ "async-trait",
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "glob",
+ "opentelemetry",
+ "percent-encoding",
+ "rand",
+ "serde_json",
+ "thiserror 2.0.11",
+ "tokio",
+ "tokio-stream",
+ "tracing",
+]
+
 [[package]]
 name = "ordered-float"
 version = "2.10.1"
@@ -3407,6 +3488,7 @@ checksum = "c7d6d2a27d57148378eb5e111173f4276ad26340ecc5c49a4a2152167a2d6a37"
 dependencies = [
  "base64 0.22.0",
  "bytes",
+ "futures-channel",
  "futures-core",
  "futures-util",
  "http",
@@ -3501,6 +3583,9 @@ dependencies = [
  "log",
  "metadata-struct",
  "mqtt-broker",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry_sdk",
  "paho-mqtt",
  "protocol",
  "serde",
diff --git a/Cargo.toml b/Cargo.toml
index 65e9151b1..c3c71732c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -77,6 +77,11 @@ grep = "0.3.2"
 prometheus = "0.13.3"
 prometheus_exporter = "0.8"
 prometheus-client = "0.23.1"
+# b6783a10984146c62ceaa6997fef1385d2ee5ae8
+#
+opentelemetry = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "b6783a10984146c62ceaa6997fef1385d2ee5ae8", features = ["trace"] }
+opentelemetry_sdk = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "b6783a10984146c62ceaa6997fef1385d2ee5ae8", features = ["rt-tokio"] }
+opentelemetry-otlp = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "b6783a10984146c62ceaa6997fef1385d2ee5ae8", features = ["grpc-tonic"] }
 ## other
 signal-hook = "0.3.17"
 lazy_static = "^1.4"
diff --git a/docs/.design/observability/telemetry.md b/docs/.design/observability/telemetry.md
new file mode 100644
index 000000000..1965dae76
--- /dev/null
+++ b/docs/.design/observability/telemetry.md
@@ -0,0 +1,46 @@
+1. 增加配置telemetry相关内容
+
+    ```toml
+
+    [telemetry]
+    enable = true # 是否开启
+    exporter_type = "otlp" # otlp 类型
+    exporter_endpoint = "grpc://127.0.0.1:4317" # 导出的数据，这里是用 jaegertracing/all-in-one:latest
+
+    ```
+
+    在 telemetry 中，如果需要跨系统进行追踪需要在不同系统之间传递  Context  。Context 中最重要的数据是
+
+    ```json
+
+    {
+    "traceparent": "00-7ba70cddfcc3f1135b659866752a5556-ca61e958b8f40d0c-01",
+    "tracestate": "",
+    }
+
+    ```
+    traceparent 确保了，系统追踪的连贯性。
+
+2. 创建了 CustomContext 接收传递的 parent_id   ，用于存储分布式其他程序传递的 Context
+
+3. 创建 `src/common/base/src/telemetry/trace.rs` 文件,实现 `init_tracer_provider` 和 `stop_tracer_provider` 方法
+
+4. 在 start 方法中调用 `init_tracer_provider`。在 stop 方法中调用 `stop_tracer_provider`
+
+5. 在需要 trace 的地方，调用
+
+```
+                let tracer = global::tracer("robustmq/publish");
+                let mut span = tracer
+                    .span_builder("command/publish")
+                    .with_kind(SpanKind::Server)
+                    .start_with_context(&tracer, &parent_cx); // 简单的可以理解为这个 span 的计时开始
+
+
+                // span add_event 计时结束
+                span.add_event(
+                    format!("connection_id: {:?}", tcp_connection.connection_id),
+                    vec![],
+                );
+
+```
diff --git a/example/mqtt-cluster/mqtt-server/node-1.toml b/example/mqtt-cluster/mqtt-server/node-1.toml
index 96bed5ed3..2d20193b3 100644
--- a/example/mqtt-cluster/mqtt-server/node-1.toml
+++ b/example/mqtt-cluster/mqtt-server/node-1.toml
@@ -58,3 +58,8 @@ storage_type = "memory"
 [log]
 log_config = "./config/log-config/mqtt-log4rs.yaml"
 log_path = "./robust-data-test/mqtt-broker-1/logs"
+
+[telemetry]
+enable = false
+exporter_type = "otlp"
+exporter_endpoint = "grpc://127.0.0.1:4317"
diff --git a/src/common/base/Cargo.toml b/src/common/base/Cargo.toml
index 5dbd3e3b1..fbcea9239 100644
--- a/src/common/base/Cargo.toml
+++ b/src/common/base/Cargo.toml
@@ -44,3 +44,6 @@ clap.workspace = true
 regex.workspace = true
 opendal.workspace = true
 prometheus-client.workspace = true
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+opentelemetry-otlp.workspace = true
diff --git a/src/common/base/src/config/broker_mqtt.rs b/src/common/base/src/config/broker_mqtt.rs
index 749d76d4b..9aff54cc4 100644
--- a/src/common/base/src/config/broker_mqtt.rs
+++ b/src/common/base/src/config/broker_mqtt.rs
@@ -32,12 +32,13 @@ use std::sync::OnceLock;
 
 use serde::{Deserialize, Serialize};
 
-use super::common::{override_default_by_env, Auth, Log, Storage};
+use super::common::{override_default_by_env, Auth, Log, Storage, Telemetry};
 use super::default_mqtt::{
     default_auth, default_grpc_port, default_http_port, default_log, default_network,
     default_network_quic_port, default_network_tcp_port, default_network_tcps_port,
     default_network_websocket_port, default_network_websockets_port, default_offline_message,
     default_placement_center, default_storage, default_system, default_tcp_thread,
+    default_telemetry,
 };
 use crate::tools::{read_file, try_create_fold};
 
@@ -65,6 +66,8 @@ pub struct BrokerMqttConfig {
     pub log: Log,
     #[serde(default = "default_offline_message")]
     pub offline_messages: OfflineMessage,
+    #[serde(default = "default_telemetry")]
+    pub telemetry: Telemetry,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone, Default)]
diff --git a/src/common/base/src/config/common.rs b/src/common/base/src/config/common.rs
index 1b3ce58d9..449f9f900 100644
--- a/src/common/base/src/config/common.rs
+++ b/src/common/base/src/config/common.rs
@@ -43,6 +43,13 @@ pub struct Log {
     pub log_path: String,
 }
 
+#[derive(Debug, Deserialize, Serialize, Clone, Default, PartialEq, Eq)]
+pub struct Telemetry {
+    pub enable: bool,
+    pub exporter_type: String,
+    pub exporter_endpoint: String,
+}
+
 /** `override_default_by_env` 根据环境变量覆盖内容
 
 ```
diff --git a/src/common/base/src/config/default_mqtt.rs b/src/common/base/src/config/default_mqtt.rs
index 131df5df4..eee3d3f85 100644
--- a/src/common/base/src/config/default_mqtt.rs
+++ b/src/common/base/src/config/default_mqtt.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::broker_mqtt::{Network, OfflineMessage, System, TcpThread};
-use super::common::{Auth, Log, Storage};
+use super::common::{Auth, Log, Storage, Telemetry};
 
 pub fn default_grpc_port() -> u32 {
     9981
@@ -107,3 +107,11 @@ pub fn default_auth() -> Auth {
         mysql_addr: "".to_string(),
     }
 }
+
+pub fn default_telemetry() -> Telemetry {
+    Telemetry {
+        enable: false,
+        exporter_endpoint: "grpc://127.0.0.1:4317".to_string(),
+        exporter_type: "otlp".to_string(),
+    }
+}
diff --git a/src/common/base/src/lib.rs b/src/common/base/src/lib.rs
index 6ee79bcca..63a24f8a5 100644
--- a/src/common/base/src/lib.rs
+++ b/src/common/base/src/lib.rs
@@ -20,6 +20,7 @@ pub mod http_response;
 pub mod logs;
 pub mod metrics;
 pub mod runtime;
+pub mod telemetry;
 pub mod tools;
 pub mod utils;
 pub mod version;
diff --git a/src/common/base/src/telemetry/mod.rs b/src/common/base/src/telemetry/mod.rs
new file mode 100644
index 000000000..d3b78ed63
--- /dev/null
+++ b/src/common/base/src/telemetry/mod.rs
@@ -0,0 +1,15 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod trace;
diff --git a/src/common/base/src/telemetry/trace.rs b/src/common/base/src/telemetry/trace.rs
new file mode 100644
index 000000000..7fb895c89
--- /dev/null
+++ b/src/common/base/src/telemetry/trace.rs
@@ -0,0 +1,140 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::config::broker_mqtt::BrokerMqttConfig;
+use opentelemetry::{global, propagation::Extractor, trace::noop::NoopTracerProvider};
+use opentelemetry_otlp::{SpanExporter, WithExportConfig};
+use opentelemetry_sdk::{propagation::TraceContextPropagator, trace as sdktrace, Resource};
+use std::{collections::HashMap, sync::OnceLock};
+
+// Prepare to support three types
+// 1. For performance, turn it off
+// 2. For debugging, output to standard output
+// 3. For release, output to otlp
+#[derive(Debug)]
+pub enum TraceExporterProvider {
+    Noop(NoopTracerProvider),
+    Stdout(NoopTracerProvider),
+    Otlp(sdktrace::SdkTracerProvider),
+}
+
+static GLOBAL_PROVIDER: OnceLock<TraceExporterProvider> = OnceLock::new();
+
+pub async fn init_tracer_provider(broker_config: &'static BrokerMqttConfig) {
+    if !broker_config.telemetry.enable {
+        global::set_tracer_provider(NoopTracerProvider::new());
+        return;
+    }
+    match broker_config.telemetry.exporter_type.as_str() {
+        "otlp" => {
+            let exporter = SpanExporter::builder()
+                .with_tonic()
+                .with_endpoint(broker_config.telemetry.exporter_endpoint.as_str())
+                .build()
+                .unwrap();
+            global::set_text_map_propagator(TraceContextPropagator::new());
+            let provider = sdktrace::SdkTracerProvider::builder()
+                .with_batch_exporter(exporter)
+                .with_resource(Resource::builder().with_service_name("robustmq").build())
+                .build();
+            GLOBAL_PROVIDER
+                .set(TraceExporterProvider::Otlp(provider.clone()))
+                .unwrap();
+            global::set_tracer_provider(provider.clone());
+        }
+        _ => {
+            global::set_tracer_provider(NoopTracerProvider::new());
+        }
+    }
+}
+
+pub async fn stop_tracer_provider() {
+    if let Some(provider) = GLOBAL_PROVIDER.get() {
+        match provider {
+            TraceExporterProvider::Otlp(provider) => {
+                provider.shutdown().unwrap();
+            }
+            TraceExporterProvider::Noop(_provider) => {
+                // Ignore
+            }
+            TraceExporterProvider::Stdout(_provider) => {
+                // Ignore
+            }
+        }
+    }
+}
+
+pub struct CustomContext {
+    pub inner: HashMap<String, String>,
+}
+
+impl Extractor for CustomContext {
+    /// Get a value for a key from the MetadataMap.  If the value can't be converted to &str, returns None
+    fn get(&self, key: &str) -> Option<&str> {
+        self.inner.get(key).map(|metadata| metadata.as_str())
+    }
+
+    /// Collect all the keys from the MetadataMap.
+    fn keys(&self) -> Vec<&str> {
+        self.inner
+            .keys()
+            .map(|key| key.as_str())
+            .collect::<Vec<_>>()
+    }
+}
+
+impl CustomContext {
+    pub fn new() -> Self {
+        CustomContext {
+            inner: HashMap::new(),
+        }
+    }
+}
+
+impl Default for CustomContext {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::config::broker_mqtt::broker_mqtt_conf;
+    use crate::config::broker_mqtt::init_broker_mqtt_conf_by_path;
+    use std::path::PathBuf;
+
+    use super::init_tracer_provider;
+
+    #[test]
+    fn telemetry_config_test() {
+        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+        path.push("../../../example/mqtt-cluster/mqtt-server/node-1.toml");
+        init_broker_mqtt_conf_by_path(path.to_str().unwrap());
+        let telemetry_config = &broker_mqtt_conf().telemetry;
+        assert!(!telemetry_config.enable);
+        assert_eq!(
+            telemetry_config.exporter_endpoint,
+            "grpc://127.0.0.1:4317".to_string()
+        );
+        assert_eq!(telemetry_config.exporter_type, "otlp".to_string());
+    }
+
+    #[tokio::test]
+    async fn telemetry_test_init() {
+        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+        path.push("../../../example/mqtt-cluster/mqtt-server/node-1.toml");
+        init_broker_mqtt_conf_by_path(path.to_str().unwrap());
+        init_tracer_provider(broker_mqtt_conf()).await
+    }
+}
diff --git a/src/mqtt-broker/Cargo.toml b/src/mqtt-broker/Cargo.toml
index 6d086708f..ccd3b7b22 100644
--- a/src/mqtt-broker/Cargo.toml
+++ b/src/mqtt-broker/Cargo.toml
@@ -53,3 +53,4 @@ grep.workspace = true
 # observability
 prometheus.workspace = true
 prometheus-client.workspace = true
+opentelemetry.workspace = true
diff --git a/src/mqtt-broker/src/handler/command.rs b/src/mqtt-broker/src/handler/command.rs
index 10fef1e4c..4ea3e1f76 100644
--- a/src/mqtt-broker/src/handler/command.rs
+++ b/src/mqtt-broker/src/handler/command.rs
@@ -15,13 +15,6 @@
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use grpc_clients::pool::ClientPool;
-use log::info;
-use protocol::mqtt::common::{
-    is_mqtt3, is_mqtt4, is_mqtt5, ConnectReturnCode, DisconnectReasonCode, MqttPacket, MqttProtocol,
-};
-use storage_adapter::storage::StorageAdapter;
-
 use super::flow_control::is_qos_message;
 use super::mqtt::MqttService;
 use crate::handler::cache::CacheManager;
@@ -32,6 +25,15 @@ use crate::security::AuthDriver;
 use crate::server::connection::NetworkConnection;
 use crate::server::connection_manager::ConnectionManager;
 use crate::subscribe::subscribe_manager::SubscribeManager;
+use common_base::telemetry::trace::CustomContext;
+use grpc_clients::pool::ClientPool;
+use log::info;
+use opentelemetry::global;
+use opentelemetry::trace::{Span, SpanKind, Tracer};
+use protocol::mqtt::common::{
+    is_mqtt3, is_mqtt4, is_mqtt5, ConnectReturnCode, DisconnectReasonCode, MqttPacket, MqttProtocol,
+};
+use storage_adapter::storage::StorageAdapter;
 
 // S: message storage adapter
 #[derive(Clone)]
@@ -188,6 +190,18 @@ where
             }
 
             MqttPacket::Publish(publish, publish_properties) => {
+                let mut context = CustomContext::default();
+                if let Some(ref p) = publish_properties {
+                    p.user_properties.iter().for_each(|(k, v)| {
+                        context.inner.insert(k.clone(), v.clone());
+                    });
+                };
+                let parent_cx = global::get_text_map_propagator(|prop| prop.extract(&context));
+                let tracer = global::tracer("robustmq/publish");
+                let mut span = tracer
+                    .span_builder("command/publish")
+                    .with_kind(SpanKind::Server)
+                    .start_with_context(&tracer, &parent_cx);
                 let connection = if let Some(se) = self
                     .metadata_cache
                     .connection_info
@@ -240,7 +254,11 @@ where
                         connection.recv_qos_message_decr();
                     }
                 }
-
+                // span add_event
+                span.add_event(
+                    format!("connection_id: {:?}", tcp_connection.connection_id),
+                    vec![],
+                );
                 return resp;
             }
 
diff --git a/src/mqtt-broker/src/lib.rs b/src/mqtt-broker/src/lib.rs
index b79299672..a49f4af1d 100644
--- a/src/mqtt-broker/src/lib.rs
+++ b/src/mqtt-broker/src/lib.rs
@@ -147,6 +147,8 @@ where
     }
 
     pub fn start(&self, stop_send: broadcast::Sender<bool>) {
+        self.start_tracer_provider();
+
         self.register_node();
         self.start_cluster_heartbeat_report(stop_send.clone());
 
@@ -169,6 +171,11 @@ where
         self.awaiting_stop(stop_send);
     }
 
+    fn start_tracer_provider(&self) {
+        self.runtime.spawn(async move {
+            common_base::telemetry::trace::init_tracer_provider(broker_mqtt_conf()).await;
+        });
+    }
     fn start_mqtt_server(&self, stop_send: broadcast::Sender<bool>) {
         let cache = self.cache_manager.clone();
         let message_storage_adapter = self.message_storage_adapter.clone();
@@ -407,6 +414,7 @@ where
     async fn stop_server(&self) {
         let cluster_storage = ClusterStorage::new(self.client_pool.clone());
         let config = broker_mqtt_conf();
+        common_base::telemetry::trace::stop_tracer_provider().await;
         match cluster_storage.unregister_node(config).await {
             Ok(()) => {
                 info!("Node {} exits successfully", config.broker_id);


Test patch:
diff --git a/tests/Cargo.toml b/tests/Cargo.toml
index afb10a2f5..6c2a98566 100644
--- a/tests/Cargo.toml
+++ b/tests/Cargo.toml
@@ -37,4 +37,7 @@ mqtt-broker.workspace = true
 grpc-clients.workspace = true
 metadata-struct.workspace = true
 log.workspace = true
-tonic.workspace = true
\ No newline at end of file
+tonic.workspace = true
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+opentelemetry-otlp.workspace = true
diff --git a/tests/tests/mqtt_protocol/pub_qos_test.rs b/tests/tests/mqtt_protocol/pub_qos_test.rs
index fd94b5628..b638bb5d5 100644
--- a/tests/tests/mqtt_protocol/pub_qos_test.rs
+++ b/tests/tests/mqtt_protocol/pub_qos_test.rs
@@ -12,9 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use common_base::tools::unique_id;
+
 use paho_mqtt::{Message, MessageBuilder, Properties, PropertyCode};
 
+use opentelemetry::{global, propagation::Injector};
+use opentelemetry_sdk::Resource;
+use opentelemetry_sdk::{propagation::TraceContextPropagator, trace as sdktrace};
+// use opentelemetry_stdout::SpanExporter;
+
+use opentelemetry::{
+    trace::{SpanKind, TraceContextExt, Tracer},
+    Context, KeyValue,
+};
+
 use crate::mqtt_protocol::common::{broker_addr, connect_server34, connect_server5, distinct_conn};
 
 async fn publish34_qos(num: i32, qos: i32) {
@@ -80,12 +93,98 @@ async fn publish5_qos(num: i32, qos: i32, retained: bool) {
     distinct_conn(cli);
 }
 
+async fn publish5_qos_trace(num: i32, qos: i32, retained: bool) {
+    let client_id = unique_id();
+    let addr = broker_addr();
+    let cli = connect_server5(&client_id, &addr, false, false);
+    let topic = "/tests/t1".to_string();
+
+    //  A program may require many targets
+    let tracer = global::tracer("mqtt/client");
+    let span = tracer
+        .span_builder("cli/publish")
+        .with_kind(SpanKind::Client)
+        .with_attributes([KeyValue::new("component", "mqtt")])
+        .start(&tracer); // start span
+
+    // Tracing of distributed systems requires cross-system context
+    let cx = Context::current_with_span(span);
+
+    let mut new_map = MqttContext::default();
+    let mut props = Properties::default();
+
+    global::get_text_map_propagator(|propagator| propagator.inject_context(&cx, &mut new_map));
+
+    new_map.inner.iter().for_each(|(key, value)| {
+        println!("key: {}, value: {}", key, value);
+        props
+            .push_string_pair(PropertyCode::UserProperty, key, value)
+            .unwrap();
+    });
+
+    props
+        .push_u32(PropertyCode::MessageExpiryInterval, 50)
+        .unwrap();
+    for i in 0..num {
+        let payload = format!("mqtt {i} message");
+        let msg = MessageBuilder::new()
+            .properties(props.clone())
+            .payload(payload)
+            .topic(topic.clone())
+            .qos(qos)
+            .retained(retained)
+            .finalize();
+        match cli.publish(msg) {
+            Ok(_) => {}
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
+    cx.span().add_event(
+        "Got response!",
+        vec![KeyValue::new("status", 200.to_string())],
+    );
+    // If the scope is not left, you must manually end() to ensure the closure of the span.
+    cx.span().end();
+    distinct_conn(cli);
+}
+
+#[derive(Debug)]
+pub struct MqttContext {
+    pub inner: HashMap<String, String>,
+}
+
+impl MqttContext {
+    pub fn new() -> Self {
+        MqttContext {
+            inner: HashMap::new(),
+        }
+    }
+}
+impl Injector for MqttContext {
+    /// Set a key and value in the MetadataMap.  Does nothing if the key or value are not valid inputs
+    fn set(&mut self, key: &str, value: String) {
+        println!("Injecting: {} -> {}", key, value);
+        self.inner.insert(key.to_string(), value);
+    }
+}
+
+impl Default for MqttContext {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[cfg(test)]
 mod tests {
+    use super::*;
+    use core::time;
 
+    use opentelemetry_otlp::SpanExporter;
     use paho_mqtt::{QOS_0, QOS_1, QOS_2};
 
-    use crate::mqtt_protocol::pub_qos_test::{publish34_qos, publish5_qos};
+    use crate::mqtt_protocol::pub_qos_test::{publish34_qos, publish5_qos, publish5_qos_trace};
 
     #[tokio::test]
     async fn client34_publish_test() {
@@ -106,4 +205,27 @@ mod tests {
         publish5_qos(num, QOS_1, true).await;
         publish5_qos(num, QOS_2, true).await;
     }
+
+    #[tokio::test]
+    #[ignore]
+    async fn client5_publish_trace_test() {
+        let num = 1;
+        let exporter = SpanExporter::builder().with_tonic().build().unwrap();
+        global::set_text_map_propagator(TraceContextPropagator::new());
+        let provider = sdktrace::SdkTracerProvider::builder()
+            .with_batch_exporter(exporter)
+            .with_resource(Resource::builder().with_service_name("mqtt-cli").build())
+            .build();
+        global::set_tracer_provider(provider.clone());
+        publish5_qos_trace(num, QOS_1, false).await;
+        tokio::time::sleep(time::Duration::from_millis(10000)).await;
+        match provider.shutdown() {
+            Ok(_) => {
+                println!("shutdown success");
+            }
+            Err(e) => {
+                panic!("{:?}", e);
+            }
+        }
+    }
 }