Report new received crds signatures and their respective origins to metrics (#32504)

* screwed up old branch and syncing with upstream branch

* add fixed size ring buff instead of variable sized vecdeque for reporting signatures

* modify difficulty to take in n 0 bits and check if signature ending ends in n 0 bits

* update to only push every 18 trailing zero bits. and clean up

* report origin with signature. and set trailing 0s to 8 for local testing

* change back to 18 trailing zeros and rm unused imports

* run cargo rmt

* run ./scripts/cargo-for-all-lock-files.sh tree

* allow integer arithmetic for bit comparison

* rm unused lifetime

* rm duplicate entry?

* re implement ring buf

* put ringbuf in sorted order

* ringbuf in cargo.toml now in sorted order

* rm ring buf, refactor, fix trailing zero bug

* fix bug in trailing zeros. was comparing wrong ones

* fix needless range loop bug

* fix bug

* change trailing zero checking to build in methods and only report first 8 bytes of signature and origin pubkey

* report full origin string and first 8 bytes of signature

* set SIGNATURE_SAMPLE_TRAILING_ZEROS back to 18

* forgot to run cargo tree

* avoid panic and change working

* rm bs58

* pass in Option<String> into datapoint_info

* shorten metric names

gossip/src/crds.rs

@@ -44,6 +44,7 @@ use {
         clock::Slot,
         hash::{hash, Hash},
         pubkey::Pubkey,
+        signature::Signature,
     },
     std::{
         cmp::Ordering,
@@ -56,6 +57,13 @@ 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;
+// Required number of leading 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 signature reported per minute
+// log2(500k) = ~18.9.
+const SIGNATURE_SAMPLE_LEADING_ZEROS: u32 = 19;
 
 pub struct Crds {
     /// Stores the map of labels and values
@@ -669,6 +677,22 @@ impl CrdsDataStats {
                 self.votes.put(slot, num_nodes + 1);
             }
         }
+
+        if should_report_message_signature(&entry.value.signature) {
+            datapoint_info!(
+                "gossip_crds_sample",
+                (
+                    "origin",
+                    entry.value.pubkey().to_string().get(..8),
+                    Option<String>
+                ),
+                (
+                    "signature",
+                    entry.value.signature.to_string().get(..8),
+                    Option<String>
+                )
+            );
+        }
     }
 
     fn record_fail(&mut self, entry: &VersionedCrdsValue) {
@@ -714,6 +738,15 @@ impl CrdsStats {
     }
 }
 
+/// check if first SIGNATURE_SAMPLE_LEADING_ZEROS bits of signature are 0
+#[inline]
+fn should_report_message_signature(signature: &Signature) -> bool {
+    let Some(Ok(bytes)) = signature.as_ref().get(..8).map(<[u8; 8]>::try_from) else {
+        return false;
+    };
+    u64::from_le_bytes(bytes).trailing_zeros() >= SIGNATURE_SAMPLE_LEADING_ZEROS
+}
+
 #[cfg(test)]
 mod tests {
     use {