feat: Introduce more tracing instrumentation

core/lib/object_store/src/mirror.rs

@@ -23,7 +23,7 @@ impl<S: ObjectStore> MirroringObjectStore<S> {
 
 #[async_trait]
 impl<S: ObjectStore> ObjectStore for MirroringObjectStore<S> {
-    #[tracing::instrument(skip(self))]
+    #[tracing::instrument(name = "MirroringObjectStore::get_raw", skip(self))]
     async fn get_raw(&self, bucket: Bucket, key: &str) -> Result<Vec<u8>, ObjectStoreError> {
         match self.mirror_store.get_raw(bucket, key).await {
             Ok(object) => {
@@ -49,7 +49,11 @@ impl<S: ObjectStore> ObjectStore for MirroringObjectStore<S> {
         }
     }
 
-    #[tracing::instrument(skip(self, value), fields(value.len = value.len()))]
+    #[tracing::instrument(
+        name = "MirroringObjectStore::put_raw",
+        skip(self, value),
+        fields(value.len = value.len())
+    )]
     async fn put_raw(
         &self,
         bucket: Bucket,
@@ -66,7 +70,7 @@ impl<S: ObjectStore> ObjectStore for MirroringObjectStore<S> {
         Ok(())
     }
 
-    #[tracing::instrument(skip(self))]
+    #[tracing::instrument(name = "MirroringObjectStore::remove_raw", skip(self))]
     async fn remove_raw(&self, bucket: Bucket, key: &str) -> Result<(), ObjectStoreError> {
         self.inner.remove_raw(bucket, key).await?;
         // Only remove the value from the mirror once it has been removed in the underlying store

core/lib/object_store/src/objects.rs

@@ -128,8 +128,15 @@ impl dyn ObjectStore + '_ {
     ///
     /// Returns an error if an object with the `key` does not exist, cannot be accessed,
     /// or cannot be deserialized.
+    #[tracing::instrument(
+        name = "ObjectStore::get",
+        skip_all,
+        fields(key) // Will be recorded within the function.
+    )]
     pub async fn get<V: StoredObject>(&self, key: V::Key<'_>) -> Result<V, ObjectStoreError> {
         let key = V::encode_key(key);
+        // Record the key for tracing.
+        tracing::Span::current().record("key", key.as_str());
         let bytes = self.get_raw(V::BUCKET, &key).await?;
         V::deserialize(bytes).map_err(ObjectStoreError::Serialization)
     }
@@ -140,6 +147,11 @@ impl dyn ObjectStore + '_ {
     ///
     /// Returns an error if an object with the `encoded_key` does not exist, cannot be accessed,
     /// or cannot be deserialized.
+    #[tracing::instrument(
+        name = "ObjectStore::get_by_encoded_key",
+        skip_all,
+        fields(key = %encoded_key)
+    )]
     pub async fn get_by_encoded_key<V: StoredObject>(
         &self,
         encoded_key: String,
@@ -154,12 +166,19 @@ impl dyn ObjectStore + '_ {
     /// # Errors
     ///
     /// Returns an error if serialization or the insertion / replacement operation fails.
+    #[tracing::instrument(
+        name = "ObjectStore::put",
+        skip_all,
+        fields(key) // Will be recorded within the function.
+    )]
     pub async fn put<V: StoredObject>(
         &self,
         key: V::Key<'_>,
         value: &V,
     ) -> Result<String, ObjectStoreError> {
         let key = V::encode_key(key);
+        // Record the key for tracing.
+        tracing::Span::current().record("key", key.as_str());
         let bytes = value.serialize().map_err(ObjectStoreError::Serialization)?;
         self.put_raw(V::BUCKET, &key, bytes).await?;
         Ok(key)
@@ -170,8 +189,15 @@ impl dyn ObjectStore + '_ {
     /// # Errors
     ///
     /// Returns I/O errors specific to the storage.
+    #[tracing::instrument(
+        name = "ObjectStore::put",
+        skip_all,
+        fields(key) // Will be recorded within the function.
+    )]
     pub async fn remove<V: StoredObject>(&self, key: V::Key<'_>) -> Result<(), ObjectStoreError> {
         let key = V::encode_key(key);
+        // Record the key for tracing.
+        tracing::Span::current().record("key", key.as_str());
         self.remove_raw(V::BUCKET, &key).await
     }
 

