diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index 77c353d069..ed8cd4728a 100644 --- a/mixnode/src/config/mod.rs +++ b/mixnode/src/config/mod.rs @@ -17,6 +17,7 @@ const DEFAULT_LISTENING_PORT: u16 = 1789; // where applicable, the below are defined in milliseconds const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 1500; const DEFAULT_METRICS_SENDING_DELAY: u64 = 1000; +const DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY: u64 = 60_000; // 1min const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min @@ -213,6 +214,10 @@ impl Config { time::Duration::from_millis(self.debug.metrics_sending_delay) } + pub fn get_metrics_running_stats_logging_delay(&self) -> time::Duration { + time::Duration::from_millis(self.debug.metrics_running_stats_logging_delay) + } + pub fn get_layer(&self) -> u64 { self.mixnode.layer } @@ -331,6 +336,10 @@ pub struct Debug { /// The provided value is interpreted as milliseconds. metrics_sending_delay: u64, + /// Delay between each subsequent running metrics statistics being logged. + /// The provided value is interpreted as milliseconds. + metrics_running_stats_logging_delay: u64, + /// Initial value of an exponential backoff to reconnect to dropped TCP connection when /// forwarding sphinx packets. /// The provided value is interpreted as milliseconds. @@ -360,6 +369,7 @@ impl Default for Debug { presence_sending_delay: DEFAULT_PRESENCE_SENDING_DELAY, metrics_directory_server: Self::default_directory_server(), metrics_sending_delay: DEFAULT_METRICS_SENDING_DELAY, + metrics_running_stats_logging_delay: DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY, packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF, packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF, } diff --git a/mixnode/src/node/metrics.rs b/mixnode/src/node/metrics.rs index f62a470407..87712380db 100644 --- a/mixnode/src/node/metrics.rs +++ b/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; + pub(crate) enum MetricEvent { Sent(String), Received, @@ -25,7 +27,7 @@ struct MixMetrics { struct MixMetricsInner { received: u64, - sent: HashMap, + sent: SentMetricsMap, } impl MixMetrics { @@ -49,7 +51,7 @@ impl MixMetrics { *receiver_count += 1; } - async fn acquire_and_reset_metrics(&mut self) -> (u64, HashMap) { + 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::() + ); + 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::() + ); + 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, @@ -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), diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index ee992f9568..547d6c2e63 100644 --- a/mixnode/src/node/mod.rs +++ b/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", diff --git a/sfw-provider/src/provider/mod.rs b/sfw-provider/src/provider/mod.rs index b0a3b8ecfb..fc2aba3e20 100644 --- a/sfw-provider/src/provider/mod.rs +++ b/sfw-provider/src/provider/mod.rs @@ -105,6 +105,7 @@ impl ServiceProvider { // considering, presumably, there will be more mix packets received than client requests: // create 2 separate runtimes - one with bigger threadpool dedicated solely for // the mix handling and the other one for the rest of tasks + info!("Starting nym sfw-provider"); if let Some(duplicate_provider_key) = self.check_if_same_ip_provider_exists() { error!( @@ -124,6 +125,8 @@ impl ServiceProvider { self.start_mix_socket_listener(client_storage.clone()); self.start_client_socket_listener(client_storage); + info!("Finished nym sfw-provider startup procedure - it should now be able to receive mix and client 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",