feat: add tracing

crates/block-producer/Cargo.toml

@@ -42,7 +42,6 @@ async-channel = "1.4.2"
 async-jsonrpc-client = { version = "0.3.0", default-features = false, features = ["http-tokio"] }
 clap = "2.33.3"
 ctrlc = {version = "3.2.1", features = ["termination"]}
-env_logger = "0.8.3"
 futures = "0.3.13"
 log = "0.4.14"
 serde_json = "1.0"
@@ -53,9 +52,15 @@ hex = "0.4"
 async-trait = "0.1"
 semver = "1.0"
 rayon = "1.5"
-sentry = "0.23.0"
-sentry-log = "0.23.0"
 thiserror = "1.0"
+# For latest tracing
+sentry = { git = "https://github.com/getsentry/sentry-rust", rev = "df694a49595d6890c510d80b85cfbb4b5ae6159a" }
+sentry-tracing = { git = "https://github.com/getsentry/sentry-rust", rev = "df694a49595d6890c510d80b85cfbb4b5ae6159a" }
+tracing = { version = "0.1", features = ["attributes"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "parking_lot", "smallvec", "tracing-log"] }
+tracing-opentelemetry = "0.16"
+opentelemetry-jaeger = { version = "0.15", features = ["rt-tokio"] }
+opentelemetry = { version = "0.16", features = ["rt-tokio"] }
 
 [target.'cfg(all(not(target_env = "msvc"), not(target_os="macos")))'.dependencies]
 tikv-jemallocator = { version = "0.4.0", features = ["unprefixed_malloc_on_supported_platforms"] }

crates/block-producer/src/block_producer.rs

@@ -50,6 +50,7 @@ use std::{
     time::{Duration, Instant},
 };
 use tokio::sync::Mutex;
+use tracing::instrument;
 
 const MAX_BLOCK_OUTPUT_PARAM_RETRY_COUNT: usize = 10;
 const TRANSACTION_SCRIPT_ERROR: &str = "TransactionScriptError";
@@ -79,7 +80,9 @@ fn generate_custodian_cells(
     deposit_cells.iter().map(to_custodian).collect()
 }
 
