Added tracing layer

Cargo.toml

@@ -43,9 +43,9 @@ async-stream = "0.3"
 base64 = "0.22"
 bytes = "1"
 chrono = { version = "0.4", default-features = false, features = [
-  "std",
-  "alloc",
-  "clock",
+    "std",
+    "alloc",
+    "clock",
 ] }
 clap = "4"
 colored = "2"
@@ -141,6 +141,9 @@ tokio-native-tls = "0.3"
 tungstenite = "0.24"
 tokio-tungstenite = "0.24"
 
+opentelemetry = { version = "0.26.0", features = ["trace"] }
+tracing-opentelemetry = "0.27.0"
+
 [profile.release]
 opt-level = 3
 debug = false

volo-grpc/Cargo.toml

@@ -67,6 +67,9 @@ tower = { workspace = true, features = [
 ] }
 tracing.workspace = true
 
+opentelemetry = { workspace = true, optional = true }
+tracing-opentelemetry = { workspace = true, optional = true }
+
 tokio-rustls = { workspace = true, optional = true }
 tokio-native-tls = { workspace = true, optional = true }
 
@@ -82,3 +85,5 @@ native-tls = ["__tls", "dep:tokio-native-tls", "volo/native-tls"]
 native-tls-vendored = ["native-tls", "volo/native-tls-vendored"]
 
 grpc-web = ["dep:tonic", "dep:tonic-web"]
+
+opentelemetry = ["dep:opentelemetry", "dep:tracing-opentelemetry"]

volo-grpc/src/layer/mod.rs

@@ -3,4 +3,6 @@ pub mod grpc_timeout;
 #[cfg(feature = "grpc-web")]
 pub mod grpc_web;
 pub mod loadbalance;
+#[cfg(feature = "opentelemetry")]
+pub mod tracing;
 pub mod user_agent;

volo-grpc/src/layer/tracing.rs

@@ -0,0 +1,116 @@
+use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceId};
+use std::fmt::Debug;
+use std::str::FromStr;
+use tracing::Instrument;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+
+use volo::context::Context;
+use volo::{Layer, Service};
+
+use crate::metadata::{KeyRef, MetadataKey, MetadataValue};
+use crate::{Request, Response};
+
+impl opentelemetry::propagation::Extractor for crate::metadata::MetadataMap {
+    fn get(&self, key: &str) -> Option<&str> {
+        self.get(key).and_then(|v| v.to_str().ok())
+    }
+
+    fn keys(&self) -> Vec<&str> {
+        self.keys()
+            .filter_map(|k| match k {
+                KeyRef::Ascii(k) => Some(k.as_str()),
+                KeyRef::Binary(_) => None,
+            })
+            .collect::<Vec<_>>()
+    }
+}
+
+impl opentelemetry::propagation::Injector for crate::metadata::MetadataMap {
+    fn set(&mut self, key: &str, value: String) {
+        self.insert(
+            MetadataKey::from_str(key).unwrap(),
+            MetadataValue::from_str(value.as_str()).unwrap(),
+        );
+    }
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+pub struct ClientTracingLayer;
+
+impl<S> Layer<S> for ClientTracingLayer {
+    type Service = ClientTracingService<S>;
+
+    fn layer(self, inner: S) -> Self::Service {
+        ClientTracingService(inner)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct ClientTracingService<S>(S);
+
+#[volo::service]
+impl<Cx, ReqBody, S> Service<Cx, Request<ReqBody>> for ClientTracingService<S>
+where
+    S: Service<Cx, Request<ReqBody>> + Send + 'static + Sync,
+    Cx: Context<Config = crate::context::Config> + 'static + Send,
+    ReqBody: Send + 'static,
+{
+    async fn call(&self, cx: &mut Cx, mut req: Request<ReqBody>) -> Result<S::Response, S::Error> {
+        let span = tracing::span!(
+            tracing::Level::INFO,
+            "rpc_call",
+            method = cx.rpc_info().method().as_str()
+        );
+
+        let otel_cx = span.context();
+        opentelemetry::global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(&otel_cx, req.metadata_mut());
+        });
+
+        self.0.call(cx, req).await
+    }
+}
+
+pub struct ServerTracingLayer;
+
+impl<S> Layer<S> for ServerTracingLayer {
+    type Service = ServerTracingService<S>;
+
+    fn layer(self, inner: S) -> Self::Service {
+        ServerTracingService(inner)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct ServerTracingService<S>(S);
+
+#[volo::service]
+impl<Cx, ReqBody, ResBody, ResErr, S> Service<Cx, Request<ReqBody>> for ServerTracingService<S>
+where
+    S: Service<Cx, Request<ReqBody>, Response = Response<ResBody>, Error = ResErr>
+        + Send
+        + 'static
+        + Sync,
+    Cx: Context<Config = crate::context::Config> + 'static + Send,
+    ReqBody: Send + 'static,
+    ResBody: Send + 'static,
+    ResErr: Debug + Send + 'static,
+{
+    async fn call(&self, cx: &mut Cx, req: Request<ReqBody>) -> Result<S::Response, S::Error> {
+        let method = cx.rpc_info().method().as_str();
+        let span = tracing::span!(
+            tracing::Level::INFO,
+            "rpc_call",
+            rpc.method = method,
+            otel.name = format!("RPC {}", method),
+            otel.kind = "server",
+        );
+
+        opentelemetry::global::get_text_map_propagator(|propagator| {
+            let cx = propagator.extract(req.metadata());
+            span.set_parent(cx);
+        });
+
+        self.0.call(cx, req).instrument(span).await
+    }
+}