core/lib/object_store/src/retries.rs

@@ -19,7 +19,11 @@ enum Request<'a> {
 }
 
 impl Request<'_> {
-    #[tracing::instrument(skip(f))] // output request and store as a part of structured logs
+    #[tracing::instrument(
+        name = "object_store::Request::retry",
+        skip(f), // output request and store as a part of structured logs
+        fields(retries) // Will be recorded before returning from the function
+    )]
     async fn retry<T, Fut, F>(
         self,
         store: &impl fmt::Debug,
@@ -32,13 +36,13 @@ impl Request<'_> {
     {
         let mut retries = 1;
         let mut backoff_secs = 1;
-        loop {
+        let result = loop {
             match f().await {
-                Ok(result) => return Ok(result),
+                Ok(result) => break Ok(result),
                 Err(err) if err.is_transient() => {
                     if retries > max_retries {
                         tracing::warn!(%err, "Exhausted {max_retries} retries performing request; returning last error");
-                        return Err(err);
+                        break Err(err);
                     }
                     tracing::info!(%err, "Failed request, retries: {retries}/{max_retries}");
                     retries += 1;
@@ -50,10 +54,12 @@ impl Request<'_> {
                 }
                 Err(err) => {
                     tracing::warn!(%err, "Failed request with a fatal error");
-                    return Err(err);
+                    break Err(err);
                 }
             }
-        }
+        };
+        tracing::Span::current().record("retries", retries);
+        result
     }
 }
 

core/node/commitment_generator/src/lib.rs

@@ -75,6 +75,7 @@ impl CommitmentGenerator {
         self.health_updater.subscribe()
     }
 
+    #[tracing::instrument(skip(self))]
     async fn calculate_aux_commitments(
         &self,
         l1_batch_number: L1BatchNumber,
@@ -145,6 +146,7 @@ impl CommitmentGenerator {
         })
     }
 
