fix tracing (#136)

* fix tracing

* fix config

* cleanup

Cargo.toml

@@ -27,9 +27,11 @@ http = "0.2.8"
 hyper = "0.14.23"
 log = "0.4.17"
 moka = { version = "0.11.0", features = ["future"] }
-opentelemetry = { version = "0.20.0", features = ["rt-tokio"] }
-opentelemetry-datadog = { version = "0.8.0", features = ["reqwest-client"] }
-opentelemetry-jaeger = { version = "0.19.0", features = ["rt-tokio"] }
+opentelemetry = { version = "0.21.0" }
+opentelemetry-datadog = { version = "0.9.0", features = ["reqwest-client"] }
+opentelemetry-jaeger = { version = "0.20.0", features = ["rt-tokio"] }
+opentelemetry_sdk = { version = "0.21.1", features = ["rt-tokio", "trace"] }
+
 rand = "0.8.5"
 serde = "1.0.152"
 serde_json = "1.0.92"

config.yml

@@ -1,4 +1,4 @@
-extensions:  
+extensions:
   client:
     endpoints:
       - wss://acala-rpc.dwellir.com

src/extensions/telemetry/mod.rs

@@ -1,7 +1,8 @@
 use std::env;
 
 use async_trait::async_trait;
-use opentelemetry::{global, sdk::trace::Tracer, trace::TraceError};
+use opentelemetry::{global, trace::TraceError};
+use opentelemetry_sdk::trace::Tracer;
 use serde::Deserialize;
 
 use super::{Extension, ExtensionRegistry};
@@ -64,7 +65,7 @@ pub fn setup_telemetry(options: &TelemetryConfig) -> Result<Option<Tracer>, Trac
                 tracer = tracer.with_endpoint(agent_endpoint.clone());
             }
 
-            let tracer = tracer.install_batch(opentelemetry::runtime::Tokio)?;
+            let tracer = tracer.install_batch(opentelemetry_sdk::runtime::Tokio)?;
 
             Some(tracer)
         }
@@ -80,7 +81,7 @@ pub fn setup_telemetry(options: &TelemetryConfig) -> Result<Option<Tracer>, Trac
                 tracer = tracer.with_agent_endpoint(agent_endpoint);
             }
 
-            let tracer = tracer.install_batch(opentelemetry::runtime::Tokio)?;
+            let tracer = tracer.install_batch(opentelemetry_sdk::runtime::Tokio)?;
 
             Some(tracer)
         }

src/middlewares/methods/block_tag.rs

@@ -2,10 +2,11 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use jsonrpsee::{core::JsonValue, types::ErrorObjectOwned};
+use opentelemetry::trace::FutureExt;
 
 use crate::{
     extensions::api::EthApi,
-    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod},
+    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod, TRACER},
     utils::{TypeRegistry, TypeRegistryRef},
 };
 
@@ -97,8 +98,12 @@ impl Middleware<CallRequest, Result<JsonValue, ErrorObjectOwned>> for BlockTagMi
         context: TypeRegistry,
         next: NextFn<CallRequest, Result<JsonValue, ErrorObjectOwned>>,
     ) -> Result<JsonValue, ErrorObjectOwned> {
-        let (request, context) = self.replace(request, context).await;
-        next(request, context).await
+        async move {
+            let (request, context) = self.replace(request, context).await;
+            next(request, context).await
+        }
+        .with_context(TRACER.context("block_tag"))
+        .await
     }
 }
 

src/middlewares/methods/cache.rs

@@ -9,8 +9,8 @@ use opentelemetry::trace::FutureExt;
 use crate::{
     config::CacheParams,
     extensions::cache::Cache as CacheExtension,
-    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod},
-    utils::{telemetry, Cache, CacheKey, TypeRegistry, TypeRegistryRef},
+    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod, TRACER},
+    utils::{Cache, CacheKey, TypeRegistry, TypeRegistryRef},
 };
 
 pub struct BypassCache(pub bool);
@@ -62,8 +62,6 @@ impl MiddlewareBuilder<RpcMethod, CallRequest, CallResult> for CacheMiddleware {
     }
 }
 
-const TRACER: telemetry::Tracer = telemetry::Tracer::new("cache-middleware");
-
 #[async_trait]
 impl Middleware<CallRequest, Result<JsonValue, ErrorObjectOwned>> for CacheMiddleware {
     async fn call(
@@ -95,7 +93,7 @@ impl Middleware<CallRequest, Result<JsonValue, ErrorObjectOwned>> for CacheMiddl
 
             result
         }
-        .with_context(TRACER.context("call"))
+        .with_context(TRACER.context("cache"))
         .await
     }
 }

src/middlewares/methods/delay.rs

@@ -1,9 +1,10 @@
 use std::time::Duration;
 
 use async_trait::async_trait;
+use opentelemetry::trace::FutureExt;
 
 use crate::{
-    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod},
+    middlewares::{CallRequest, CallResult, Middleware, MiddlewareBuilder, NextFn, RpcMethod, TRACER},
     utils::{TypeRegistry, TypeRegistryRef},
 };
 
@@ -39,7 +40,11 @@ impl Middleware<CallRequest, CallResult> for DelayMiddleware {
         context: TypeRegistry,
         next: NextFn<CallRequest, CallResult>,
     ) -> CallResult {
-        tokio::time::sleep(self.delay).await;
-        next(request, context).await
+        async move {
+            tokio::time::sleep(self.delay).await;
+            next(request, context).await
+        }
+        .with_context(TRACER.context("delay"))
+        .await
     }
 }