Feature/mixing stats logging (#175)

* Extra startup log messages

* Type alias for sent metrics map

* Initial metrics informer

* Separated report and running stats

* Decreases report logging level

* Added logging delay as a config value

* New metrics informer constructor

* Determining if running stats should be logged

* Separated running total and reports logging

* Missing changes

mixnode/src/node/metrics.rs

@@ -4,13 +4,15 @@ use directory_client::DirectoryClient;
 use futures::channel::mpsc;
 use futures::lock::Mutex;
 use futures::StreamExt;
-use log::{debug, error};
+use log::*;
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
 
+type SentMetricsMap = HashMap<String, u64>;
+
 pub(crate) enum MetricEvent {
     Sent(String),
     Received,
@@ -25,7 +27,7 @@ struct MixMetrics {
 
 struct MixMetricsInner {
     received: u64,
-    sent: HashMap<String, u64>,
+    sent: SentMetricsMap,
 }
 
 impl MixMetrics {
@@ -49,7 +51,7 @@ impl MixMetrics {
         *receiver_count += 1;
     }
 
-    async fn acquire_and_reset_metrics(&mut self) -> (u64, HashMap<String, u64>) {
+    async fn acquire_and_reset_metrics(&mut self) -> (u64, SentMetricsMap) {
         let mut unlocked = self.inner.lock().await;
         let received = unlocked.received;
 
@@ -92,6 +94,7 @@ struct MetricsSender {
     directory_client: directory_client::Client,
     pub_key_str: String,
     sending_delay: Duration,
+    metrics_informer: MetricsInformer,
 }
 
 impl MetricsSender {
@@ -100,6 +103,7 @@ impl MetricsSender {
         directory_server: String,
         pub_key_str: String,
         sending_delay: Duration,
+        running_logging_delay: Duration,
     ) -> Self {
         MetricsSender {
             metrics,
@@ -108,6 +112,7 @@ impl MetricsSender {
             )),
             pub_key_str,
             sending_delay,
+            metrics_informer: MetricsInformer::new(running_logging_delay),
         }
     }
 
@@ -118,6 +123,10 @@ impl MetricsSender {
                 let sending_delay = tokio::time::delay_for(self.sending_delay);
                 let (received, sent) = self.metrics.acquire_and_reset_metrics().await;
 
+                self.metrics_informer.update_running_stats(received, &sent);
+                self.metrics_informer.log_report_stats(received, &sent);
+                self.metrics_informer.try_log_running_stats();
+
                 match self.directory_client.metrics_post.post(&MixMetric {
                     pub_key: self.pub_key_str.clone(),
                     received,
@@ -134,6 +143,71 @@ impl MetricsSender {
     }
 }
 
+struct MetricsInformer {
+    total_received: u64,
+    sent_map: SentMetricsMap,
+
+    running_stats_logging_delay: Duration,
+    last_reported_stats: SystemTime,
+}
+
+impl MetricsInformer {
+    fn new(running_stats_logging_delay: Duration) -> Self {
+        MetricsInformer {
+            total_received: 0,
+            sent_map: HashMap::new(),
+            running_stats_logging_delay,
+            last_reported_stats: SystemTime::now(),
+        }
+    }
+
+    fn should_log_running_stats(&self) -> bool {
+        self.last_reported_stats + self.running_stats_logging_delay < SystemTime::now()
+    }
+
+    fn try_log_running_stats(&mut self) {
+        if self.should_log_running_stats() {
+            self.log_running_stats()
+        }
+    }
+
+    fn update_running_stats(&mut self, pre_reset_received: u64, pre_reset_sent: &SentMetricsMap) {
+        self.total_received += pre_reset_received;
+
+        for (mix, count) in pre_reset_sent.iter() {
+            *self.sent_map.entry(mix.clone()).or_insert(0) += *count;
+        }
+    }
+
+    fn log_report_stats(&self, pre_reset_received: u64, pre_reset_sent: &SentMetricsMap) {
+        debug!(
+            "Since last metrics report mixed {} packets!",
+            pre_reset_received
+        );
+        debug!(
+            "Since last metrics report received {} packets",
+            pre_reset_sent.values().sum::<u64>()
+        );
+        trace!(
+            "Since last metrics report sent packets to the following: \n{:#?}",
+            pre_reset_sent
+        );
+    }
+
+    fn log_running_stats(&mut self) {
+        info!(
+            "Since startup mixed {} packets!",
+            self.sent_map.values().sum::<u64>()
+        );
+        debug!("Since startup received {} packets", self.total_received);
+        trace!(
+            "Since startup sent packets to the following: \n{:#?}",
+            self.sent_map
+        );
+        self.last_reported_stats = SystemTime::now();
+    }
+}
+
 #[derive(Clone)]
 pub struct MetricsReporter {
     metrics_tx: mpsc::UnboundedSender<MetricEvent>,
@@ -173,6 +247,7 @@ impl MetricsController {
         directory_server: String,
         pub_key_str: String,
         sending_delay: Duration,
+        running_stats_logging_delay: Duration,
     ) -> Self {
         let (metrics_tx, metrics_rx) = mpsc::unbounded();
         let shared_metrics = MixMetrics::new();
@@ -183,6 +258,7 @@ impl MetricsController {
                 directory_server,
                 pub_key_str,
                 sending_delay,
+                running_stats_logging_delay,
             ),
             receiver: MetricsReceiver::new(shared_metrics, metrics_rx),
             reporter: MetricsReporter::new(metrics_tx),

mixnode/src/node/mod.rs

@@ -64,6 +64,7 @@ impl MixNode {
             self.config.get_metrics_directory_server(),
             self.sphinx_keypair.public_key().to_base58_string(),
             self.config.get_metrics_sending_delay(),
+            self.config.get_metrics_running_stats_logging_delay(),
         )
         .start(self.runtime.handle())
     }
@@ -110,6 +111,8 @@ impl MixNode {
     }
 
     pub fn run(&mut self) {
+        info!("Starting nym mixnode");
+
         if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists() {
             error!(
                 "Our announce-host is identical to one of existing nodes! (its key is {:?}",
@@ -122,6 +125,8 @@ impl MixNode {
         self.start_socket_listener(metrics_reporter, forwarding_channel);
         self.start_presence_notifier();
 
+        info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
+
         if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) {
             error!(
                 "There was an error while capturing SIGINT - {:?}. We will terminate regardless",