+#[instrument(skip_all)]
 async fn resolve_tx_deps(rpc_client: &RPCClient, tx_hash: [u8; 32]) -> Result<Vec<CellInfo>> {
+    #[instrument(skip_all)]
     async fn resolve_dep_group(rpc_client: &RPCClient, dep: CellDep) -> Result<Vec<CellDep>> {
         // return dep
         if dep.dep_type() == DepType::Code.into() {
@@ -237,6 +240,7 @@ impl BlockProducer {
         self.last_submitted_tx_hash.clone()
     }
 
+    #[instrument(skip_all, fields(event = %event))]
     pub async fn handle_event(&mut self, event: ChainEvent) -> Result<()> {
         if let Some(ref tests_control) = self.tests_control {
             match tests_control.payload().await {
@@ -372,6 +376,7 @@ impl BlockProducer {
         Ok(())
     }
 
+    #[instrument(skip_all, fields(retry_count = retry_count))]
     async fn compose_next_block_submit_tx(
         &mut self,
         median_time: Duration,
@@ -535,6 +540,7 @@ impl BlockProducer {
         }
     }
 
+    #[instrument(skip_all, fields(block = block_number))]
     async fn submit_block_tx(
         &mut self,
         block_number: u64,
@@ -645,6 +651,7 @@ impl BlockProducer {
         }
     }
 
+    #[instrument(skip_all, fields(block = args.block.raw().number().unpack()))]
     async fn complete_tx_skeleton(&self, args: CompleteTxArgs) -> Result<Transaction> {
         let CompleteTxArgs {
             deposit_cells,

crates/block-producer/src/custodian.rs

@@ -1,9 +1,11 @@
 use anyhow::Result;
 use gw_rpc_client::rpc_client::{QueryResult, RPCClient};
 use gw_types::offchain::CollectedCustodianCells;
+use tracing::instrument;
 
 pub const MAX_CUSTODIANS: usize = 50;
 
+#[instrument(skip_all, fields(last_finalized_block_number = last_finalized_block_number))]
 pub async fn query_mergeable_custodians(
     rpc_client: &RPCClient,
     collected_custodians: CollectedCustodianCells,
@@ -31,6 +33,7 @@ pub async fn query_mergeable_custodians(
     .await
 }
 
+#[instrument(skip_all, fields(last_finalized_block_number = last_finalized_block_number))]
 async fn query_mergeable_ckb_custodians(
     rpc_client: &RPCClient,
     collected: CollectedCustodianCells,
@@ -49,6 +52,7 @@ async fn query_mergeable_ckb_custodians(
         .await
 }
 
+#[instrument(skip_all, fields(last_finalized_block_number = last_finalized_block_number))]
 async fn query_mergeable_sudt_custodians(
     rpc_client: &RPCClient,
     collected: CollectedCustodianCells,

crates/block-producer/src/lib.rs

@@ -11,6 +11,7 @@ pub mod replay_block;
 pub mod runner;
 pub mod stake;
 pub mod test_mode_control;
+pub mod trace;
 pub mod types;
 pub mod utils;
 pub mod withdrawal;

crates/block-producer/src/main.rs

@@ -4,10 +4,9 @@ static GLOBAL_ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
 use anyhow::{Context, Result};
 use clap::{App, Arg, SubCommand};
-use gw_block_producer::{db_block_validator, runner};
+use gw_block_producer::{db_block_validator, runner, trace};
 use gw_config::Config;
 use gw_version::Version;
-use sentry_log::LogFilter;
 use std::{fs, path::Path};
 
 const COMMAND_RUN: &str = "run";
@@ -103,15 +102,18 @@ async fn run_cli() -> Result<()> {
         (COMMAND_RUN, Some(m)) => {
             let config_path = m.value_of(ARG_CONFIG).unwrap();
             let config = read_config(&config_path)?;
+            let _guard = trace::init(config.trace)?;
             runner::run(config, m.is_present(ARG_SKIP_CONFIG_CHECK)).await?;
         }
         (COMMAND_EXAMPLE_CONFIG, Some(m)) => {
             let path = m.value_of(ARG_OUTPUT_PATH).unwrap();
+            let _guard = trace::init(None)?;
             generate_example_config(path)?;
         }
         (COMMAND_VERIFY_DB_BLOCK, Some(m)) => {
             let config_path = m.value_of(ARG_CONFIG).unwrap();
             let config = read_config(&config_path)?;
+            let _guard = trace::init(None)?;
             let from_block: Option<u64> = m.value_of(ARG_FROM_BLOCK).map(str::parse).transpose()?;
             let to_block: Option<u64> = m.value_of(ARG_TO_BLOCK).map(str::parse).transpose()?;
             db_block_validator::verify(config, from_block, to_block).await?;
@@ -120,6 +122,7 @@ async fn run_cli() -> Result<()> {
             // default command: start a Godwoken node
             let config_path = "./config.toml";
             let config = read_config(&config_path)?;
+            let _guard = trace::init(config.trace)?;
             runner::run(config, false).await?;
         }
     };
@@ -130,20 +133,5 @@ async fn run_cli() -> Result<()> {
 /// Default to number of cpus, pass `worker_threads` to manually configure workers.
 #[tokio::main(flavor = "multi_thread")]
 async fn main() {
-    init_log();
     run_cli().await.expect("run cli");
 }
-
-fn init_log() {
-    let logger = env_logger::builder()
-        .parse_env(env_logger::Env::default().default_filter_or("info"))
-        .build();
-    let level = logger.filter();
-    let logger = sentry_log::SentryLogger::with_dest(logger).filter(|md| match md.level() {
-        log::Level::Error | log::Level::Warn => LogFilter::Event,
-        _ => LogFilter::Ignore,
-    });
-    log::set_boxed_logger(Box::new(logger))
-        .map(|()| log::set_max_level(level))
-        .expect("set log");
-}

crates/block-producer/src/poller.rs

@@ -29,6 +29,7 @@ use gw_web3_indexer::indexer::Web3Indexer;
 use serde_json::json;
 use std::{collections::HashSet, sync::Arc};
 use tokio::sync::Mutex;
+use tracing::instrument;
 
 #[derive(thiserror::Error, Debug)]
 #[error("chain updater query l1 tx {tx_hash} error {source}")]
@@ -79,6 +80,7 @@ impl ChainUpdater {
     }
 
     // Start syncing
+    #[instrument(skip_all, fields(event = %_event))]
     pub async fn handle_event(&mut self, _event: ChainEvent) -> Result<()> {
         let initial_syncing = !self.initialized;
         // Always start from last valid tip on l1
@@ -135,6 +137,7 @@ impl ChainUpdater {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     pub async fn try_sync(&mut self) -> anyhow::Result<()> {
         let valid_tip_l1_block_number = {
             let chain = self.chain.lock().await;
@@ -195,6 +198,7 @@ impl ChainUpdater {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     pub async fn update(&mut self, txs: &[Tx]) -> anyhow::Result<()> {
         for tx in txs.iter() {
             self.update_single(&tx.tx_hash).await?;
@@ -204,6 +208,7 @@ impl ChainUpdater {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     async fn update_single(&mut self, tx_hash: &H256) -> anyhow::Result<()> {
         if let Some(last_tx_hash) = &self.last_tx_hash {
             if last_tx_hash == tx_hash {
@@ -306,6 +311,7 @@ impl ChainUpdater {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     async fn find_l2block_on_l1(&self, committed_info: L2BlockCommittedInfo) -> Result<bool> {
         let rpc_client = &self.rpc_client;
         let tx_hash: gw_common::H256 =
@@ -320,6 +326,7 @@ impl ChainUpdater {
         Ok(l1_block_hash == Some(block_hash))
     }
 
+    #[instrument(skip_all)]
     async fn revert_to_valid_tip_on_l1(&mut self) -> Result<()> {
         let db = { self.chain.lock().await.store().begin_transaction() };
         let mut revert_l1_actions = Vec::new();
@@ -404,6 +411,7 @@ impl ChainUpdater {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     fn extract_rollup_action(&self, tx: &Transaction) -> Result<RollupAction> {
         let rollup_type_hash: [u8; 32] = {
             let hash = self.rollup_type_script.calc_script_hash();
@@ -438,6 +446,7 @@ impl ChainUpdater {
         RollupAction::from_slice(&output_type).map_err(|e| anyhow!("invalid rollup action {}", e))
     }
 
+    #[instrument(skip_all)]
     async fn extract_challenge_context(
         &self,
         tx: &Transaction,
@@ -485,6 +494,7 @@ impl ChainUpdater {
         unreachable!("challenge output not found");
     }
 
+    #[instrument(skip_all)]
     async fn extract_deposit_requests(
         &self,
         tx: &Transaction,
@@ -534,6 +544,7 @@ impl ChainUpdater {
     }
 }
 
+#[instrument(skip_all)]
 fn try_parse_deposit_request(
     cell_output: &CellOutput,
     cell_data: &Bytes,

crates/block-producer/src/produce_block.rs

@@ -26,6 +26,7 @@ use gw_types::{
     },
     prelude::*,
 };
+use tracing::instrument;
 
 #[derive(Clone)]
 pub struct ProduceBlockResult {
@@ -45,6 +46,7 @@ pub struct ProduceBlockParam {
 /// this method take txs & withdrawal requests from tx pool and produce a new block
 /// the package method should packs the items in order:
 /// withdrawals, then deposits, finally the txs. Thus, the state-validator can verify this correctly
+#[instrument(skip_all)]
 pub fn produce_block(
     db: &StoreTransaction,
     generator: &Generator,
@@ -172,6 +174,7 @@ pub fn produce_block(
 }
 
 // Generate produce block param
+#[instrument(skip_all, fields(mem_block = mem_block.block_info().number().unpack()))]
 pub fn generate_produce_block_param(
     store: &Store,
     mut mem_block: MemBlock,

crates/block-producer/src/runner.rs

@@ -488,6 +488,7 @@ pub async fn run(config: Config, skip_config_check: bool) -> Result<()> {
         )),
         None => sentry::init(()),
     };
+
     // Enable smol threads before smol::spawn
     let runtime_threads = match std::env::var(SMOL_THREADS_ENV_VAR) {
         Ok(s) => s.parse()?,

crates/block-producer/src/trace.rs

@@ -0,0 +1,51 @@
+use anyhow::Result;
+use gw_config::Trace;
+use sentry_tracing::EventFilter;
+use tracing_subscriber::prelude::*;
+
+pub struct ShutdownGuard {
+    trace: Option<Trace>,
+}
+
+impl Drop for ShutdownGuard {
+    fn drop(&mut self) {
+        if let Some(Trace::Jaeger) = self.trace {
+            opentelemetry::global::shutdown_tracer_provider(); // Sending remaining spans
+        }
+    }
+}
+
+pub fn init(trace: Option<Trace>) -> Result<ShutdownGuard> {
+    let env_filter_layer = tracing_subscriber::EnvFilter::try_from_default_env()
+        .or_else(|_| tracing_subscriber::EnvFilter::try_new("info"))?;
+
+    // NOTE: `traces_sample_rate` in sentry client option is 0.0 by default, which disable sentry
+    // tracing. Here we just use sentry-log feature.
+    let sentry_layer = sentry_tracing::layer().event_filter(|md| match md.level() {
+        &tracing::Level::ERROR | &tracing::Level::WARN => EventFilter::Event,
+        _ => EventFilter::Ignore,
+    });
+
+    let registry = tracing_subscriber::registry()
+        .with(tracing_subscriber::fmt::layer())
+        .with(env_filter_layer)
+        .with(sentry_layer);
+
+    match trace {
+        Some(Trace::Jaeger) => {
+            opentelemetry::global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
+
+            let jaeger_layer = {
+                let tracer = opentelemetry_jaeger::new_pipeline()
+                    .with_service_name("godwoken")
+                    .install_batch(opentelemetry::runtime::Tokio)?;
+                tracing_opentelemetry::layer().with_tracer(tracer)
+            };
+
+            registry.with(jaeger_layer).try_init()?
+        }
+        None => registry.try_init()?,
+    }
+
+    Ok(ShutdownGuard { trace })
+}

crates/config/src/config.rs

@@ -10,6 +10,12 @@ use std::{
     path::PathBuf,
 };
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, Hash)]
+#[serde(rename_all = "lowercase")]
+pub enum Trace {
+    Jaeger,
+}
+
 #[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
 pub struct Config {
     pub node_mode: NodeMode,
@@ -32,6 +38,8 @@ pub struct Config {
     pub store: StoreConfig,
     pub sentry_dsn: Option<String>,
     #[serde(default)]
+    pub trace: Option<Trace>,
+    #[serde(default)]
     pub consensus: ConsensusConfig,
     pub reload_config_github_url: Option<GithubConfigUrl>,
     #[serde(default)]

crates/generator/Cargo.toml

@@ -33,6 +33,7 @@ log = "0.4"
 hex = "0.4"
 tokio = "1.15"
 arc-swap = "1.5"
+tracing = { version = "0.1", features = ["attributes"] }
 
 [dev-dependencies]
 gw-utils = {path = "../utils" }