feat: Relayer tokio task instrumentation

rust/.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+rustflags = ["--cfg", "tokio_unstable"]

rust/Cargo.toml

@@ -65,6 +65,7 @@ bytes = "1"
 clap = "4"
 color-eyre = "0.6"
 config = "0.13.3"
+console-subscriber = "0.2.0"
 convert_case = "0.6"
 cosmrs = { version = "0.14", default-features = false, features = [
   "cosmwasm",
@@ -170,7 +171,8 @@ tendermint-rpc = { version = "0.32.0", features = ["http-client", "tokio"] }
 thiserror = "1.0"
 time = "0.3"
 tiny-keccak = "2.0.2"
-tokio = { version = "1", features = ["parking_lot"] }
+tokio = { version = "1", features = ["parking_lot", "tracing"] }
+tokio-metrics = { version = "0.3.1", default-features = false }
 tokio-test = "0.4"
 toml_edit = "0.19.14"
 tonic = "0.9.2"

rust/Dockerfile

@@ -27,7 +27,7 @@ RUN \
   --mount=id=cargo,type=cache,sharing=locked,target=/usr/src/target \
   --mount=id=cargo-home-registry,type=cache,sharing=locked,target=/usr/local/cargo/registry \
   --mount=id=cargo-home-git,type=cache,sharing=locked,target=/usr/local/cargo/git \
-    cargo build --release --bin validator --bin relayer --bin scraper && \
+    RUSTFLAGS="--cfg tokio_unstable" cargo build --release --bin validator --bin relayer --bin scraper && \
     mkdir -p /release && \
     cp /usr/src/target/release/validator /release && \
     cp /usr/src/target/release/relayer /release && \

rust/agents/relayer/.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+rustflags = ["--cfg", "tokio_unstable"]

rust/agents/relayer/Cargo.toml

@@ -13,6 +13,7 @@ version.workspace = true
 async-trait.workspace = true
 axum.workspace = true
 config.workspace = true
+console-subscriber.workspace = true
 convert_case.workspace = true
 derive-new.workspace = true
 derive_more.workspace = true
@@ -32,6 +33,7 @@ serde_json.workspace = true
 strum.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "macros", "parking_lot", "rt-multi-thread"] }
+tokio-metrics.workspace = true
 tracing-futures.workspace = true
 tracing.workspace = true
 

rust/agents/relayer/src/msg/op_submitter.rs

@@ -4,10 +4,10 @@ use derive_new::new;
 use futures::future::join_all;
 use futures_util::future::try_join_all;
 use prometheus::{IntCounter, IntGaugeVec};
-use tokio::spawn;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio::time::sleep;
+use tokio_metrics::TaskMonitor;
 use tracing::{debug, info_span, instrument, instrument::Instrumented, trace, Instrument};
 use tracing::{info, warn};
 
@@ -82,12 +82,18 @@ pub struct SerialSubmitter {
     metrics: SerialSubmitterMetrics,
     /// Max batch size for submitting messages
     max_batch_size: u32,
+    /// tokio task monitor
+    task_monitor: TaskMonitor,
 }
 
 impl SerialSubmitter {
     pub fn spawn(self) -> Instrumented<JoinHandle<()>> {
         let span = info_span!("SerialSubmitter", destination=%self.domain);
-        spawn(async move { self.run().await }).instrument(span)
+        let task_monitor = self.task_monitor.clone();
+        tokio::spawn(TaskMonitor::instrument(&task_monitor, async move {
+            self.run().await
+        }))
+        .instrument(span)
     }
 
     async fn run(self) {
@@ -97,6 +103,7 @@ impl SerialSubmitter {
             rx: rx_prepare,
             retry_rx,
             max_batch_size,
+            task_monitor,
         } = self;
         let prepare_queue = OpQueue::new(
             metrics.submitter_queue_length.clone(),
@@ -115,32 +122,40 @@ impl SerialSubmitter {
         );
 
         let tasks = [
-            spawn(receive_task(
-                domain.clone(),
-                rx_prepare,
-                prepare_queue.clone(),
+            tokio::spawn(TaskMonitor::instrument(
+                &task_monitor,
+                receive_task(domain.clone(), rx_prepare, prepare_queue.clone()),
             )),
-            spawn(prepare_task(
-                domain.clone(),
-                prepare_queue.clone(),
-                submit_queue.clone(),
-                confirm_queue.clone(),
-                max_batch_size,
-                metrics.clone(),
+            tokio::spawn(TaskMonitor::instrument(
+                &task_monitor,
+                prepare_task(
+                    domain.clone(),
+                    prepare_queue.clone(),
+                    submit_queue.clone(),
+                    confirm_queue.clone(),
+                    max_batch_size,
+                    metrics.clone(),
+                ),
             )),
-            spawn(submit_task(
-                domain.clone(),
-                submit_queue,
-                confirm_queue.clone(),
-                max_batch_size,
-                metrics.clone(),
+            tokio::spawn(TaskMonitor::instrument(
+                &task_monitor,
+                submit_task(
+                    domain.clone(),
+                    submit_queue,
+                    confirm_queue.clone(),
+                    max_batch_size,
+                    metrics.clone(),
+                ),
             )),
-            spawn(confirm_task(
-                domain.clone(),
-                prepare_queue,
-                confirm_queue,
-                max_batch_size,
-                metrics,
+            tokio::spawn(TaskMonitor::instrument(
+                &task_monitor,
+                confirm_task(
+                    domain.clone(),
+                    prepare_queue,
+                    confirm_queue,
+                    max_batch_size,
+                    metrics,
+                ),
             )),
         ];