diff --git a/common/client-libs/metrics-client/src/lib.rs b/common/client-libs/metrics-client/src/lib.rs index cbf4588f1a..a38caeb4fb 100644 --- a/common/client-libs/metrics-client/src/lib.rs +++ b/common/client-libs/metrics-client/src/lib.rs @@ -15,7 +15,7 @@ pub mod models; pub mod requests; -use crate::models::metrics::{MixMetric, PersistedMixMetric}; +use crate::models::metrics::{MixMetric, MixMetricInterval, PersistedMixMetric}; use crate::requests::metrics_mixes_get::Request as MetricsMixRequest; use crate::requests::metrics_mixes_post::Request as MetricsMixPost; @@ -43,12 +43,14 @@ impl Client { } } - pub async fn post_mix_metrics(&self, metrics: MixMetric) -> reqwest::Result { + pub async fn post_mix_metrics(&self, metrics: MixMetric) -> reqwest::Result { let req = MetricsMixPost::new(&self.base_url, metrics); self.reqwest_client .post(&req.url()) .json(req.json_payload()) .send() + .await? + .json() .await } diff --git a/common/client-libs/metrics-client/src/models/metrics.rs b/common/client-libs/metrics-client/src/models/metrics.rs index 265ae16575..23a4bcd537 100644 --- a/common/client-libs/metrics-client/src/models/metrics.rs +++ b/common/client-libs/metrics-client/src/models/metrics.rs @@ -31,3 +31,9 @@ pub struct MixMetric { pub received: u64, pub sent: HashMap, } + +#[derive(Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct MixMetricInterval { + pub next_report_in: u64, +} diff --git a/common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs b/common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs index 60f7adee59..1e38f2d46a 100644 --- a/common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs +++ b/common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs @@ -55,7 +55,7 @@ mod metrics_post_request { let _m = mock("POST", PATH).with_status(400).create(); let client = client_test_fixture(&mockito::server_url()); let result = client.post_mix_metrics(fixtures::new_metric()).await; - assert_eq!(400, result.unwrap().status()); + assert!(result.is_err()); _m.assert(); } } @@ -65,7 +65,7 @@ mod metrics_post_request { use super::*; #[tokio::test] async fn it_returns_a_response_with_200() { - let json = fixtures::mix_metrics_response_json(); + let json = fixtures::metrics_interval_json(); let _m = mock("POST", "/api/metrics/mixes") .with_status(201) .with_body(json) @@ -90,17 +90,11 @@ mod metrics_post_request { } #[cfg(test)] - pub fn mix_metrics_response_json() -> String { + pub fn metrics_interval_json() -> String { r#" - { - "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=", - "sent": { - "35.178.213.77:1789": 1, - "52.56.99.196:1789": 2 - }, - "received": 10, - "timestamp": 1576061080635800000 - } + { + "nextReportIn": 5 + } "# .to_string() } diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index 78fa77df95..1bc808e8c4 100644 --- a/mixnode/src/config/mod.rs +++ b/mixnode/src/config/mod.rs @@ -35,7 +35,6 @@ pub(crate) const DEFAULT_VALIDATOR_REST_ENDPOINT: &str = "https://directory.nymt pub(crate) const DEFAULT_METRICS_SERVER: &str = "https://metrics.nymtech.net"; // 'DEBUG' -const DEFAULT_METRICS_SENDING_DELAY: Duration = Duration::from_millis(5_000); const DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY: Duration = Duration::from_millis(60_000); const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: Duration = Duration::from_millis(10_000); const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_millis(300_000); @@ -303,10 +302,6 @@ impl Config { self.mixnode.metrics_server_url.clone() } - pub fn get_metrics_sending_delay(&self) -> Duration { - self.debug.metrics_sending_delay - } - pub fn get_metrics_running_stats_logging_delay(&self) -> Duration { self.debug.metrics_running_stats_logging_delay } @@ -465,13 +460,6 @@ impl Default for Logging { #[derive(Debug, Deserialize, PartialEq, Serialize)] #[serde(default, deny_unknown_fields)] pub struct Debug { - /// Delay between each subsequent metrics data being sent. - #[serde( - deserialize_with = "deserialize_duration", - serialize_with = "humantime_serde::serialize" - )] - metrics_sending_delay: Duration, - /// Delay between each subsequent running metrics statistics being logged. #[serde( deserialize_with = "deserialize_duration", @@ -513,7 +501,6 @@ pub struct Debug { impl Default for Debug { fn default() -> Self { Debug { - 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 5cf8b03e3e..929923bfd0 100644 --- a/mixnode/src/node/metrics.rs +++ b/mixnode/src/node/metrics.rs @@ -22,6 +22,8 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use tokio::task::JoinHandle; +const METRICS_FAILURE_BACKOFF: Duration = Duration::from_secs(30); + type SentMetricsMap = HashMap; pub(crate) enum MetricEvent { @@ -104,7 +106,6 @@ struct MetricsSender { metrics: MixMetrics, metrics_client: metrics_client::Client, pub_key_str: String, - sending_delay: Duration, metrics_informer: MetricsInformer, } @@ -113,7 +114,6 @@ impl MetricsSender { metrics: MixMetrics, metrics_server: String, pub_key_str: String, - sending_delay: Duration, running_logging_delay: Duration, ) -> Self { MetricsSender { @@ -122,7 +122,6 @@ impl MetricsSender { metrics_server, )), pub_key_str, - sending_delay, metrics_informer: MetricsInformer::new(running_logging_delay), } } @@ -130,15 +129,15 @@ impl MetricsSender { fn start(mut self) -> JoinHandle<()> { tokio::spawn(async move { loop { - // set the deadline in the future - let sending_delay = tokio::time::delay_for(self.sending_delay); + // // set the deadline in the future + // 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 + let sending_delay = match self .metrics_client .post_mix_metrics(MixMetric { pub_key: self.pub_key_str.clone(), @@ -147,9 +146,16 @@ impl MetricsSender { }) .await { - Err(err) => error!("failed to send metrics - {:?}", err), - Ok(_) => debug!("sent metrics information"), - } + Err(err) => { + error!("failed to send metrics - {:?}", err); + tokio::time::delay_for(METRICS_FAILURE_BACKOFF) + } + Ok(new_interval) => { + debug!("sent metrics information"); + println!("received delay: {:?}", new_interval.next_report_in); + tokio::time::delay_for(Duration::from_secs(new_interval.next_report_in)) + } + }; // wait for however much is left sending_delay.await; @@ -261,7 +267,6 @@ impl MetricsController { pub(crate) fn new( directory_server: String, pub_key_str: String, - sending_delay: Duration, running_stats_logging_delay: Duration, ) -> Self { let (metrics_tx, metrics_rx) = mpsc::unbounded(); @@ -272,7 +277,6 @@ impl MetricsController { shared_metrics.clone(), directory_server, pub_key_str, - sending_delay, running_stats_logging_delay, ), receiver: MetricsReceiver::new(shared_metrics, metrics_rx), diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index ad1f99d095..e72b76035e 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -52,7 +52,6 @@ impl MixNode { metrics::MetricsController::new( self.config.get_metrics_server(), self.sphinx_keypair.public_key().to_base58_string(), - self.config.get_metrics_sending_delay(), self.config.get_metrics_running_stats_logging_delay(), ) .start()