+    #[tracing::instrument(skip(self))]
     async fn prepare_input(
         &self,
         l1_batch_number: L1BatchNumber,
@@ -285,6 +287,7 @@ impl CommitmentGenerator {
         Ok(input)
     }
 
+    #[tracing::instrument(skip(self))]
     async fn process_batch(
         &self,
         l1_batch_number: L1BatchNumber,
@@ -307,6 +310,7 @@ impl CommitmentGenerator {
         Ok(artifacts)
     }
 
+    #[tracing::instrument(skip(self))]
     async fn step(
         &self,
         l1_batch_numbers: ops::RangeInclusive<L1BatchNumber>,
@@ -384,6 +388,7 @@ impl CommitmentGenerator {
         }
     }
 
+    #[tracing::instrument(skip(self))]
     async fn next_batch_range(&self) -> anyhow::Result<Option<ops::RangeInclusive<L1BatchNumber>>> {
         let mut connection = self
             .connection_pool

core/node/eth_sender/src/eth_tx_aggregator.rs

@@ -340,7 +340,7 @@ impl EthTxAggregator {
         Ok(vk_hash)
     }
 
-    #[tracing::instrument(skip(self, storage))]
+    #[tracing::instrument(skip_all, name = "EthTxAggregator::loop_iteration")]
     async fn loop_iteration(
         &mut self,
         storage: &mut Connection<'_, Core>,

core/node/eth_sender/src/eth_tx_manager.rs

@@ -630,7 +630,7 @@ impl EthTxManager {
         Ok(())
     }
 
-    #[tracing::instrument(skip(self, storage))]
+    #[tracing::instrument(skip_all, name = "EthTxManager::loop_iteration")]
     async fn loop_iteration(
         &mut self,
         storage: &mut Connection<'_, Core>,

core/node/eth_watch/src/lib.rs

@@ -93,6 +93,7 @@ impl EthWatch {
         })
     }
 
+    #[tracing::instrument(name = "EthWatch::initialize_state", skip_all)]
     async fn initialize_state(
         client: &dyn EthClient,
         storage: &mut Connection<'_, Core>,
@@ -166,7 +167,7 @@ impl EthWatch {
         Ok(())
     }
 
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(name = "EthWatch::loop_iteration", skip_all)]
     async fn loop_iteration(
         &mut self,
         storage: &mut Connection<'_, Core>,

core/node/house_keeper/src/periodic_job.rs

@@ -3,6 +3,7 @@ use std::time::Duration;
 use anyhow::Context;
 use async_trait::async_trait;
 use tokio::sync::watch;
+use tracing::Instrument;
 
 #[async_trait]
 pub trait PeriodicJob: Sync + Send {
@@ -20,8 +21,12 @@ pub trait PeriodicJob: Sync + Send {
             "Starting periodic job: {} with frequency: {timeout:?}",
             Self::SERVICE_NAME
         );
+
         while !*stop_receiver.borrow_and_update() {
             self.run_routine_task()
+                .instrument(
+                    tracing::info_span!("run_routine_task", service_name = %Self::SERVICE_NAME),
+                )
                 .await
                 .context("run_routine_task()")?;
             // Error here corresponds to a timeout w/o `stop_receiver` changed; we're OK with this.

core/node/metadata_calculator/src/updater.rs

@@ -196,6 +196,7 @@ impl TreeUpdater {
         Ok(())
     }
 
+    #[tracing::instrument(name = "TreeUpdater::step", skip(self, pool))]
     async fn step(
         &mut self,
         pool: &ConnectionPool<Core>,

core/node/proof_data_handler/src/request_processor.rs

@@ -45,6 +45,7 @@ impl RequestProcessor {
         }
     }
 
+    #[tracing::instrument(skip_all)]
     pub(crate) async fn get_proof_generation_data(
         &self,
         request: Json<ProofGenerationDataRequest>,
@@ -105,6 +106,7 @@ impl RequestProcessor {
     ///
     /// Expects all the data to be present in the database.
     /// Will panic if any of the required data is missing.
+    #[tracing::instrument(skip(self))]
     async fn proof_generation_data_for_existing_batch(
         &self,
         l1_batch_number: L1BatchNumber,

core/node/state_keeper/src/batch_executor/mod.rs

@@ -122,6 +122,7 @@ impl BatchExecutorHandle {
         }
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn execute_tx(&mut self, tx: Transaction) -> anyhow::Result<TxExecutionResult> {
         let tx_gas_limit = tx.gas_limit().as_u64();
 
@@ -161,6 +162,7 @@ impl BatchExecutorHandle {
         Ok(res)
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn start_next_l2_block(&mut self, env: L2BlockEnv) -> anyhow::Result<()> {
         // While we don't get anything from the channel, it's useful to have it as a confirmation that the operation
         // indeed has been processed.
@@ -184,6 +186,7 @@ impl BatchExecutorHandle {
         Ok(())
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn rollback_last_tx(&mut self) -> anyhow::Result<()> {
         // While we don't get anything from the channel, it's useful to have it as a confirmation that the operation
         // indeed has been processed.
@@ -207,6 +210,7 @@ impl BatchExecutorHandle {
         Ok(())
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn finish_batch(mut self) -> anyhow::Result<FinishedL1Batch> {
         let (response_sender, response_receiver) = oneshot::channel();
         let send_failed = self

core/node/state_keeper/src/io/output_handler.rs

@@ -64,6 +64,11 @@ impl OutputHandler {
         Ok(())
     }
 
+    #[tracing::instrument(
+        name = "OutputHandler::handle_l2_block"
+        skip_all,
+        fields(l2_block = %updates_manager.l2_block.number)
+    )]
     pub(crate) async fn handle_l2_block(
         &mut self,
         updates_manager: &UpdatesManager,
@@ -82,6 +87,11 @@ impl OutputHandler {
         Ok(())
     }
 
+    #[tracing::instrument(
+        name = "OutputHandler::handle_l1_batch"
+        skip_all,
+        fields(l1_batch = %updates_manager.l1_batch.number)
+    )]
     pub(crate) async fn handle_l1_batch(
         &mut self,
         updates_manager: Arc<UpdatesManager>,