Rate limit transaction counters (#5447)

* Rate limit transaction counters

* @sakridge feedback

* Set default high metrics rate for multinode demo

* Fix tests

* Swap defaults and fix env var tests

* Only set metrics rate if not already set

core/src/banking_stage.rs

@@ -228,10 +228,13 @@ impl BankingStage {
             (new_tx_count as f32) / (proc_start.as_s())
         );
 
-        inc_new_counter_info!("banking_stage-rebuffered_packets", rebuffered_packets);
-        inc_new_counter_info!("banking_stage-consumed_buffered_packets", new_tx_count);
-        inc_new_counter_debug!("banking_stage-process_transactions", new_tx_count);
-        inc_new_counter_debug!("banking_stage-dropped_batches_count", dropped_batches_count);
+        inc_new_high_rate_counter_info!("banking_stage-rebuffered_packets", rebuffered_packets);
+        inc_new_high_rate_counter_info!("banking_stage-consumed_buffered_packets", new_tx_count);
+        inc_new_high_rate_counter_debug!("banking_stage-process_transactions", new_tx_count);
+        inc_new_high_rate_counter_debug!(
+            "banking_stage-dropped_batches_count",
+            dropped_batches_count
+        );
 
         Ok(unprocessed_packets)
     }
@@ -810,7 +813,7 @@ impl BankingStage {
             count,
             id,
         );
-        inc_new_counter_debug!("banking_stage-transactions_received", count);
+        inc_new_high_rate_counter_debug!("banking_stage-transactions_received", count);
         let mut proc_start = Measure::start("process_received_packets_process");
         let mut new_tx_count = 0;
 
@@ -882,9 +885,12 @@ impl BankingStage {
             count,
             id,
         );
-        inc_new_counter_debug!("banking_stage-process_packets", count);
-        inc_new_counter_debug!("banking_stage-process_transactions", new_tx_count);
-        inc_new_counter_debug!("banking_stage-dropped_batches_count", dropped_batches_count);
+        inc_new_high_rate_counter_debug!("banking_stage-process_packets", count);
+        inc_new_high_rate_counter_debug!("banking_stage-process_transactions", new_tx_count);
+        inc_new_high_rate_counter_debug!(
+            "banking_stage-dropped_batches_count",
+            dropped_batches_count
+        );
 
         *recv_start = Instant::now();
 

core/src/cluster_info.rs

@@ -1155,7 +1155,7 @@ impl ClusterInfo {
         stakes: &HashMap<Pubkey, u64>,
     ) -> Vec<SharedBlob> {
         let self_id = me.read().unwrap().gossip.id;
-        inc_new_counter_debug!("cluster_info-push_message", 1, 0, 1000);
+        inc_new_high_rate_counter_debug!("cluster_info-push_message", 1);
 
         let updated: Vec<_> =
             me.write()

core/src/poh_recorder.rs

@@ -296,11 +296,9 @@ impl PohRecorder {
     pub fn tick(&mut self) {
         let now = Instant::now();
         let poh_entry = self.poh.lock().unwrap().tick();
-        inc_new_counter_warn!(
+        inc_new_high_rate_counter_warn!(
             "poh_recorder-tick_lock_contention",
-            timing::duration_as_ms(&now.elapsed()) as usize,
-            0,
-            1000
+            timing::duration_as_ms(&now.elapsed()) as usize
         );
         let now = Instant::now();
         if let Some(poh_entry) = poh_entry {
@@ -308,11 +306,9 @@ impl PohRecorder {
             trace!("tick {}", self.tick_height);
 
             if self.start_leader_at_tick.is_none() {
-                inc_new_counter_warn!(
+                inc_new_high_rate_counter_warn!(
                     "poh_recorder-tick_overhead",
-                    timing::duration_as_ms(&now.elapsed()) as usize,
-                    0,
-                    1000
+                    timing::duration_as_ms(&now.elapsed()) as usize
                 );
                 return;
             }
@@ -326,11 +322,9 @@ impl PohRecorder {
             self.tick_cache.push((entry, self.tick_height));
             let _ = self.flush_cache(true);
         }
-        inc_new_counter_warn!(
+        inc_new_high_rate_counter_warn!(
             "poh_recorder-tick_overhead",
-            timing::duration_as_ms(&now.elapsed()) as usize,
-            0,
-            1000
+            timing::duration_as_ms(&now.elapsed()) as usize
         );
     }
 
@@ -356,11 +350,9 @@ impl PohRecorder {
 
             let now = Instant::now();
             if let Some(poh_entry) = self.poh.lock().unwrap().record(mixin) {
-                inc_new_counter_warn!(
+                inc_new_high_rate_counter_warn!(
                     "poh_recorder-record_lock_contention",
-                    timing::duration_as_ms(&now.elapsed()) as usize,
-                    0,
-                    1000
+                    timing::duration_as_ms(&now.elapsed()) as usize
                 );
                 let entry = Entry {
                     num_hashes: poh_entry.num_hashes,

core/src/replay_stage.rs

@@ -556,7 +556,7 @@ impl ReplayStage {
                 );
             }
         }
-        inc_new_counter_info!("replay_stage-replay_transactions", tx_count);
+        inc_new_high_rate_counter_info!("replay_stage-replay_transactions", tx_count);
         did_complete_bank
     }
 

core/src/window_service.rs

@@ -39,12 +39,7 @@ fn retransmit_blobs(blobs: &[SharedBlob], retransmit: &BlobSender, id: &Pubkey)
     }
 
     if !retransmit_queue.is_empty() {
-        inc_new_counter_debug!(
-            "streamer-recv_window-retransmit",
-            retransmit_queue.len(),
-            0,
-            1000
-        );
+        inc_new_high_rate_counter_debug!("streamer-recv_window-retransmit", retransmit_queue.len());
         retransmit.send(retransmit_queue)?;
     }
     Ok(())
@@ -118,7 +113,7 @@ where
         blobs.append(&mut blob)
     }
     let now = Instant::now();
-    inc_new_counter_debug!("streamer-recv_window-recv", blobs.len(), 0, 1000);
+    inc_new_high_rate_counter_debug!("streamer-recv_window-recv", blobs.len());
 
     let blobs: Vec<_> = thread_pool.install(|| {
         blobs