Add tracing spans to validator client duty cycles (#8482)

Co-Authored-By: Jimmy Chen <jchen.tc@gmail.com>

Co-Authored-By: Jimmy Chen <jimmy@sigmaprime.io>

validator_client/validator_services/src/attestation_service.rs

@@ -8,7 +8,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 use task_executor::TaskExecutor;
 use tokio::time::{Duration, Instant, sleep, sleep_until};
-use tracing::{debug, error, info, trace, warn};
+use tracing::{Instrument, debug, error, info, info_span, instrument, trace, warn};
 use tree_hash::TreeHash;
 use types::{Attestation, AttestationData, ChainSpec, CommitteeIndex, EthSpec, Slot};
 use validator_store::{Error as ValidatorStoreError, ValidatorStore};
@@ -243,6 +243,11 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     ///
     /// The given `validator_duties` should already be filtered to only contain those that match
     /// `slot` and `committee_index`. Critical errors will be logged if this is not the case.
+    #[instrument(
+        name = "attestation_duty_cycle",
+        skip_all,
+        fields(%slot, %committee_index)
+    )]
     async fn publish_attestations_and_aggregates(
         self,
         slot: Slot,
@@ -328,6 +333,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     ///
     /// Only one `Attestation` is downloaded from the BN. It is then cloned and signed by each
     /// validator and the list of individually-signed `Attestation` objects is returned to the BN.
+    #[instrument(skip_all, fields(%slot, %committee_index))]
     async fn produce_and_publish_attestations(
         &self,
         slot: Slot,
@@ -357,6 +363,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
                     .map(|result| result.data)
             })
+            .instrument(info_span!("fetch_attestation_data"))
             .await
             .map_err(|e| e.to_string())?;
 
@@ -439,6 +446,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
 
         // Execute all the futures in parallel, collecting any successful results.
         let (ref attestations, ref validator_indices): (Vec<_>, Vec<_>) = join_all(signing_futures)
+            .instrument(info_span!(
+                "sign_attestations",
+                count = validator_duties.len()
+            ))
             .await
             .into_iter()
             .flatten()
@@ -487,6 +498,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     .post_beacon_pool_attestations_v2::<S::E>(single_attestations, fork_name)
                     .await
             })
+            .instrument(info_span!(
+                "publish_attestations",
+                count = attestations.len()
+            ))
             .await
         {
             Ok(()) => info!(
@@ -523,6 +538,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
     /// Only one aggregated `Attestation` is downloaded from the BN. It is then cloned and signed
     /// by each validator and the list of individually-signed `SignedAggregateAndProof` objects is
     /// returned to the BN.
+    #[instrument(skip_all, fields(slot = %attestation_data.slot, %committee_index))]
     async fn produce_and_publish_aggregates(
         &self,
         attestation_data: &AttestationData,
@@ -575,6 +591,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                         .map(|result| result.data)
                 }
             })
+            .instrument(info_span!("fetch_aggregate_attestation"))
             .await
             .map_err(|e| e.to_string())?;
 
@@ -617,7 +634,12 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
         });
 
         // Execute all the futures in parallel, collecting any successful results.
+        let aggregator_count = validator_duties
+            .iter()
+            .filter(|d| d.selection_proof.is_some())
+            .count();
         let signed_aggregate_and_proofs = join_all(signing_futures)
+            .instrument(info_span!("sign_aggregates", count = aggregator_count))
             .await
             .into_iter()
             .flatten()
@@ -647,6 +669,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             .await
                     }
                 })
