Report new received crds signatures and their respective origins to metrics

Cargo.toml

@@ -270,6 +270,7 @@ rayon = "1.7.0"
 rcgen = "0.10.0"
 reed-solomon-erasure = "6.0.0"
 regex = "1.9.1"
+ringbuf = "0.3.3"
 rolling-file = "0.2.0"
 reqwest = { version = "0.11.17", default-features = false }
 rpassword = "7.2"

gossip/Cargo.toml

@@ -24,6 +24,7 @@ num-traits = { workspace = true }
 rand = { workspace = true }
 rand_chacha = { workspace = true }
 rayon = { workspace = true }
+ringbuf = { workspace = true }
 serde = { workspace = true }
 serde_bytes = { workspace = true }
 serde_derive = { workspace = true }

gossip/src/cluster_info_metrics.rs

@@ -1,8 +1,9 @@
 use {
     crate::crds_gossip::CrdsGossip,
     itertools::Itertools,
+    ringbuf::{HeapRb, Rb},
     solana_measure::measure::Measure,
-    solana_sdk::{clock::Slot, pubkey::Pubkey},
+    solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Signature},
     std::{
         cmp::Reverse,
         collections::HashMap,
@@ -189,7 +190,14 @@ pub(crate) fn submit_gossip_stats(
     gossip: &CrdsGossip,
     stakes: &HashMap<Pubkey, u64>,
 ) {
-    let (crds_stats, table_size, num_nodes, num_pubkeys, purged_values_size, failed_inserts_size) = {
+    let (
+        mut crds_stats,
+        table_size,
+        num_nodes,
+        num_pubkeys,
+        purged_values_size,
+        failed_inserts_size,
+    ) = {
         let gossip_crds = gossip.crds.read().unwrap();
         (
             gossip_crds.take_stats(),
@@ -675,6 +683,12 @@ pub(crate) fn submit_gossip_stats(
         ("all-push", crds_stats.push.fails.iter().sum::<usize>(), i64),
         ("all-pull", crds_stats.pull.fails.iter().sum::<usize>(), i64),
     );
+
+    submit_crds_signature_stats(
+        "cluster_info_crds_stats_message_signatures_received",
+        &mut crds_stats.push.crds_signatures,
+    );
+
     if !log::log_enabled!(log::Level::Trace) {
         return;
     }
@@ -704,3 +718,17 @@ where
         datapoint_trace!(name, ("slot", slot, i64), ("num_votes", num_votes, i64));
     }
 }
+
+fn submit_crds_signature_stats(
+    name: &'static str,
+    crds_signatures: &mut HeapRb<(Pubkey, Signature)>,
+) {
+    // submit all message signatures to metrics.
+    while let Some((origin, signature)) = crds_signatures.pop() {
+        datapoint_info!(
+            name,
+            ("crds_origin", origin.to_string(), String),
+            ("crds_signature", signature.to_string(), String)
+        );
+    }
+}

gossip/src/crds.rs

@@ -40,10 +40,12 @@ use {
     lru::LruCache,
     matches::debug_assert_matches,
     rayon::{prelude::*, ThreadPool},
+    ringbuf::{HeapRb, Rb},
     solana_sdk::{
         clock::Slot,
         hash::{hash, Hash},
         pubkey::Pubkey,
+        signature::Signature,
     },
     std::{
         cmp::Ordering,
@@ -56,6 +58,15 @@ use {
 const CRDS_SHARDS_BITS: u32 = 12;
 // Number of vote slots to track in an lru-cache for metrics.
 const VOTE_SLOTS_METRICS_CAP: usize = 100;
+// Number of ending zero bits for crds signature to get reported to influx
+// mean new push messages received per minute per node
+//      testnet: ~500k,
+//      mainnet: ~280k
+// target: 1-2 signatures reported per minute
+// log2(250k) = ~17.9.
+const TRAILING_ZEROS: u8 = 18;
+// size of ring buffer to store signatures for metrics
+const SIGNATURE_SLOTS_METRICS_CAP: usize = 64;
 
 pub struct Crds {
     /// Stores the map of labels and values
@@ -101,6 +112,7 @@ pub(crate) struct CrdsDataStats {
     pub(crate) counts: CrdsCountsArray,
     pub(crate) fails: CrdsCountsArray,
     pub(crate) votes: LruCache<Slot, /*count:*/ usize>,
+    pub(crate) crds_signatures: HeapRb<(Pubkey, Signature)>, // crds origin and signature
 }
 
 #[derive(Default)]
@@ -656,6 +668,7 @@ impl Default for CrdsDataStats {
             counts: CrdsCountsArray::default(),
             fails: CrdsCountsArray::default(),
             votes: LruCache::new(VOTE_SLOTS_METRICS_CAP),
+            crds_signatures: HeapRb::<(Pubkey, Signature)>::new(SIGNATURE_SLOTS_METRICS_CAP),
         }
     }
 }
@@ -669,6 +682,12 @@ impl CrdsDataStats {
                 self.votes.put(slot, num_nodes + 1);
             }
         }
+
+        // check trailing zero bits on signature for metrics
+        if entry.value.signature.last_n_bits_are_zero(TRAILING_ZEROS) {
+            self.crds_signatures
+                .push_overwrite((entry.value.pubkey(), entry.value.signature));
+        }
     }
 
     fn record_fail(&mut self, entry: &VersionedCrdsValue) {

sdk/src/signature.rs

@@ -1,5 +1,6 @@
 //! Functionality for public and private keys.
 #![cfg(feature = "full")]
+#![allow(clippy::integer_arithmetic)]
 
 // legacy module paths
 pub use crate::signer::{keypair::*, null_signer::*, presigner::*, *};
@@ -54,6 +55,21 @@ impl Signature {
     pub fn verify(&self, pubkey_bytes: &[u8], message_bytes: &[u8]) -> bool {
         self.verify_verbose(pubkey_bytes, message_bytes).is_ok()
     }
+
+    pub fn last_n_bits_are_zero(&self, n: u8) -> bool {
+        let signature_bytes: &[u8] = self.0.as_slice();
+        let num_bytes = (n as usize + 7) / 8; // Number of bytes required to represent n bits
+        let last_signature_byte_index = signature_bytes.len() - 1;
+
+        let mut signature_ending: u64 = 0;
+        for i in 0..num_bytes {
+            let shift = (num_bytes - i - 1) * 8;
+            signature_ending |= (signature_bytes[last_signature_byte_index - i] as u64) << shift;
+        }
+
+        // check if last n bits of signature (aka signature_ending) are all 0.
+        (signature_ending & ((1 << n) - 1)) == 0
+    }
 }
 
 pub trait Signable {