Datapoints overwhelm the metrics queue and blow up ram usage. (#8272)

automerge

core/src/banking_stage.rs

@@ -404,7 +404,7 @@ impl BankingStage {
                     if unprocessed_packets.is_empty() {
                         continue;
                     }
-                    let num = unprocessed_packets
+                    let num: usize = unprocessed_packets
                         .iter()
                         .map(|(_, unprocessed)| unprocessed.len())
                         .sum();

core/src/cluster_info.rs

@@ -1112,6 +1112,7 @@ impl ClusterInfo {
             .unwrap()
     }
 
+    #[allow(clippy::cognitive_complexity)]
     fn handle_packets(
         me: &Arc<RwLock<Self>>,
         recycler: &PacketsRecycler,

core/src/replay_stage.rs

@@ -193,6 +193,7 @@ impl ReplayStage {
         let (lockouts_sender, commitment_service) =
             AggregateCommitmentService::new(&exit, block_commitment_cache);
 
+        #[allow(clippy::cognitive_complexity)]
         let t_replay = Builder::new()
             .name("solana-replay-stage".to_string())
             .spawn(move || {

core/src/validator.rs

@@ -132,6 +132,7 @@ pub struct Validator {
 }
 
 impl Validator {
+    #[allow(clippy::cognitive_complexity)]
     pub fn new(
         mut node: Node,
         keypair: &Arc<Keypair>,

ledger/src/blockstore_meta.rs

@@ -1,6 +1,5 @@
 use crate::erasure::ErasureConfig;
 use serde::{Deserialize, Serialize};
-use solana_metrics::datapoint;
 use solana_sdk::clock::Slot;
 use std::{collections::BTreeSet, ops::RangeBounds};
 
@@ -138,7 +137,7 @@ impl SlotMeta {
 
         // Should never happen
         if self.consumed > self.last_index + 1 {
-            datapoint!(
+            datapoint_error!(
                 "blockstore_error",
                 (
                     "error",

ledger/src/blockstore_processor.rs

@@ -13,7 +13,7 @@ use log::*;
 use rand::{seq::SliceRandom, thread_rng};
 use rayon::{prelude::*, ThreadPool};
 use solana_measure::{measure::Measure, thread_mem_usage};
-use solana_metrics::{datapoint, datapoint_error, inc_new_counter_debug};
+use solana_metrics::{datapoint_error, inc_new_counter_debug};
 use solana_rayon_threadlimit::get_thread_count;
 use solana_runtime::{
     bank::{Bank, TransactionBalancesSet, TransactionProcessResult, TransactionResults},
@@ -201,7 +201,7 @@ fn process_entries_with_callback(
             if batches.is_empty() {
                 // An entry has account lock conflicts with *itself*, which should not happen
                 // if generated by a properly functioning leader
-                datapoint!(
+                datapoint_error!(
                     "validator_process_entry_error",
                     (
                         "error",

metrics/src/datapoint.rs

@@ -47,8 +47,9 @@ impl fmt::Display for DataPoint {
         Ok(())
     }
 }
+
 #[macro_export]
-macro_rules! datapoint {
+macro_rules! create_datapoint {
     (@field $point:ident $name:expr, $string:expr, String) => {
         $point.add_field_str($name, &$string);
     };
@@ -64,67 +65,85 @@ macro_rules! datapoint {
 
     (@fields $point:ident) => {};
     (@fields $point:ident ($name:expr, $value:expr, $type:ident) , $($rest:tt)*) => {
-        $crate::datapoint!(@field $point $name, $value, $type);
+        $crate::create_datapoint!(@field $point $name, $value, $type);
-        $crate::datapoint!(@fields $point $($rest)*);
+        $crate::create_datapoint!(@fields $point $($rest)*);
     };
     (@fields $point:ident ($name:expr, $value:expr, $type:ident)) => {
-        $crate::datapoint!(@field $point $name, $value, $type);
+        $crate::create_datapoint!(@field $point $name, $value, $type);
     };
 
     (@point $name:expr, $($fields:tt)+) => {
         {
             let mut point = $crate::datapoint::DataPoint::new(&$name);
-            $crate::datapoint!(@fields point $($fields)+);
+            $crate::create_datapoint!(@fields point $($fields)+);
             point
         }
     };
     (@point $name:expr) => {
         $crate::datapoint::DataPoint::new(&$name)
     };
-    ($name:expr, $($fields:tt)+) => {
-        if log::log_enabled!(log::Level::Debug) {
-            $crate::submit($crate::datapoint!(@point $name, $($fields)+), log::Level::Debug);
-        }
-    };
 }
 
+#[macro_export]
+macro_rules! datapoint {
+    ($level:expr, $name:expr) => {
+        if log::log_enabled!($level) {
+            $crate::submit($crate::create_datapoint!(@point $name), $level);
+        }
+    };
+    ($level:expr, $name:expr, $($fields:tt)+) => {
+        if log::log_enabled!($level) {
+            $crate::submit($crate::create_datapoint!(@point $name, $($fields)+), $level);
+        }
+    };
+}
 #[macro_export]
 macro_rules! datapoint_error {
     ($name:expr) => {
-        $crate::submit($crate::datapoint!(@point $name), log::Level::Error);
+        $crate::datapoint!(log::Level::Error, $name);
     };
     ($name:expr, $($fields:tt)+) => {
-        $crate::submit($crate::datapoint!(@point $name, $($fields)+), log::Level::Error);
+        $crate::datapoint!(log::Level::Error, $name, $($fields)+);
     };
 }
 
 #[macro_export]
 macro_rules! datapoint_warn {
     ($name:expr) => {
-        $crate::submit($crate::datapoint!(@point $name), log::Level::Warn);
+        $crate::datapoint!(log::Level::Warn, $name);
     };
     ($name:expr, $($fields:tt)+) => {
-        $crate::submit($crate::datapoint!(@point $name, $($fields)+), log::Level::Warn);
+        $crate::datapoint!(log::Level::Warn, $name, $($fields)+);
     };
 }
 
 #[macro_export]
 macro_rules! datapoint_info {
     ($name:expr) => {
-        $crate::submit($crate::datapoint!(@point $name), log::Level::Info);
+        $crate::datapoint!(log::Level::Info, $name);
     };
     ($name:expr, $($fields:tt)+) => {
-        $crate::submit($crate::datapoint!(@point $name, $($fields)+), log::Level::Info);
+        $crate::datapoint!(log::Level::Info, $name, $($fields)+);
     };
 }
 
 #[macro_export]
 macro_rules! datapoint_debug {
     ($name:expr) => {
-        $crate::submit($crate::datapoint!(@point $name), log::Level::Debug);
+        $crate::datapoint!(log::Level::Debug, $name);
     };
     ($name:expr, $($fields:tt)+) => {
-        $crate::submit($crate::datapoint!(@point $name, $($fields)+), log::Level::Debug);
+        $crate::datapoint!(log::Level::Debug, $name, $($fields)+);
+    };
+}
+
+#[macro_export]
+macro_rules! datapoint_trace {
+    ($name:expr) => {
+        $crate::datapoint!(log::Level::Trace, $name);
+    };
+    ($name:expr, $($fields:tt)+) => {
+        $crate::datapoint!(log::Level::Trace, $name, $($fields)+);
     };
 }
 
@@ -132,27 +151,33 @@ macro_rules! datapoint_debug {
 mod test {
     #[test]
     fn test_datapoint() {
-        datapoint!("name", ("field name", "test".to_string(), String));
+        datapoint_debug!("name", ("field name", "test".to_string(), String));
-        datapoint!("name", ("field name", 12.34_f64, f64));
+        datapoint_info!("name", ("field name", 12.34_f64, f64));
-        datapoint!("name", ("field name", true, bool));
+        datapoint_trace!("name", ("field name", true, bool));
-        datapoint!("name", ("field name", 1, i64));
+        datapoint_warn!("name", ("field name", 1, i64));
-        datapoint!("name", ("field name", 1, i64),);
+        datapoint_error!("name", ("field name", 1, i64),);
-        datapoint!("name", ("field1 name", 2, i64), ("field2 name", 2, i64));
-        datapoint!("name", ("field1 name", 2, i64), ("field2 name", 2, i64),);
         datapoint!(
+            log::Level::Warn,
+            "name",
+            ("field1 name", 2, i64),
+            ("field2 name", 2, i64)
+        );
+        datapoint_info!("name", ("field1 name", 2, i64), ("field2 name", 2, i64),);
+        datapoint_trace!(
             "name",
             ("field1 name", 2, i64),
             ("field2 name", 2, i64),
             ("field3 name", 3, i64)
         );
         datapoint!(
+            log::Level::Error,
             "name",
             ("field1 name", 2, i64),
             ("field2 name", 2, i64),
             ("field3 name", 3, i64),
         );
 
-        let point = datapoint!(
+        let point = create_datapoint!(
             @point "name",
             ("i64", 1, i64),
             ("String", "string space string".to_string(), String),