+                .instrument(info_span!(
+                    "publish_aggregates",
+                    count = signed_aggregate_and_proofs.len()
+                ))
                 .await
             {
                 Ok(()) => {

validator_client/validator_services/src/block_service.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use task_executor::TaskExecutor;
 use tokio::sync::mpsc;
-use tracing::{debug, error, info, trace, warn};
+use tracing::{Instrument, debug, error, info, info_span, instrument, trace, warn};
 use types::{BlockType, ChainSpec, EthSpec, Graffiti, PublicKeyBytes, Slot};
 use validator_store::{Error as ValidatorStoreError, SignedBlock, UnsignedBlock, ValidatorStore};
 
@@ -320,6 +320,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
     }
 
     #[allow(clippy::too_many_arguments)]
+    #[instrument(skip_all, fields(%slot, ?validator_pubkey))]
     async fn sign_and_publish_block(
         &self,
         proposer_fallback: ProposerFallback<T>,
@@ -333,6 +334,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
         let res = self
             .validator_store
             .sign_block(*validator_pubkey, unsigned_block, slot)
+            .instrument(info_span!("sign_block"))
             .await;
 
         let signed_block = match res {
@@ -389,6 +391,11 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
         Ok(())
     }
 
+    #[instrument(
+        name = "block_proposal_duty_cycle",
+        skip_all,
+        fields(%slot, ?validator_pubkey)
+    )]
     async fn publish_block(
         self,
         slot: Slot,
@@ -483,6 +490,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     async fn publish_signed_block_contents(
         &self,
         signed_block: &SignedBlock<S::E>,
@@ -518,6 +526,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
         Ok::<_, BlockError>(())
     }
 
+    #[instrument(skip_all, fields(%slot))]
     async fn get_validator_block(
         beacon_node: &BeaconNodeHttpClient,
         slot: Slot,

validator_client/validator_services/src/sync_committee_service.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use task_executor::TaskExecutor;
 use tokio::time::{Duration, Instant, sleep, sleep_until};
-use tracing::{debug, error, info, trace, warn};
+use tracing::{Instrument, debug, error, info, info_span, instrument, trace, warn};
 use types::{
     ChainSpec, EthSpec, Hash256, PublicKeyBytes, Slot, SyncCommitteeSubscription,
     SyncContributionData, SyncDuty, SyncSelectionProof, SyncSubnetId,
@@ -208,7 +208,8 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                     .publish_sync_committee_signatures(slot, block_root, validator_duties)
                     .map(|_| ())
                     .await
-            },
+            }
+            .instrument(info_span!("sync_committee_signature_publish", %slot)),
             "sync_committee_signature_publish",
         );
 
@@ -225,7 +226,8 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                     )
                     .map(|_| ())
                     .await
-            },
+            }
+            .instrument(info_span!("sync_committee_aggregate_publish", %slot)),
             "sync_committee_aggregate_publish",
         );
 
@@ -233,6 +235,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
     }
 
     /// Publish sync committee signatures.
+    #[instrument(skip_all, fields(%slot, ?beacon_block_root))]
     async fn publish_sync_committee_signatures(
         &self,
         slot: Slot,
@@ -277,6 +280,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
 
         // Execute all the futures in parallel, collecting any successful results.
         let committee_signatures = &join_all(signature_futures)
+            .instrument(info_span!(
+                "sign_sync_signatures",
+                count = validator_duties.len()
+            ))
             .await
             .into_iter()
             .flatten()
@@ -288,6 +295,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                     .post_beacon_pool_sync_committee_signatures(committee_signatures)
                     .await
             })
+            .instrument(info_span!(
+                "publish_sync_signatures",
+                count = committee_signatures.len()
+            ))
             .await
             .map_err(|e| {
                 error!(
@@ -328,7 +339,8 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                         )
                         .map(|_| ())
                         .await
-                },
+                }
+                .instrument(info_span!("publish_sync_committee_aggregate_for_subnet", %slot, ?beacon_block_root, %subnet_id)),
                 "sync_committee_aggregate_publish_subnet",
             );
         }
@@ -357,6 +369,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                     .get_validator_sync_committee_contribution(&sync_contribution_data)
                     .await
             })
+            .instrument(info_span!("fetch_sync_contribution"))
             .await
             .map_err(|e| {
                 crit!(
@@ -372,6 +385,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
             .data;
 
         // Create futures to produce signed contributions.
+        let aggregator_count = subnet_aggregators.len();
         let signature_futures = subnet_aggregators.into_iter().map(
             |(aggregator_index, aggregator_pk, selection_proof)| async move {
                 match self
@@ -405,6 +419,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
 
         // Execute all the futures in parallel, collecting any successful results.
         let signed_contributions = &join_all(signature_futures)
+            .instrument(info_span!(
+                "sign_sync_contributions",
+                count = aggregator_count
+            ))
             .await
             .into_iter()
             .flatten()
@@ -417,6 +435,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> SyncCommitteeService<S
                     .post_validator_contribution_and_proofs(signed_contributions)
                     .await
             })
+            .instrument(info_span!(
+                "publish_sync_contributions",
+                count = signed_contributions.len()
+            ))
             .await
             .map_err(|e| {
                 error!(