feat: track active requests using jsonrpsee counters (#1746)

Author: dinhani-cw

Cargo.lock

@@ -76,7 +76,7 @@ rlp = "=0.5.2"
 triehash = "=0.8.4"
 
 # network
-jsonrpsee = { version = "=0.23.2", features = ["server", "client"] }
+jsonrpsee = { version = "=0.24.5", features = ["server", "client"] }
 reqwest = { version = "=0.12.4", features = ["json"] }
 tonic = "=0.11.0"
 tower = "=0.4.13"

Cargo.toml

@@ -8,6 +8,7 @@ use futures::future::BoxFuture;
 use jsonrpsee::server::middleware::rpc::layer::ResponseFuture;
 use jsonrpsee::server::middleware::rpc::RpcService;
 use jsonrpsee::server::middleware::rpc::RpcServiceT;
+use jsonrpsee::server::ConnectionGuard;
 use jsonrpsee::types::error::INTERNAL_ERROR_CODE;
 use jsonrpsee::types::Params;
 use jsonrpsee::MethodResponse;
@@ -60,7 +61,6 @@ impl<'a> RpcServiceT<'a> for RpcMiddleware {
     type Future = RpcResponse<'a>;
 
     fn call(&self, mut request: jsonrpsee::types::Request<'a>) -> Self::Future {
-        // track request
         let span = info_span!(
             parent: None,
             "rpc::request",
@@ -91,7 +91,7 @@ impl<'a> RpcServiceT<'a> for RpcMiddleware {
             client = tx_client;
         }
 
-        // trace request
+        // trace event
         Span::with(|s| {
             s.rec_str("rpc_id", &request.id);
             s.rec_str("rpc_client", &client);
@@ -113,15 +113,22 @@ impl<'a> RpcServiceT<'a> for RpcMiddleware {
             "rpc request"
         );
 
-        // metrify request
+        // track metrics
         #[cfg(feature = "metrics")]
         {
+            // started requests
             let tx_ref = tx.as_ref();
             metrics::inc_rpc_requests_started(&client, &method, tx_ref.map(|tx| tx.contract), tx_ref.map(|tx| tx.function));
+
+            // active requests
+            if let Some(guard) = request.extensions.get::<ConnectionGuard>() {
+                let active = guard.max_connections() - guard.available_connections(); // TODO: wait for this to be fixed https://github.com/paritytech/jsonrpsee/issues/1467
+                metrics::set_rpc_requests_active(active as u64);
+            }
         }
-        drop(middleware_enter);
 
         // make span available to rpc-server
+        drop(middleware_enter);
         request.extensions_mut().insert(span);
 
         RpcResponse {

src/eth/rpc/rpc_middleware.rs

@@ -29,8 +29,6 @@ use crate::alias::JsonValue;
 use crate::ext::not;
 use crate::ext::to_json_string;
 use crate::ext::to_json_value;
-#[cfg(feature = "metrics")]
-use crate::infra::metrics;
 
 // -----------------------------------------------------------------------------
 // Tracing service: Span field recorder
@@ -120,42 +118,20 @@ where
             }
         });
 
-        // TODO: temporary metrics from events
-        let fields = to_json_value(event.field_map());
-        #[cfg(feature = "metrics")]
-        {
-            event_to_metrics(&fields);
-        }
-
         // parse metadata and event
         let log = TracingLog {
             timestamp: Utc::now(),
             level: meta.level().as_serde(),
             target: meta.target(),
             thread: std::thread::current(),
-            fields,
+            fields: to_json_value(event.field_map()),
             context,
         };
 
         writeln!(writer, "{}", to_json_string(&log))
     }
 }
 
-#[cfg(feature = "metrics")]
-fn event_to_metrics(json: &JsonValue) {
-    let Some(message) = json.as_object().and_then(|obj| obj.get("message")).and_then(|msg| msg.as_str()) else {
-        return;
-    };
-
-    // jsonrpsee active connections
-    let Some(message) = message.strip_prefix("Accepting new connection ") else {
-        return;
-    };
-    let Some((current, _)) = message.split_once('/') else { return };
-    let Ok(current) = current.parse::<u64>() else { return };
-    metrics::set_rpc_requests_active(current);
-}
-
 #[derive(derive_new::new)]
 struct TracingLog<'a> {
     timestamp: DateTime<Utc>,