feat: improving tracing configuration and adding tokio-console (#881)

.cargo/config.toml

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

Cargo.toml

@@ -72,15 +72,16 @@ reqwest = { version = "=0.12.4", features = ["json"] }
 tower = "=0.4.13"
 
 # observability
+console-subscriber = "=0.2.0"
 metrics = { version = "=0.21.1", optional = true }
 metrics-exporter-prometheus = { version = "=0.12.2", optional = true }
-tracing = { version = "=0.1.40", features = ["attributes"] }
-tracing-subscriber = { version = "=0.3.18", features = ["env-filter", "json"] }
-tracing-opentelemetry = "=0.23.0"
-sentry = "=0.32.3"
 opentelemetry = "=0.22.0"
-opentelemetry-otlp = "=0.15.0"
 opentelemetry_sdk = { version = "=0.22.1", features = ["rt-tokio"] }
+opentelemetry-otlp = "=0.15.0"
+sentry = "=0.32.3"
+tracing = { version = "=0.1.40", features = ["attributes"] }
+tracing-opentelemetry = "=0.23.0"
+tracing-subscriber = { version = "=0.3.18", features = ["env-filter", "json"] }
 
 
 # storage

docker/Dockerfile.run_stratus

@@ -14,6 +14,7 @@ RUN apt-get install -y libclang-dev cmake
 
 ENV CARGO_PROFILE_RELEASE_DEBUG=1
 ENV JSON_LOGS=1
+ENV NO_COLOR=1
 RUN cargo build --release --features metrics
 
 # Runtime

docker/Dockerfile.run_with_importer

@@ -14,6 +14,7 @@ RUN apt-get install -y libclang-dev cmake
 
 ENV CARGO_PROFILE_RELEASE_DEBUG=1
 ENV JSON_LOGS=1
+ENV NO_COLOR=1
 RUN cargo build --release --bin run-with-importer --features metrics,rocks
 
 # Runtime

docker/Dockerfile.run_with_importer_postgres

@@ -14,6 +14,7 @@ RUN apt-get install -y libclang-dev cmake
 
 ENV CARGO_PROFILE_RELEASE_DEBUG=1
 ENV JSON_LOGS=1
+ENV NO_COLOR=1
 RUN cargo build --release --bin run-with-importer --features metrics
 
 # Runtime

src/bin/importer_online.rs

@@ -16,6 +16,8 @@ use stratus::eth::primitives::Hash;
 use stratus::eth::storage::StratusStorage;
 use stratus::eth::BlockMiner;
 use stratus::eth::Executor;
+use stratus::ext::warn_task_cancellation;
+use stratus::ext::warn_task_tx_closed;
 #[cfg(feature = "metrics")]
 use stratus::infra::metrics;
 use stratus::infra::BlockchainClient;
@@ -125,7 +127,7 @@ async fn start_block_executor(
 ) {
     while let Some((block, receipts)) = backlog_rx.recv().await {
         if cancellation.is_cancelled() {
-            tracing::warn!("exiting block-executor because cancellation");
+            warn_task_cancellation("block-executor");
             break;
         }
 
@@ -151,7 +153,7 @@ async fn start_block_executor(
             metrics::inc_import_online_mined_block(start.elapsed());
         }
     }
-    tracing::warn!("exiting block-executor because backlog channel was closed by the other side");
+    warn_task_tx_closed("block-executor");
 }
 
 // -----------------------------------------------------------------------------
@@ -182,7 +184,7 @@ async fn start_number_fetcher(chain: Arc<BlockchainClient>, cancellation: Cancel
     loop {
         // check cancellation
         if cancellation.is_cancelled() {
-            tracing::warn!("exiting number-fetcher because cancellation");
+            warn_task_cancellation("number-fetcher");
             break;
         }
         tracing::info!("fetching current block number");
@@ -255,7 +257,7 @@ async fn start_block_fetcher(
 ) {
     loop {
         if cancellation.is_cancelled() {
-            tracing::warn!("exiting block-fetcher because cancellation");
+            warn_task_cancellation("block-fetcher");
             break;
         }
 

src/config.rs

@@ -41,6 +41,7 @@ use crate::eth::Consensus;
 use crate::eth::EvmTask;
 use crate::eth::Executor;
 use crate::eth::TransactionRelayer;
+use crate::ext::warn_task_tx_closed;
 use crate::infra::BlockchainClient;
 
 /// Loads .env files according to the binary and environment.
@@ -205,7 +206,7 @@ impl ExecutorConfig {
                         tracing::error!(reason = ?e, "failed to send evm execution result");
                     };
                 }
-                tracing::warn!("stopping evm thread because tx channel was closed");
+                warn_task_tx_closed("evm thread");
             })
             .expect("spawning evm threads should not fail");
         }

src/eth/rpc/rpc_subscriptions.rs

@@ -16,6 +16,9 @@ use crate::eth::primitives::Hash;
 use crate::eth::primitives::LogFilter;
 use crate::eth::primitives::LogMined;
 use crate::ext::not;
+use crate::ext::spawn_named;
+use crate::ext::warn_task_cancellation;
+use crate::ext::warn_task_tx_closed;
 use crate::if_else;
 use crate::infra::metrics;
 
@@ -58,10 +61,10 @@ impl RpcSubscriptions {
     fn spawn_subscriptions_cleaner(subs: Arc<RpcSubscriptionsConnected>, cancellation: CancellationToken) -> JoinHandle<anyhow::Result<()>> {
         tracing::info!("spawning rpc subscriptions cleaner");
 
-        tokio::spawn(async move {
+        spawn_named("rpc::sub::cleaner", async move {
             loop {
                 if cancellation.is_cancelled() {
-                    tracing::warn!("exiting rpc subscription cleaner because of cancellation");
+                    warn_task_cancellation("rpc subscription cleaner");
                     return Ok(());
                 }
 
@@ -89,15 +92,15 @@ impl RpcSubscriptions {
     ) -> JoinHandle<anyhow::Result<()>> {
         tracing::info!("spawning rpc newPendingTransactions notifier");
 
-        tokio::spawn(async move {
+        spawn_named("rpc::sub::newPendingTransactions", async move {
             loop {
                 if cancellation.is_cancelled() {
-                    tracing::warn!("exiting rpc newPendingTransactions notifier because of cancellation");
+                    warn_task_cancellation("rpc newPendingTransactions notifier");
                     return Ok(());
                 }
 
                 let Ok(hash) = rx.recv().await else {
-                    tracing::warn!("stopping newPendingTransactions notifier because tx channel was closed");
+                    warn_task_tx_closed("rpc newPendingTransactions notifier");
                     break;
                 };
 
@@ -116,15 +119,15 @@ impl RpcSubscriptions {
     ) -> JoinHandle<anyhow::Result<()>> {
         tracing::info!("spawning rpc newHeads notifier");
 
-        tokio::spawn(async move {
+        spawn_named("rpc::sub::newHeads", async move {
             loop {
                 if cancellation.is_cancelled() {
-                    tracing::warn!("exiting rpc newHeads notifier because of cancellation");
+                    warn_task_cancellation("rpc newHeads notifier");
                     return Ok(());
                 }
 
                 let Ok(header) = rx.recv().await else {
-                    tracing::warn!("stopping newHeads notifier because tx channel was closed");
+                    warn_task_tx_closed("rpc newHeads notifier");
                     break;
                 };
 
@@ -143,15 +146,15 @@ impl RpcSubscriptions {
     ) -> JoinHandle<anyhow::Result<()>> {
         tracing::info!("spawning rpc logs notifier");
 
-        tokio::spawn(async move {
+        spawn_named("rpc::sub::logs", async move {
             loop {
                 if cancellation.is_cancelled() {
-                    tracing::warn!("exiting rpc logs cleaner because of cancellation");
+                    warn_task_cancellation("rpc logs notifier");
                     return Ok(());
                 }
 
                 let Ok(log) = rx.recv().await else {
-                    tracing::warn!("stopping logs notifier because tx channel was closed");
+                    warn_task_tx_closed("rpc logs notifier");
                     break;
                 };