feat: record signatures origin in the signature processor

Author: turmelclem

mithril-aggregator/src/dependency_injection/builder/enablers/misc.rs

@@ -87,6 +87,7 @@ impl DependenciesBuilder {
             self.get_certifier_service().await?,
             stop_rx,
             self.root_logger(),
+            self.get_metrics_service().await?,
         );
 
         Ok(Arc::new(signature_processor))

mithril-aggregator/src/services/signature_consumer/fake.rs

@@ -37,6 +37,9 @@ impl SignatureConsumer for FakeSignatureConsumer {
             }
         }
     }
+    fn get_origin_tag(&self) -> String {
+        "FAKE".to_string()
+    }
 }
 
 #[cfg(test)]

mithril-aggregator/src/services/signature_consumer/interface.rs

@@ -9,4 +9,7 @@ use mithril_common::{
 pub trait SignatureConsumer: Sync + Send {
     /// Returns signatures when available
     async fn get_signatures(&self) -> StdResult<Vec<(SingleSignature, SignedEntityType)>>;
+
+    /// Returns the origin tag of the consumer (e.g. HTTP or DMQ)
+    fn get_origin_tag(&self) -> String;
 }

mithril-aggregator/src/services/signature_consumer/noop.rs

@@ -19,6 +19,10 @@ impl SignatureConsumer for SignatureConsumerNoop {
     > {
         future::pending().await
     }
+
+    fn get_origin_tag(&self) -> String {
+        "NOOP".to_string()
+    }
 }
 
 #[cfg(test)]

mithril-aggregator/src/services/signature_processor.rs

@@ -5,6 +5,8 @@ use slog::{error, warn, Logger};
 use mithril_common::{logging::LoggerExtensions, StdResult};
 use tokio::{select, sync::watch::Receiver};
 
+use crate::MetricsService;
+
 use super::{CertifierService, SignatureConsumer};
 
 /// A signature processor which receives signature and processes them.
@@ -24,6 +26,7 @@ pub struct SequentialSignatureProcessor {
     certifier: Arc<dyn CertifierService>,
     stop_rx: Receiver<()>,
     logger: Logger,
+    metrics_service: Arc<MetricsService>,
 }
 
 impl SequentialSignatureProcessor {
@@ -33,12 +36,14 @@ impl SequentialSignatureProcessor {
         certifier: Arc<dyn CertifierService>,
         stop_rx: Receiver<()>,
         logger: Logger,
+        metrics_service: Arc<MetricsService>,
     ) -> Self {
         Self {
             consumer,
             certifier,
             stop_rx,
             logger: logger.new_with_component_name::<Self>(),
+            metrics_service,
         }
     }
 }
@@ -55,6 +60,11 @@ impl SignatureProcessor for SequentialSignatureProcessor {
                         .await
                     {
                         error!(self.logger, "Error dispatching single signature"; "error" => ?e);
+                    } else {
+                        let origin_network = self.consumer.get_origin_tag();
+                        self.metrics_service
+                            .get_signature_registration_total_received_since_startup()
+                            .increment(&[&origin_network]);
                     }
                 }
             }
@@ -107,24 +117,29 @@ mod tests {
     #[tokio::test]
     async fn processor_process_signatures_succeeds() {
         let logger = TestLogger::stdout();
+        let single_signatures = vec![
+            (
+                fake_data::single_signature(vec![1, 2, 3]),
+                SignedEntityType::MithrilStakeDistribution(Epoch(1)),
+            ),
+            (
+                fake_data::single_signature(vec![4, 5, 6]),
+                SignedEntityType::MithrilStakeDistribution(Epoch(2)),
+            ),
+        ];
+        let single_signatures_length = single_signatures.len();
+        let network_origin = "test_network";
         let mock_consumer = {
             let mut mock_consumer = MockSignatureConsumer::new();
             mock_consumer
                 .expect_get_signatures()
-                .returning(|| {
-                    Ok(vec![
-                        (
-                            fake_data::single_signature(vec![1, 2, 3]),
-                            SignedEntityType::MithrilStakeDistribution(Epoch(1)),
-                        ),
-                        (
-                            fake_data::single_signature(vec![4, 5, 6]),
-                            SignedEntityType::MithrilStakeDistribution(Epoch(2)),
-                        ),
-                    ])
-                })
+                .returning(move || Ok(single_signatures.clone()))
                 .times(1);
             mock_consumer
+                .expect_get_origin_tag()
+                .returning(|| network_origin.to_string())
+                .times(single_signatures_length);
+            mock_consumer
         };
         let mock_certifier = {
             let mut mock_certifier = MockCertifierService::new();
@@ -144,21 +159,33 @@ mod tests {
                 )
                 .returning(|_, _| Ok(SignatureRegistrationStatus::Registered))
                 .times(1);
-
             mock_certifier
         };
         let (_stop_tx, stop_rx) = channel(());
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
+        let initial_counter_value = metrics_service
+            .get_signature_registration_total_received_since_startup()
+            .get(&[network_origin]);
+        let metrics_service = Arc::new(metrics_service);
         let processor = SequentialSignatureProcessor::new(
             Arc::new(mock_consumer),
             Arc::new(mock_certifier),
             stop_rx,
             logger,
+            metrics_service.clone(),
         );
 
         processor
             .process_signatures()
             .await
             .expect("Failed to process signatures");
+
+        assert_eq!(
+            initial_counter_value + single_signatures_length as u32,
+            metrics_service
+                .get_signature_registration_total_received_since_startup()
+                .get(&[network_origin])
+        )
     }
 
     #[tokio::test]
@@ -185,11 +212,13 @@ mod tests {
             mock_certifier
         };
         let (stop_tx, stop_rx) = channel(());
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
         let processor = SequentialSignatureProcessor::new(
             Arc::new(fake_consumer),
             Arc::new(mock_certifier),
             stop_rx,
             logger,
+            Arc::new(metrics_service),
         );
 
         tokio::select!(