Add more instrumentation

Author: ndr-ds

Cargo.lock

@@ -35,6 +35,7 @@ use linera_views::{
     views::{ClonableView, CryptoHashView, RootView, View},
 };
 use serde::{Deserialize, Serialize};
+use tracing::instrument;
 
 use crate::{
     block::{Block, ConfirmedBlock},
@@ -309,6 +310,12 @@ pub struct ChainTipState {
 impl ChainTipState {
     /// Checks that the proposed block is suitable, i.e. at the expected height and with the
     /// expected parent.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        next_block_height = %self.next_block_height,
+        block_hash = ?self.block_hash,
+        proposed_height = %new_block.height,
+        proposed_previous_hash = ?new_block.previous_block_hash
+    ))]
     pub fn verify_block_chaining(&self, new_block: &ProposedBlock) -> Result<(), ChainError> {
         ensure!(
             new_block.height == self.next_block_height,
@@ -326,6 +333,10 @@ impl ChainTipState {
 
     /// Returns `true` if the validated block's height is below the tip height. Returns an error if
     /// it is higher than the tip.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        next_block_height = %self.next_block_height,
+        height = %height
+    ))]
     pub fn already_validated_block(&self, height: BlockHeight) -> Result<bool, ChainError> {
         ensure!(
             self.next_block_height >= height,
@@ -337,6 +348,10 @@ impl ChainTipState {
     }
 
     /// Checks if the measurement counters would be valid.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        next_block_height = %self.next_block_height,
+        num_transactions = %transactions.len()
+    ))]
     pub fn update_counters(
         &mut self,
         transactions: &[Transaction],
@@ -391,6 +406,10 @@ where
         self.context().extra().chain_id()
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        next_block_height = %self.tip_state.get().next_block_height
+    ))]
     pub async fn query_application(
         &mut self,
         local_time: Timestamp,
@@ -408,6 +427,10 @@ where
             .with_execution_context(ChainExecutionContext::Query)
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        application_id = %application_id
+    ))]
     pub async fn describe_application(
         &mut self,
         application_id: ApplicationId,
@@ -419,6 +442,11 @@ where
             .with_execution_context(ChainExecutionContext::DescribeApplication)
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        target = %target,
+        height = %height
+    ))]
     pub async fn mark_messages_as_received(
         &mut self,
         target: &ChainId,
@@ -459,6 +487,10 @@ where
 
     /// Returns true if there are no more outgoing messages in flight up to the given
     /// block height.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        height = %height
+    ))]
     pub fn all_messages_delivered_up_to(&self, height: BlockHeight) -> bool {
         tracing::debug!(
             "Messages left in {:.8}'s outbox: {:?}",
@@ -478,6 +510,10 @@ where
     }
 
     /// Invariant for the states of active chains.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        local_time = %local_time
+    ))]
     pub async fn ensure_is_active(&mut self, local_time: Timestamp) -> Result<(), ChainError> {
         // Initialize ourselves.
         if self
@@ -506,6 +542,9 @@ where
 
     /// Verifies that this chain is up-to-date and all the messages executed ahead of time
     /// have been properly received by now.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id()
+    ))]
     pub async fn validate_incoming_bundles(&self) -> Result<(), ChainError> {
         let chain_id = self.chain_id();
         let pairs = self.inboxes.try_load_all_entries().await?;
@@ -526,6 +565,10 @@ where
         Ok(())
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        origin = %origin
+    ))]
     pub async fn next_block_height_to_receive(
         &self,
         origin: &ChainId,
@@ -540,13 +583,21 @@ where
     /// Returns the height of the highest block we have, plus one. Includes preprocessed blocks.
     ///
     /// The "+ 1" is so that it can be used in the same places as `next_block_height`.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        next_block_height = %self.tip_state.get().next_block_height
+    ))]
     pub async fn next_height_to_preprocess(&self) -> Result<BlockHeight, ChainError> {
         if let Some(height) = self.preprocessed_blocks.indices().await?.last() {
             return Ok(height.saturating_add(BlockHeight(1)));
         }
         Ok(self.tip_state.get().next_block_height)
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        origin = %origin
+    ))]
     pub async fn last_anticipated_block_height(
         &self,
         origin: &ChainId,
@@ -567,6 +618,11 @@ where
     /// round timeouts.
     ///
     /// Returns `true` if incoming `Subscribe` messages created new outbox entries.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        origin = %origin,
+        bundle_height = %bundle.height
+    ))]
     pub async fn receive_message_bundle(
         &mut self,
         origin: &ChainId,
@@ -630,6 +686,10 @@ where
     }
 
     /// Updates the `received_log` trackers.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        num_trackers = %new_trackers.len()
+    ))]
     pub fn update_received_certificate_trackers(
         &mut self,
         new_trackers: BTreeMap<ValidatorPublicKey, u64>,
@@ -649,6 +709,9 @@ where
         }
     }
 
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id()
+    ))]
     pub fn current_committee(&self) -> Result<(Epoch, &Committee), ChainError> {
         self.execution_state
             .system
@@ -661,6 +724,10 @@ where
     }
 
     /// Removes the incoming message bundles in the block from the inboxes.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        timestamp = %timestamp
+    ))]
     pub async fn remove_bundles_from_inboxes(
         &mut self,
         timestamp: Timestamp,
@@ -739,6 +806,10 @@ where
     }
 
     /// Returns the outboxes for the given targets, or an error if any of them are missing.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        num_targets = %targets.len()
