[Consensus] count missing ancestors and blocks per authority

consensus/core/src/block_manager.rs

@@ -229,11 +229,26 @@ impl BlockManager {
                     .or_default()
                     .insert(block_ref);
 
+                let ancestor_hostname = &self.context.committee.authority(ancestor.author).hostname;
+                self.context
+                    .metrics
+                    .node_metrics
+                    .block_manager_missing_ancestor_by_authority
+                    .with_label_values(&[ancestor_hostname])
+                    .inc();
+
                 // Add the ancestor to the missing blocks set only if it doesn't already exist in the suspended blocks - meaning
                 // that we already have its payload.
                 if !self.suspended_blocks.contains_key(ancestor) {
-                    self.missing_blocks.insert(*ancestor);
                     ancestors_to_fetch.insert(*ancestor);
+                    if self.missing_blocks.insert(*ancestor) {
+                        self.context
+                            .metrics
+                            .node_metrics
+                            .block_manager_missing_block_by_authority
+                            .with_label_values(&[ancestor_hostname])
+                            .inc();
+                    }
                 }
             }
         }

consensus/core/src/metrics.rs

@@ -152,6 +152,8 @@ pub(crate) struct NodeMetrics {
     pub(crate) block_manager_suspended_blocks: IntGauge,
     pub(crate) block_manager_missing_ancestors: IntGauge,
     pub(crate) block_manager_missing_blocks: IntGauge,
+    pub(crate) block_manager_missing_block_by_authority: IntCounterVec,
+    pub(crate) block_manager_missing_ancestor_by_authority: IntCounterVec,
     pub(crate) threshold_clock_round: IntGauge,
     pub(crate) subscriber_connection_attempts: IntCounterVec,
     pub(crate) subscriber_connections: IntGaugeVec,
@@ -488,6 +490,18 @@ impl NodeMetrics {
                 "The number of blocks missing content tracked in the block manager",
                 registry,
             ).unwrap(),
+            block_manager_missing_block_by_authority: register_int_counter_vec_with_registry!(
+                "block_manager_missing_block_by_authority",
+                "The number of new missing blocks by authority",
+                &["authority"],
+                registry,
+            ).unwrap(),
+            block_manager_missing_ancestor_by_authority: register_int_counter_vec_with_registry!(
+                "block_manager_missing_ancestor_by_authority",
+                "The number of missing ancestors by authority",
+                &["authority"],
+                registry,
+            ).unwrap(),
             threshold_clock_round: register_int_gauge_with_registry!(
                 "threshold_clock_round",
                 "The current threshold clock round. We only advance to a new round when a quorum of parents have been synced.",