Add CertificatepoolServiceStats. (anza-xyz#307)

Co-authored-by: Ashwin Sekar <ashwin@anza.xyz>

Author: 2 people

votor/src/certificate_pool_service.rs

@@ -1,6 +1,8 @@
 //! Service in charge of ingesting new messages into the certificate pool
 //! and notifying votor of new events that occur
 
+mod stats;
+
 use {
     crate::{
         certificate_pool::{
@@ -23,6 +25,7 @@ use {
         root_bank_cache::RootBankCache, vote_sender_types::BLSVerifiedMessageReceiver,
     },
     solana_sdk::clock::Slot,
+    stats::CertificatePoolServiceStats,
     std::{
         sync::{
             atomic::{AtomicBool, Ordering},
@@ -85,32 +88,39 @@ impl CertificatePoolService {
         current_root: &mut Slot,
         highest_finalized_slot: &mut Slot,
         standstill_timer: &mut Instant,
+        stats: &mut CertificatePoolServiceStats,
     ) -> Result<(), AddVoteError> {
         // If we have a new finalized slot, update the root and send new certificates
         if let Some(new_finalized_slot) = new_finalized_slot {
             // Reset standstill timer
             debug_assert!(new_finalized_slot > *highest_finalized_slot);
             *highest_finalized_slot = new_finalized_slot;
             *standstill_timer = Instant::now();
+            CertificatePoolServiceStats::incr_u16(&mut stats.new_finalized_slot);
             // Set root
             let root_bank = root_bank_cache.root_bank();
             if root_bank.slot() > *current_root {
+                CertificatePoolServiceStats::incr_u16(&mut stats.new_root);
                 *current_root = root_bank.slot();
                 cert_pool.handle_new_root(root_bank);
             }
         }
         // Send new certificates to peers
-        for certificate in new_certificates_to_send {
+        for (i, certificate) in new_certificates_to_send.iter().enumerate() {
             // The buffer should normally be large enough, so we don't handle
             // certificate re-send here.
             match bls_sender.try_send(BLSOp::PushCertificate {
                 certificate: certificate.clone(),
             }) {
-                Ok(_) => (),
+                Ok(_) => {
+                    CertificatePoolServiceStats::incr_u16(&mut stats.certificates_sent);
+                }
                 Err(TrySendError::Disconnected(_)) => {
                     return Err(AddVoteError::VotingServiceSenderDisconnected);
                 }
                 Err(TrySendError::Full(_)) => {
+                    let dropped = new_certificates_to_send.len().saturating_sub(i) as u16;
+                    stats.certificates_dropped = stats.certificates_dropped.saturating_add(dropped);
                     return Err(AddVoteError::VotingServiceQueueFull);
                 }
             }
@@ -126,7 +136,16 @@ impl CertificatePoolService {
         current_root: &mut Slot,
         highest_finalized_slot: &mut Slot,
         standstill_timer: &mut Instant,
+        stats: &mut CertificatePoolServiceStats,
     ) -> Result<(), AddVoteError> {
+        match message {
+            BLSMessage::Certificate(_) => {
+                CertificatePoolServiceStats::incr_u32(&mut stats.received_certificates);
+            }
+            BLSMessage::Vote(_) => {
+                CertificatePoolServiceStats::incr_u32(&mut stats.received_votes);
+            }
+        }
         match voting_utils::add_message_and_maybe_update_commitment(
             &ctx.my_pubkey,
             &ctx.my_vote_pubkey,
@@ -145,6 +164,7 @@ impl CertificatePoolService {
                     current_root,
                     highest_finalized_slot,
                     standstill_timer,
+                    stats,
                 )?;
             }
             Err(e) => {
@@ -153,6 +173,7 @@ impl CertificatePoolService {
                 } else {
                     // This is a non critical error, a duplicate vote for example
                     trace!("{}: unable to push vote into pool {}", &ctx.my_pubkey, e);
+                    CertificatePoolServiceStats::incr_u32(&mut stats.add_message_failed);
                 }
             }
         };
@@ -175,6 +196,7 @@ impl CertificatePoolService {
         Votor::wait_for_migration_or_exit(&ctx.exit, &ctx.start);
         info!("{}: Certificate pool loop starting", &ctx.my_pubkey);
         let mut current_root = ctx.root_bank_cache.root_bank().slot();
+        let mut stats = CertificatePoolServiceStats::new();
 
         // Standstill tracking
         let mut standstill_timer = Instant::now();
@@ -201,10 +223,12 @@ impl CertificatePoolService {
                 &cert_pool,
                 &mut ctx,
                 &mut events,
+                &mut stats,
             );
 
             if standstill_timer.elapsed() > STANDSTILL_TIMEOUT {
                 events.push(VotorEvent::Standstill(highest_finalized_slot));
+                stats.standstill = true;
                 standstill_timer = Instant::now();
             }
 
@@ -241,6 +265,7 @@ impl CertificatePoolService {
                     &mut current_root,
                     &mut highest_finalized_slot,
                     &mut standstill_timer,
+                    &mut stats,
                 ) {
                     info!(
                         "{}: Unable to process BLS message: {e}. Exiting.",
@@ -250,6 +275,7 @@ impl CertificatePoolService {
                     return Ok(());
                 }
             }
+            stats.maybe_report();
         }
         Ok(())
     }
@@ -259,6 +285,7 @@ impl CertificatePoolService {
         cert_pool: &CertificatePool,
         ctx: &mut CertificatePoolContext,
         events: &mut Vec<VotorEvent>,
+        stats: &mut CertificatePoolServiceStats,
     ) {
         let Some(new_highest_parent_ready) = events
             .iter()
@@ -304,6 +331,7 @@ impl CertificatePoolService {
                     skipping production of {start_slot}-{end_slot}",
                     ctx.my_pubkey,
                 );
+                CertificatePoolServiceStats::incr_u16(&mut stats.parent_ready_missed_window);
             }
             BlockProductionParent::ParentNotReady => {
                 // This can't happen, place holder depending on how we hook up optimistic
@@ -320,7 +348,8 @@ impl CertificatePoolService {
                     parent_block,
                     // TODO: we can just remove this
                     skip_timer: Instant::now(),
-                }))
+                }));
+                CertificatePoolServiceStats::incr_u16(&mut stats.parent_ready_produce_window);
             }
         }
     }

votor/src/certificate_pool_service/stats.rs

@@ -0,0 +1,92 @@
+use {
+    solana_metrics::datapoint_info,
+    std::time::{Duration, Instant},
+};
+
+const STATS_REPORT_INTERVAL: Duration = Duration::from_secs(10);
+
+#[derive(Debug)]
+pub(crate) struct CertificatePoolServiceStats {
+    pub(crate) add_message_failed: u32,
+    pub(crate) certificates_sent: u16,
+    pub(crate) certificates_dropped: u16,
+    pub(crate) new_finalized_slot: u16,
+    pub(crate) new_root: u16,
+    pub(crate) parent_ready_missed_window: u16,
+    pub(crate) parent_ready_produce_window: u16,
+    pub(crate) received_votes: u32,
+    pub(crate) received_certificates: u32,
+    pub(crate) standstill: bool,
+    last_request_time: Instant,
+}
+
+impl CertificatePoolServiceStats {
+    pub fn new() -> Self {
+        Self {
+            add_message_failed: 0,
+            certificates_sent: 0,
+            certificates_dropped: 0,
+            new_finalized_slot: 0,
+            new_root: 0,
+            parent_ready_missed_window: 0,
+            parent_ready_produce_window: 0,
+            received_votes: 0,
+            received_certificates: 0,
+            standstill: false,
+            last_request_time: Instant::now(),
+        }
+    }
+
+    pub fn incr_u16(value: &mut u16) {
+        *value = value.saturating_add(1);
+    }
+
+    pub fn incr_u32(value: &mut u32) {
+        *value = value.saturating_add(1);
+    }
+
+    fn reset(&mut self) {
+        self.add_message_failed = 0;
+        self.certificates_sent = 0;
+        self.certificates_dropped = 0;
+        self.new_finalized_slot = 0;
+        self.new_root = 0;
+        self.parent_ready_missed_window = 0;
+        self.parent_ready_produce_window = 0;
+        self.received_votes = 0;
+        self.received_certificates = 0;
+        self.standstill = false;
+        self.last_request_time = Instant::now();
+    }
+
+    fn report(&self) {
+        datapoint_info!(
+            "cert_pool_service",
+            ("add_message_failed", self.add_message_failed, i64),
+            ("certificates_sent", self.certificates_sent, i64),
+            ("certificates_dropped", self.certificates_dropped, i64),
+            ("new_root", self.new_root, i64),
+            ("new_finalized_slot", self.new_finalized_slot, i64),
+            (
+                "parent_ready_missed_window",
+                self.parent_ready_missed_window,
+                i64
+            ),
+            (
+                "parent_ready_produce_window",
+                self.parent_ready_produce_window,
+                i64
+            ),
+            ("received_votes", self.received_votes, i64),
+            ("received_certificates", self.received_certificates, i64),
+            ("standstill", self.standstill, i64),
+        );
+    }
+
+    pub fn maybe_report(&mut self) {
+        if self.last_request_time.elapsed() >= STATS_REPORT_INTERVAL {
+            self.report();
+            self.reset();
+        }
+    }
+}