Add metric measuring number of successfully inserted push messages (#20275)

* Add number of successfully inserted push messages

gossip/src/cluster_info.rs

@@ -2181,7 +2181,12 @@ impl ClusterInfo {
             messages
                 .into_iter()
                 .flat_map(|(from, crds_values)| {
-                    self.gossip.process_push_message(&from, crds_values, now)
+                    let (num_success, origins) =
+                        self.gossip.process_push_message(&from, crds_values, now);
+                    self.stats
+                        .process_push_success
+                        .add_relaxed(num_success as u64);
+                    origins
                 })
                 .collect()
         };

gossip/src/cluster_info_metrics.rs

@@ -131,6 +131,7 @@ pub(crate) struct GossipStats {
     pub(crate) process_pull_response_success: Counter,
     pub(crate) process_pull_response_timeout: Counter,
     pub(crate) process_push_message: Counter,
+    pub(crate) process_push_success: Counter,
     pub(crate) prune_message_count: Counter,
     pub(crate) prune_message_len: Counter,
     pub(crate) prune_received_cache: Counter,
@@ -199,6 +200,11 @@ pub(crate) fn submit_gossip_stats(
         ("repair_peers", stats.repair_peers.clear(), i64),
         ("new_push_requests", stats.new_push_requests.clear(), i64),
         ("new_push_requests2", stats.new_push_requests2.clear(), i64),
+        (
+            "process_push_success",
+            stats.process_push_success.clear(),
+            i64
+        ),
         ("purge", stats.purge.clear(), i64),
         (
             "process_gossip_packets_time",

gossip/src/crds_gossip.rs

@@ -49,12 +49,22 @@ impl CrdsGossip {
         from: &Pubkey,
         values: Vec<CrdsValue>,
         now: u64,
-    ) -> HashSet<Pubkey> {
-        self.push
-            .process_push_message(&self.crds, from, values, now)
+    ) -> (usize, HashSet<Pubkey>) {
+        let results = self
+            .push
+            .process_push_message(&self.crds, from, values, now);
+        let mut success_count = 0;
+        let successfully_inserted_origin_set: HashSet<Pubkey> = results
             .into_iter()
-            .filter_map(Result::ok)
-            .collect()
+            .filter_map(|result| {
+                if result.is_ok() {
+                    success_count += 1;
+                }
+                Result::ok(result)
+            })
+            .collect();
+
+        (success_count, successfully_inserted_origin_set)
     }
 
     /// Remove redundant paths in the network.

gossip/tests/crds_gossip.rs

@@ -357,6 +357,7 @@ fn network_run_push(
                         .unwrap()
                         .gossip
                         .process_push_message(&from, msgs.clone(), now)
+                        .1
                         .into_iter()
                         .collect();
                     let prunes_map = network