+    ))]
     pub async fn load_outboxes(
         &self,
         targets: &[ChainId],
@@ -750,6 +821,10 @@ where
 
     /// Executes a block: first the incoming messages, then the main operation.
     /// Does not update chain state other than the execution state.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %block.chain_id,
+        block_height = %block.height
+    ))]
     #[expect(clippy::too_many_arguments)]
     async fn execute_block_inner(
         chain: &mut ExecutionStateView<C>,
@@ -859,6 +934,10 @@ where
 
     /// Executes a block: first the incoming messages, then the main operation.
     /// Does not update chain state other than the execution state.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        block_height = %block.height
+    ))]
     pub async fn execute_block(
         &mut self,
         block: &ProposedBlock,
@@ -919,6 +998,10 @@ where
     /// Applies an execution outcome to the chain, updating the outboxes, state hash and chain
     /// manager. This does not touch the execution state itself, which must be updated separately.
     /// Returns the set of event streams that were updated as a result of applying the block.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        block_height = %block.inner().inner().header.height
+    ))]
     pub async fn apply_confirmed_block(
         &mut self,
         block: &ConfirmedBlock,
@@ -953,6 +1036,10 @@ where
 
     /// Adds a block to `preprocessed_blocks`, and updates the outboxes where possible.
     /// Returns the set of streams that were updated as a result of preprocessing the block.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        block_height = %block.inner().inner().header.height
+    ))]
     pub async fn preprocess_block(
         &mut self,
         block: &ConfirmedBlock,
@@ -979,6 +1066,10 @@ where
     }
 
     /// Verifies that the block is valid according to the chain's application permission settings.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        block_height = %block.height,
+        num_transactions = %block.transactions.len()
+    ))]
     fn check_app_permissions(
         app_permissions: &ApplicationPermissions,
         block: &ProposedBlock,
@@ -1021,6 +1112,10 @@ where
     }
 
     /// Returns the hashes of all blocks we have in the given range.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        next_block_height = %self.tip_state.get().next_block_height
+    ))]
     pub async fn block_hashes(
         &self,
         range: impl RangeBounds<BlockHeight>,
@@ -1066,6 +1161,10 @@ where
     /// Updates the outboxes with the messages sent in the block.
     ///
     /// Returns the set of all recipients.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        block_height = %block.header.height
+    ))]
     async fn process_outgoing_messages(
         &mut self,
         block: &Block,
@@ -1157,6 +1256,10 @@ where
     /// Updates the event streams with events emitted by the block if they form a contiguous
     /// sequence (might not be the case when preprocessing a block).
     /// Returns the set of updated event streams.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        block_height = %block.header.height
+    ))]
     async fn process_emitted_events(
         &mut self,
         block: &Block,

examples/Cargo.lock

@@ -176,6 +176,7 @@ where
 {
     /// Runs the [`ChainWorkerActor`]. The chain state is loaded when the first request
     /// arrives.
+    #[instrument(target = "telemetry_only", skip_all, fields(chain_id = %chain_id))]
     #[expect(clippy::too_many_arguments)]
     pub(crate) async fn run(
         config: ChainWorkerConfig,
@@ -207,6 +208,7 @@ where
     /// Spawns a blocking task to execute the service runtime actor.
     ///
     /// Returns the task handle and the endpoints to interact with the actor.
+    #[instrument(target = "telemetry_only", skip_all, fields(chain_id = %chain_id))]
     async fn spawn_service_runtime_actor(
         chain_id: ChainId,
     ) -> (linera_base::task::Blocking, ServiceRuntimeEndpoint) {
@@ -243,8 +245,9 @@ where
 
     /// Runs the worker until there are no more incoming requests.
     #[instrument(
+        target = "telemetry_only",
         skip_all,
-        fields(chain_id = format!("{:.8}", self.chain_id)),
+        fields(chain_id = %self.chain_id, long_lived_services = %self.config.long_lived_services),
     )]
     async fn handle_requests(
         self,
@@ -256,30 +259,37 @@ where
         trace!("Starting `ChainWorkerActor`");
 
         while let Some((request, span)) = incoming_requests.recv().await {
-            let (service_runtime_thread, service_runtime_endpoint) = {
-                if self.config.long_lived_services {
-                    let (thread, endpoint) = Self::spawn_service_runtime_actor(self.chain_id).await;
-                    (Some(thread), Some(endpoint))
-                } else {
-                    (None, None)
-                }
-            };
-
-            trace!("Loading chain state of {}", self.chain_id);
-            let mut worker = ChainWorkerState::load(
-                self.config.clone(),
-                self.storage.clone(),
-                self.block_values.clone(),
-                self.execution_state_cache.clone(),
-                self.tracked_chains.clone(),
-                self.delivery_notifier.clone(),
-                self.chain_id,
-                service_runtime_endpoint,
-            )
+            let (service_runtime_thread, mut worker) = async {
+                let (service_runtime_thread, service_runtime_endpoint) = {
+                    if self.config.long_lived_services {
+                        let (thread, endpoint) =
+                            Self::spawn_service_runtime_actor(self.chain_id).await;
+                        (Some(thread), Some(endpoint))
+                    } else {
+                        (None, None)
+                    }
+                };
+
+                trace!("Loading chain state of {}", self.chain_id);
+                let mut worker = ChainWorkerState::load(
+                    self.config.clone(),
+                    self.storage.clone(),
+                    self.block_values.clone(),
+                    self.execution_state_cache.clone(),
+                    self.tracked_chains.clone(),
+                    self.delivery_notifier.clone(),
+                    self.chain_id,
+                    service_runtime_endpoint,
+                )
+                .await?;
+
+                worker.handle_request(request).await;
+
+                Ok::<_, WorkerError>((service_runtime_thread, worker))
+            }
+            .instrument(span)
             .await?;
 
-            Box::pin(worker.handle_request(request).instrument(span)).await;
-
             loop {
                 futures::select! {
                     () = self.sleep_until_timeout().fuse() => break,