Report ConsumeWorkerMetrics at slot transitions (anza-xyz#3212)

* Report ConsumeWorkerMetrics at slot transitions

- Save the slot# while reporting in order to track
slot transitions.
- Report slot# for the three metrics
- Remove the interval as it is not needed anymore.
- Only report when there was a slot
- Reset slot after reporting

Fixes: anza-xyz#478

core/src/banking_stage/consume_worker.rs

@@ -8,7 +8,7 @@ use {
     solana_measure::measure_us,
     solana_poh::leader_bank_notifier::LeaderBankNotifier,
     solana_runtime::bank::Bank,
-    solana_sdk::timing::AtomicInterval,
+    solana_sdk::clock::Slot,
     solana_svm::transaction_error_metrics::TransactionErrorMetrics,
     std::{
         sync::{
@@ -170,32 +170,43 @@ fn try_drain_iter<T>(work: T, receiver: &Receiver<T>) -> impl Iterator<Item = T>
 /// done.
 pub(crate) struct ConsumeWorkerMetrics {
     id: String,
-    interval: AtomicInterval,
     has_data: AtomicBool,
+    slot: AtomicU64,
 
     count_metrics: ConsumeWorkerCountMetrics,
     error_metrics: ConsumeWorkerTransactionErrorMetrics,
     timing_metrics: ConsumeWorkerTimingMetrics,
 }
 
 impl ConsumeWorkerMetrics {
-    /// Report and reset metrics iff the interval has elapsed and the worker did some work.
-    pub fn maybe_report_and_reset(&self) {
-        const REPORT_INTERVAL_MS: u64 = 1000;
-        if self.interval.should_update(REPORT_INTERVAL_MS)
-            && self.has_data.swap(false, Ordering::Relaxed)
-        {
-            self.count_metrics.report_and_reset(&self.id);
-            self.timing_metrics.report_and_reset(&self.id);
-            self.error_metrics.report_and_reset(&self.id);
+    /// Report and reset metrics when the worker did some work and:
+    /// a) (when a leader) Previous slot is not the same as current.
+    /// b) (when not a leader) report the metrics accumulated so far.
+    pub fn maybe_report_and_reset(&self, slot: Option<Slot>) {
+        let prev_slot_id: u64 = self.slot.load(Ordering::Relaxed);
+        if let Some(slot) = slot {
+            if slot != prev_slot_id {
+                if !self.has_data.swap(false, Ordering::Relaxed) {
+                    return;
+                }
+                self.count_metrics.report_and_reset(&self.id, slot);
+                self.timing_metrics.report_and_reset(&self.id, slot);
+                self.error_metrics.report_and_reset(&self.id, slot);
+                self.slot.swap(slot, Ordering::Relaxed);
+            }
+        } else if prev_slot_id != 0 {
+            self.count_metrics.report_and_reset(&self.id, prev_slot_id);
+            self.timing_metrics.report_and_reset(&self.id, prev_slot_id);
+            self.error_metrics.report_and_reset(&self.id, prev_slot_id);
+            self.slot.swap(0, Ordering::Relaxed);
         }
     }
 
     fn new(id: u32) -> Self {
         Self {
             id: id.to_string(),
-            interval: AtomicInterval::default(),
             has_data: AtomicBool::new(false),
+            slot: AtomicU64::new(0),
             count_metrics: ConsumeWorkerCountMetrics::default(),
             error_metrics: ConsumeWorkerTransactionErrorMetrics::default(),
             timing_metrics: ConsumeWorkerTimingMetrics::default(),
@@ -432,7 +443,7 @@ impl Default for ConsumeWorkerCountMetrics {
 }
 
 impl ConsumeWorkerCountMetrics {
-    fn report_and_reset(&self, id: &str) {
+    fn report_and_reset(&self, id: &str, slot: u64) {
         datapoint_info!(
             "banking_stage_worker_counts",
             "id" => id,
@@ -480,6 +491,11 @@ impl ConsumeWorkerCountMetrics {
                 self.max_prioritization_fees.swap(0, Ordering::Relaxed),
                 i64
             ),
+            (
+                "slot",
+                slot,
+                i64
+            ),
         );
     }
 }
@@ -498,7 +514,7 @@ struct ConsumeWorkerTimingMetrics {
 }
 
 impl ConsumeWorkerTimingMetrics {
-    fn report_and_reset(&self, id: &str) {
+    fn report_and_reset(&self, id: &str, slot: u64) {
         datapoint_info!(
             "banking_stage_worker_timing",
             "id" => id,
@@ -539,6 +555,11 @@ impl ConsumeWorkerTimingMetrics {
                 self.wait_for_bank_failure_us.swap(0, Ordering::Relaxed),
                 i64
             ),
+            (
+                "slot",
+                slot,
+                i64
+            ),
         );
     }
 }
@@ -572,7 +593,7 @@ struct ConsumeWorkerTransactionErrorMetrics {
 }
 
 impl ConsumeWorkerTransactionErrorMetrics {
-    fn report_and_reset(&self, id: &str) {
+    fn report_and_reset(&self, id: &str, slot: u64) {
         datapoint_info!(
             "banking_stage_worker_error_metrics",
             "id" => id,
@@ -683,6 +704,11 @@ impl ConsumeWorkerTransactionErrorMetrics {
                     .swap(0, Ordering::Relaxed),
                 i64
             ),
+            (
+                "slot",
+                slot,
+                i64
+            ),
         );
     }
 }

core/src/banking_stage/transaction_scheduler/scheduler_controller.rs

@@ -140,7 +140,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
                 .maybe_report_and_reset_interval(should_report);
             self.worker_metrics
                 .iter()
-                .for_each(|metrics| metrics.maybe_report_and_reset());
+                .for_each(|metrics| metrics.maybe_report_and_reset(new_leader_slot));
         }
 
         Ok(())