From 82def1349fbbb969e5ce752fcbe297c869e7524d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 7 Jun 2021 12:01:43 +0100 Subject: [PATCH] Feature/stats endpoint (#631) * Idea for stats endpoint * Introduced /stats endpoint replacing sending data to metrics server * Removed metrics client * Removed old metrics file * cargo fmt --- Cargo.lock | 45 -- Cargo.toml | 1 - common/client-libs/metrics-client/Cargo.toml | 15 - common/client-libs/metrics-client/src/lib.rs | 74 --- .../metrics-client/src/models/metrics.rs | 39 -- .../metrics-client/src/models/mod.rs | 15 - .../src/requests/metrics_mixes_get.rs | 239 ---------- .../src/requests/metrics_mixes_post.rs | 102 ---- .../metrics-client/src/requests/mod.rs | 16 - mixnode/Cargo.toml | 1 - mixnode/src/commands/mod.rs | 4 - mixnode/src/commands/run.rs | 2 - mixnode/src/commands/upgrade.rs | 10 +- mixnode/src/config/mod.rs | 36 +- mixnode/src/config/template.rs | 3 - mixnode/src/node/http/mod.rs | 1 + mixnode/src/node/http/stats.rs | 29 ++ .../connection_handler/packet_processing.rs | 12 +- mixnode/src/node/metrics.rs | 286 ----------- mixnode/src/node/mod.rs | 48 +- mixnode/src/node/node_statistics.rs | 449 ++++++++++++++++++ mixnode/src/node/packet_delayforwarder.rs | 25 +- 22 files changed, 550 insertions(+), 902 deletions(-) delete mode 100644 common/client-libs/metrics-client/Cargo.toml delete mode 100644 common/client-libs/metrics-client/src/lib.rs delete mode 100644 common/client-libs/metrics-client/src/models/metrics.rs delete mode 100644 common/client-libs/metrics-client/src/models/mod.rs delete mode 100644 common/client-libs/metrics-client/src/requests/metrics_mixes_get.rs delete mode 100644 common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs delete mode 100644 common/client-libs/metrics-client/src/requests/mod.rs create mode 100644 mixnode/src/node/http/stats.rs delete mode 100644 mixnode/src/node/metrics.rs create mode 100644 mixnode/src/node/node_statistics.rs diff --git a/Cargo.lock b/Cargo.lock index a3e1814e16..d96ba293e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -106,16 +106,6 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cff77d8686867eceff3105329d4698d96c2391c176d5d03adc90c7389162b5b8" -[[package]] -name = "assert-json-diff" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50f1c3703dd33532d7f0ca049168930e9099ecac238e23cf932f3a69c42f06da" -dependencies = [ - "serde", - "serde_json", -] - [[package]] name = "async-stream" version = "0.3.0" @@ -701,12 +691,6 @@ dependencies = [ "syn", ] -[[package]] -name = "difference" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "524cbf6897b527295dff137cec09ecf3a05f4fddffd7dfcd1585403449e74198" - [[package]] name = "digest" version = "0.8.1" @@ -1450,16 +1434,6 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8452105ba047068f40ff7093dd1d9da90898e63dd61736462e9cdda6a90ad3c3" -[[package]] -name = "metrics-client" -version = "0.1.0" -dependencies = [ - "mockito", - "reqwest", - "serde", - "tokio", -] - [[package]] name = "mime" version = "0.3.16" @@ -1534,24 +1508,6 @@ dependencies = [ "version-checker", ] -[[package]] -name = "mockito" -version = "0.30.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d10030163d67f681db11810bc486df3149e6d91c8b4f3f96fa8b62b546c2cef8" -dependencies = [ - "assert-json-diff", - "colored", - "difference", - "httparse", - "lazy_static", - "log", - "rand 0.8.3", - "regex", - "serde_json", - "serde_urlencoded", -] - [[package]] name = "multer" version = "1.2.2" @@ -1722,7 +1678,6 @@ dependencies = [ "futures", "humantime-serde", "log", - "metrics-client", "mixnet-client", "mixnode-common", "nonexhaustive-delayqueue", diff --git a/Cargo.toml b/Cargo.toml index 20024c072f..bbccf08d84 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,6 @@ members = [ "clients/webassembly", "clients/client-core", "common/client-libs/gateway-client", - "common/client-libs/metrics-client", "common/client-libs/mixnet-client", "common/client-libs/validator-client", "common/config", diff --git a/common/client-libs/metrics-client/Cargo.toml b/common/client-libs/metrics-client/Cargo.toml deleted file mode 100644 index 690832f8d4..0000000000 --- a/common/client-libs/metrics-client/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "metrics-client" -version = "0.1.0" -authors = ["Jedrzej Stuczynski "] -edition = "2018" - -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html - -[dependencies] -serde = { version = "1.0", features = ["derive"] } -reqwest = { version = "0.11.3", features = ["json"] } - -[dev-dependencies] -mockito = "0.30" -tokio = "1.4" \ No newline at end of file diff --git a/common/client-libs/metrics-client/src/lib.rs b/common/client-libs/metrics-client/src/lib.rs deleted file mode 100644 index a38caeb4fb..0000000000 --- a/common/client-libs/metrics-client/src/lib.rs +++ /dev/null @@ -1,74 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -pub mod models; -pub mod requests; - -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; - -pub struct Config { - pub base_url: String, -} - -impl Config { - pub fn new(base_url: String) -> Self { - Config { base_url } - } -} - -pub struct Client { - base_url: String, - reqwest_client: reqwest::Client, -} - -impl Client { - pub fn new(config: Config) -> Client { - let reqwest_client = reqwest::Client::new(); - Client { - base_url: config.base_url, - reqwest_client, - } - } - - 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 - } - - pub async fn get_mix_metrics(&self) -> reqwest::Result> { - let req = MetricsMixRequest::new(&self.base_url); - self.reqwest_client - .get(&req.url()) - .send() - .await? - .json() - .await - } -} - -#[cfg(test)] -pub(crate) fn client_test_fixture(base_url: &str) -> Client { - Client { - base_url: base_url.to_string(), - reqwest_client: reqwest::Client::new(), - } -} diff --git a/common/client-libs/metrics-client/src/models/metrics.rs b/common/client-libs/metrics-client/src/models/metrics.rs deleted file mode 100644 index 23a4bcd537..0000000000 --- a/common/client-libs/metrics-client/src/models/metrics.rs +++ /dev/null @@ -1,39 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; - -#[derive(Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct PersistedMixMetric { - pub pub_key: String, - pub received: u64, - pub sent: HashMap, - pub timestamp: u64, -} - -#[derive(Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct MixMetric { - pub pub_key: String, - 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/models/mod.rs b/common/client-libs/metrics-client/src/models/mod.rs deleted file mode 100644 index a5e5de4800..0000000000 --- a/common/client-libs/metrics-client/src/models/mod.rs +++ /dev/null @@ -1,15 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -pub mod metrics; diff --git a/common/client-libs/metrics-client/src/requests/metrics_mixes_get.rs b/common/client-libs/metrics-client/src/requests/metrics_mixes_get.rs deleted file mode 100644 index fdc01c4ccd..0000000000 --- a/common/client-libs/metrics-client/src/requests/metrics_mixes_get.rs +++ /dev/null @@ -1,239 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -const PATH: &str = "/api/metrics/mixes"; - -pub struct Request { - base_url: String, - path: String, -} - -impl Request { - pub(crate) fn new(base_url: &str) -> Self { - Request { - base_url: base_url.to_string(), - path: PATH.to_string(), - } - } - - pub(crate) fn url(&self) -> String { - format!("{}{}", self.base_url, self.path) - } -} - -#[cfg(test)] -mod metrics_get_request { - use super::*; - use crate::client_test_fixture; - use mockito::mock; - - #[cfg(test)] - mod on_a_400_status { - use super::*; - - #[tokio::test] - async fn it_returns_an_error() { - let _m = mock("GET", PATH).with_status(400).create(); - let client = client_test_fixture(&mockito::server_url()); - let result = client.get_mix_metrics().await; - assert!(result.is_err()); - _m.assert(); - } - } - - #[cfg(test)] - mod on_a_200 { - use super::*; - #[tokio::test] - async fn it_returns_a_response_with_200_status_and_a_correct_metrics() { - let json = fixtures::mix_metrics_response_json(); - let _m = mock("GET", PATH).with_status(200).with_body(json).create(); - let client = client_test_fixture(&mockito::server_url()); - let result = client.get_mix_metrics().await; - let unwrapped = result.unwrap(); - - assert_eq!(10, unwrapped.first().unwrap().received); - _m.assert(); - } - } - - #[cfg(test)] - mod fixtures { - #[cfg(test)] - pub fn mix_metrics_response_json() -> &'static str { - r#"[ - { - "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=", - "sent": { - "35.178.213.77:1789": 1, - "52.56.99.196:1789": 2 - }, - "received": 10, - "timestamp": 1576061080635800000 - }, - { - "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=", - "sent": { - "3.8.176.11:1789": 2, - "35.178.212.193:1789": 2 - }, - "received": 4, - "timestamp": 1576061080806225700 - }, - { - "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=", - "sent": { - "3.9.12.238:1789": 3 - }, - "received": 3, - "timestamp": 1576061080894667300 - }, - { - "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=", - "sent": { - "3.9.12.238:1789": 5, - "35.176.155.107:1789": 3 - }, - "received": 8, - "timestamp": 1576061081254846500 - }, - { - "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=", - "sent": { - "35.178.213.77:1789": 4, - "52.56.99.196:1789": 6 - }, - "received": 19, - "timestamp": 1576061081371549000 - }, - { - "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=", - "sent": { - "3.8.176.11:1789": 4, - "35.178.212.193:1789": 2 - }, - "received": 6, - "timestamp": 1576061081498404900 - }, - { - "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=", - "sent": { - "35.178.213.77:1789": 2, - "52.56.99.196:1789": 3 - }, - "received": 6, - "timestamp": 1576061081637625000 - }, - { - "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=", - "sent": { - "3.8.176.11:1789": 5, - "35.178.212.193:1789": 4 - }, - "received": 9, - "timestamp": 1576061081805484800 - }, - { - "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=", - "sent": { - "3.9.12.238:1789": 4, - "35.176.155.107:1789": 4 - }, - "received": 8, - "timestamp": 1576061081896562400 - }, - { - "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=", - "sent": { - "3.9.12.238:1789": 2, - "35.176.155.107:1789": 4 - }, - "received": 6, - "timestamp": 1576061079255938600 - }, - { - "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=", - "sent": { - "35.178.213.77:1789": 6 - }, - "received": 10, - "timestamp": 1576061079370829300 - }, - { - "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=", - "sent": { - "3.8.176.11:1789": 2, - "35.178.212.193:1789": 5 - }, - "received": 7, - "timestamp": 1576061079497993200 - }, - { - "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=", - "sent": { - "35.178.213.77:1789": 5, - "52.56.99.196:1789": 2 - }, - "received": 13, - "timestamp": 1576061079637208600 - }, - { - "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=", - "sent": { - "3.8.176.11:1789": 5, - "35.178.212.193:1789": 4 - }, - "received": 9, - "timestamp": 1576061079806557200 - }, - { - "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=", - "sent": { - "3.9.12.238:1789": 2, - "35.176.155.107:1789": 7 - }, - "received": 9, - "timestamp": 1576061079895988000 - }, - { - "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=", - "sent": { - "3.9.12.238:1789": 3, - "35.176.155.107:1789": 2 - }, - "received": 5, - "timestamp": 1576061080255701500 - }, - { - "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=", - "sent": { - "35.178.213.77:1789": 3, - "52.56.99.196:1789": 3 - }, - "received": 7, - "timestamp": 1576061080370956300 - }, - { - "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=", - "sent": { - "3.8.176.11:1789": 5, - "35.178.212.193:1789": 1 - }, - "received": 6, - "timestamp": 1576061080501732900 - } - ]"# - } - } -} 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 deleted file mode 100644 index 1e38f2d46a..0000000000 --- a/common/client-libs/metrics-client/src/requests/metrics_mixes_post.rs +++ /dev/null @@ -1,102 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::models::metrics::MixMetric; - -const PATH: &str = "/api/metrics/mixes"; - -pub struct Request { - base_url: String, - path: String, - payload: MixMetric, -} - -impl Request { - pub(crate) fn url(&self) -> String { - format!("{}{}", self.base_url, self.path) - } - - pub(crate) fn new(base_url: &str, payload: MixMetric) -> Self { - Request { - base_url: base_url.to_string(), - path: PATH.to_string(), - payload, - } - } - - pub(crate) fn json_payload(&self) -> &MixMetric { - &self.payload - } -} - -#[cfg(test)] -mod metrics_post_request { - use super::*; - use crate::client_test_fixture; - use mockito::mock; - - #[cfg(test)] - mod on_a_400_status { - use super::*; - - #[tokio::test] - async fn it_returns_an_error() { - 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!(result.is_err()); - _m.assert(); - } - } - - #[cfg(test)] - mod on_a_200 { - use super::*; - #[tokio::test] - async fn it_returns_a_response_with_200() { - let json = fixtures::metrics_interval_json(); - let _m = mock("POST", "/api/metrics/mixes") - .with_status(201) - .with_body(json) - .create(); - let client = client_test_fixture(&mockito::server_url()); - let result = client.post_mix_metrics(fixtures::new_metric()).await; - assert!(result.is_ok()); - _m.assert(); - } - } - - #[cfg(test)] - mod fixtures { - use crate::models::metrics::MixMetric; - - pub fn new_metric() -> MixMetric { - MixMetric { - pub_key: "abc".to_string(), - received: 666, - sent: Default::default(), - } - } - - #[cfg(test)] - pub fn metrics_interval_json() -> String { - r#" - { - "nextReportIn": 5 - } - "# - .to_string() - } - } -} diff --git a/common/client-libs/metrics-client/src/requests/mod.rs b/common/client-libs/metrics-client/src/requests/mod.rs deleted file mode 100644 index 4fe8f6c3d3..0000000000 --- a/common/client-libs/metrics-client/src/requests/mod.rs +++ /dev/null @@ -1,16 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -pub mod metrics_mixes_get; -pub mod metrics_mixes_post; diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index 417a9832ea..bf1ba20d10 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -30,7 +30,6 @@ toml = "0.5" ## internal config = { path = "../common/config" } crypto = { path = "../common/crypto" } -metrics-client = { path = "../common/client-libs/metrics-client" } mixnet-client = { path = "../common/client-libs/mixnet-client" } mixnode-common = { path = "../common/mixnode-common" } nonexhaustive-delayqueue = { path = "../common/nonexhaustive-delayqueue" } diff --git a/mixnode/src/commands/mod.rs b/mixnode/src/commands/mod.rs index 6c452f4b8f..11b5ad56b4 100644 --- a/mixnode/src/commands/mod.rs +++ b/mixnode/src/commands/mod.rs @@ -52,10 +52,6 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi config = config.with_custom_mixnet_contract(contract_address) } - if let Some(metrics_server) = matches.value_of("metrics-server") { - config = config.with_custom_metrics_server(metrics_server); - } - if let Some(announce_host) = matches.value_of("announce-host") { config = config.with_announce_host(announce_host); } else if was_host_overridden { diff --git a/mixnode/src/commands/run.rs b/mixnode/src/commands/run.rs index 2d05632d51..94d20d6510 100644 --- a/mixnode/src/commands/run.rs +++ b/mixnode/src/commands/run.rs @@ -157,8 +157,6 @@ pub fn execute(matches: &ArgMatches) { "Validator servers: {:?}", config.get_validator_rest_endpoints() ); - println!("Metrics server: {}", config.get_metrics_server()); - println!( "Listening for incoming packets on {}", config.get_listening_address() diff --git a/mixnode/src/commands/upgrade.rs b/mixnode/src/commands/upgrade.rs index 6d03182592..3ab6796e8d 100644 --- a/mixnode/src/commands/upgrade.rs +++ b/mixnode/src/commands/upgrade.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::config::{ - default_validator_rest_endpoints, missing_string_value, Config, DEFAULT_METRICS_SERVER, + default_validator_rest_endpoints, missing_string_value, Config, DEFAULT_MIXNET_CONTRACT_ADDRESS, MISSING_VALUE, }; use clap::{App, Arg, ArgMatches}; @@ -141,12 +141,6 @@ fn pre_090_upgrade(from: &str, config: Config) -> Config { process::exit(1); } - if config.get_metrics_server() != missing_string_value::() { - eprintln!("existing config seems to have specified new metrics-server endpoint which was only introduced in 0.9.0! Can't perform upgrade."); - print_failed_upgrade(&from_version, &to_version); - process::exit(1); - } - if config.get_validator_rest_endpoints()[0] != missing_string_value::() { eprintln!("existing config seems to have specified new validator rest endpoint which was only introduced in 0.9.0! Can't perform upgrade."); print_failed_upgrade(&from_version, &to_version); @@ -155,10 +149,8 @@ fn pre_090_upgrade(from: &str, config: Config) -> Config { let mut upgraded_config = config .with_custom_version(to_version.to_string().as_ref()) - .with_custom_metrics_server(DEFAULT_METRICS_SERVER) .with_custom_validators(default_validator_rest_endpoints()); - println!("Setting metrics server to {}", DEFAULT_METRICS_SERVER); println!( "Setting validator REST endpoints to {:?}", default_validator_rest_endpoints() diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index c0bd0d138a..10d84ed849 100644 --- a/mixnode/src/config/mod.rs +++ b/mixnode/src/config/mod.rs @@ -22,7 +22,6 @@ pub(crate) const DEFAULT_VALIDATOR_REST_ENDPOINTS: &[&str] = &[ "http://testnet-finney-validator2.nymtech.net:1317", "http://mixnet.club:1317", ]; -pub(crate) const DEFAULT_METRICS_SERVER: &str = "http://testnet-metrics.nymtech.net:8080"; pub const DEFAULT_MIXNET_CONTRACT_ADDRESS: &str = "hal1k0jntykt7e4g3y88ltc60czgjuqdy4c9c6gv94"; // 'RTT MEASUREMENT' @@ -35,7 +34,8 @@ const DEFAULT_TESTING_INTERVAL: Duration = Duration::from_secs(60 * 60 * 12); const DEFAULT_RETRY_TIMEOUT: Duration = Duration::from_secs(60 * 30); // 'DEBUG' -const DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY: Duration = Duration::from_millis(60_000); +const DEFAULT_NODE_STATS_LOGGING_DELAY: Duration = Duration::from_millis(60_000); +const DEFAULT_NODE_STATS_UPDATING_DELAY: Duration = Duration::from_millis(30_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); const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500); @@ -152,11 +152,6 @@ impl Config { self } - pub fn with_custom_metrics_server>(mut self, server: S) -> Self { - self.mixnode.metrics_server_url = server.into(); - self - } - pub fn with_listening_host>(mut self, host: S) -> Self { // see if the provided `host` is just an ip address or ip:port let host = host.into(); @@ -265,12 +260,12 @@ impl Config { self.mixnode.mixnet_contract_address.clone() } - pub fn get_metrics_server(&self) -> String { - self.mixnode.metrics_server_url.clone() + pub fn get_node_stats_logging_delay(&self) -> Duration { + self.debug.node_stats_logging_delay } - pub fn get_metrics_running_stats_logging_delay(&self) -> Duration { - self.debug.metrics_running_stats_logging_delay + pub fn get_node_stats_updating_delay(&self) -> Duration { + self.debug.node_stats_updating_delay } pub fn get_layer(&self) -> u64 { @@ -391,10 +386,6 @@ pub struct MixNode { #[serde(default = "missing_string_value")] mixnet_contract_address: String, - /// Metrics server to which the node will be reporting their metrics data. - #[serde(default = "missing_string_value")] - metrics_server_url: String, - /// nym_home_directory specifies absolute path to the home nym MixNodes directory. /// It is expected to use default value and hence .toml file should not redefine this field. nym_root_directory: PathBuf, @@ -434,7 +425,6 @@ impl Default for MixNode { public_sphinx_key_file: Default::default(), validator_rest_urls: default_validator_rest_endpoints(), mixnet_contract_address: DEFAULT_MIXNET_CONTRACT_ADDRESS.to_string(), - metrics_server_url: DEFAULT_METRICS_SERVER.to_string(), nym_root_directory: Config::default_root_directory(), } } @@ -493,12 +483,19 @@ impl Default for RttMeasurement { #[derive(Debug, Deserialize, PartialEq, Serialize)] #[serde(default)] pub struct Debug { - /// Delay between each subsequent running metrics statistics being logged. + /// Delay between each subsequent node statistics being logged to the console #[serde( deserialize_with = "deserialize_duration", serialize_with = "humantime_serde::serialize" )] - metrics_running_stats_logging_delay: Duration, + node_stats_logging_delay: Duration, + + /// Delay between each subsequent node statistics being updated + #[serde( + deserialize_with = "deserialize_duration", + serialize_with = "humantime_serde::serialize" + )] + node_stats_updating_delay: Duration, /// Initial value of an exponential backoff to reconnect to dropped TCP connection when /// forwarding sphinx packets. @@ -537,7 +534,8 @@ pub struct Debug { impl Default for Debug { fn default() -> Self { Debug { - metrics_running_stats_logging_delay: DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY, + node_stats_logging_delay: DEFAULT_NODE_STATS_LOGGING_DELAY, + node_stats_updating_delay: DEFAULT_NODE_STATS_UPDATING_DELAY, packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF, packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF, initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT, diff --git a/mixnode/src/config/template.rs b/mixnode/src/config/template.rs index 32c9b9b4eb..58299e0bba 100644 --- a/mixnode/src/config/template.rs +++ b/mixnode/src/config/template.rs @@ -54,9 +54,6 @@ validator_rest_urls = [ {{/each}} ] -# Metrics server to which the node will be reporting their metrics data. -metrics_server_url = '{{ mixnode.metrics_server_url }}' - # Address of the validator contract managing the network. mixnet_contract_address = '{{ mixnode.mixnet_contract_address }}' diff --git a/mixnode/src/node/http/mod.rs b/mixnode/src/node/http/mod.rs index 13c134b8d1..27eb8c9b1c 100644 --- a/mixnode/src/node/http/mod.rs +++ b/mixnode/src/node/http/mod.rs @@ -1,4 +1,5 @@ pub(crate) mod description; +pub(crate) mod stats; pub(crate) mod verloc; use rocket::Request; diff --git a/mixnode/src/node/http/stats.rs b/mixnode/src/node/http/stats.rs new file mode 100644 index 0000000000..6c03eb1f82 --- /dev/null +++ b/mixnode/src/node/http/stats.rs @@ -0,0 +1,29 @@ +use crate::node::node_statistics::{NodeStats, NodeStatsSimple, NodeStatsWrapper}; +use rocket::State; +use rocket_contrib::json::Json; +use serde::Serialize; + +#[derive(Serialize)] +#[serde(untagged)] +pub(crate) enum NodeStatsResponse { + Full(NodeStats), + Simple(NodeStatsSimple), +} + +/// Returns a running stats of the node. +#[get("/stats?")] +pub(crate) async fn stats( + stats: State<'_, NodeStatsWrapper>, + debug: Option, +) -> Json { + let snapshot_data = stats.clone_data().await; + + // there's no point in returning the entire hashmap of sending destinations in regular mode + if let Some(debug) = debug { + if debug { + return Json(NodeStatsResponse::Full(snapshot_data)); + } + } + + Json(NodeStatsResponse::Simple(snapshot_data.simplify())) +} diff --git a/mixnode/src/node/listener/connection_handler/packet_processing.rs b/mixnode/src/node/listener/connection_handler/packet_processing.rs index df3e7aed33..93fb655dea 100644 --- a/mixnode/src/node/listener/connection_handler/packet_processing.rs +++ b/mixnode/src/node/listener/connection_handler/packet_processing.rs @@ -1,7 +1,7 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::metrics; +use crate::node::node_statistics; use crypto::asymmetric::encryption; use mixnode_common::cached_packet_processor::error::MixProcessingError; use mixnode_common::cached_packet_processor::processor::CachedPacketProcessor; @@ -15,25 +15,25 @@ pub struct PacketProcessor { inner_processor: CachedPacketProcessor, /// Responsible for updating metrics data - metrics_reporter: metrics::MetricsReporter, + node_stats_update_sender: node_statistics::UpdateSender, } impl PacketProcessor { pub(crate) fn new( encryption_key: &encryption::PrivateKey, - metrics_reporter: metrics::MetricsReporter, + node_stats_update_sender: node_statistics::UpdateSender, cache_entry_ttl: Duration, ) -> Self { PacketProcessor { inner_processor: CachedPacketProcessor::new(encryption_key.into(), cache_entry_ttl), - metrics_reporter, + node_stats_update_sender, } } pub(crate) fn clone_without_cache(&self) -> Self { PacketProcessor { inner_processor: self.inner_processor.clone_without_cache(), - metrics_reporter: self.metrics_reporter.clone(), + node_stats_update_sender: self.node_stats_update_sender.clone(), } } @@ -41,7 +41,7 @@ impl PacketProcessor { &self, received: FramedSphinxPacket, ) -> Result { - self.metrics_reporter.report_received(); + self.node_stats_update_sender.report_received(); self.inner_processor.process_received(received) } } diff --git a/mixnode/src/node/metrics.rs b/mixnode/src/node/metrics.rs deleted file mode 100644 index fd01194228..0000000000 --- a/mixnode/src/node/metrics.rs +++ /dev/null @@ -1,286 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use futures::channel::mpsc; -use futures::lock::Mutex; -use futures::StreamExt; -use log::trace; -use std::collections::HashMap; -use std::ops::DerefMut; -use std::sync::atomic::{AtomicU64, Ordering}; -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 { - Sent(String), - Received, -} - -#[derive(Debug, Clone)] -// Note: you should NEVER create more than a single instance of this using 'new()'. -// You should always use .clone() to create additional instances -struct MixMetrics { - inner: Arc, -} - -#[derive(Debug)] -struct MixMetricsInner { - received: AtomicU64, - sent: Mutex, -} - -impl MixMetrics { - pub(crate) fn new() -> Self { - MixMetrics { - inner: Arc::new(MixMetricsInner { - received: AtomicU64::new(0), - sent: Mutex::new(HashMap::new()), - }), - } - } - - fn increment_received_metrics(&mut self) { - self.inner.received.fetch_add(1, Ordering::SeqCst); - } - - async fn increment_sent_metrics(&mut self, destination: String) { - let mut unlocked = self.inner.sent.lock().await; - let receiver_count = unlocked.entry(destination).or_insert(0); - *receiver_count += 1; - } - - async fn acquire_and_reset_metrics(&mut self) -> (u64, SentMetricsMap) { - let mut unlocked = self.inner.sent.lock().await; - let received = self.inner.received.swap(0, Ordering::SeqCst); - - let sent = std::mem::take(unlocked.deref_mut()); - - (received, sent) - } -} - -struct MetricsReceiver { - metrics: MixMetrics, - metrics_rx: mpsc::UnboundedReceiver, -} - -impl MetricsReceiver { - fn new(metrics: MixMetrics, metrics_rx: mpsc::UnboundedReceiver) -> Self { - MetricsReceiver { - metrics, - metrics_rx, - } - } - - fn start(mut self) -> JoinHandle<()> { - tokio::spawn(async move { - while let Some(metrics_data) = self.metrics_rx.next().await { - match metrics_data { - MetricEvent::Received => self.metrics.increment_received_metrics(), - MetricEvent::Sent(destination) => { - self.metrics.increment_sent_metrics(destination).await - } - } - } - }) - } -} - -struct MetricsSender { - metrics: MixMetrics, - #[allow(dead_code)] - metrics_client: metrics_client::Client, - #[allow(dead_code)] - pub_key_str: String, - metrics_informer: MetricsInformer, -} - -impl MetricsSender { - fn new( - metrics: MixMetrics, - metrics_server: String, - pub_key_str: String, - running_logging_delay: Duration, - ) -> Self { - MetricsSender { - metrics, - metrics_client: metrics_client::Client::new(metrics_client::Config::new( - metrics_server, - )), - pub_key_str, - metrics_informer: MetricsInformer::new(running_logging_delay), - } - } - - fn start(mut self) -> JoinHandle<()> { - tokio::spawn(async move { - loop { - 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(); - - tokio::time::sleep(METRICS_FAILURE_BACKOFF).await; - - // let sending_delay = match self - // .metrics_client - // .post_mix_metrics(MixMetric { - // pub_key: self.pub_key_str.clone(), - // received, - // sent, - // }) - // .await - // { - // Err(err) => { - // error!("failed to send metrics - {:?}", err); - // tokio::time::sleep(METRICS_FAILURE_BACKOFF) - // } - // Ok(new_interval) => { - // debug!("sent metrics information"); - // tokio::time::sleep(Duration::from_secs(new_interval.next_report_in)) - // } - // }; - // - // // wait for however much is left - // sending_delay.await; - } - }) - } -} - -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, -} - -impl MetricsReporter { - pub(crate) fn new(metrics_tx: mpsc::UnboundedSender) -> Self { - MetricsReporter { metrics_tx } - } - - pub(crate) fn report_sent(&self, destination: String) { - // in unbounded_send() failed it means that the receiver channel was disconnected - // and hence something weird must have happened without a way of recovering - self.metrics_tx - .unbounded_send(MetricEvent::Sent(destination)) - .unwrap() - } - - // TODO: in the future this could be slightly optimised to get rid of the channel - // in favour of incrementing value directly - pub(crate) fn report_received(&self) { - // in unbounded_send() failed it means that the receiver channel was disconnected - // and hence something weird must have happened without a way of recovering - self.metrics_tx - .unbounded_send(MetricEvent::Received) - .unwrap() - } -} - -// basically an easy single entry point to start all metrics related tasks -pub struct MetricsController { - receiver: MetricsReceiver, - reporter: MetricsReporter, - sender: MetricsSender, -} - -impl MetricsController { - pub(crate) fn new( - directory_server: String, - pub_key_str: String, - running_stats_logging_delay: Duration, - ) -> Self { - let (metrics_tx, metrics_rx) = mpsc::unbounded(); - let shared_metrics = MixMetrics::new(); - - MetricsController { - sender: MetricsSender::new( - shared_metrics.clone(), - directory_server, - pub_key_str, - running_stats_logging_delay, - ), - receiver: MetricsReceiver::new(shared_metrics, metrics_rx), - reporter: MetricsReporter::new(metrics_tx), - } - } - - // reporter is how node is going to be accessing the metrics data - pub(crate) fn start(self) -> MetricsReporter { - // TODO: should we do anything with JoinHandle(s) returned by start methods? - self.receiver.start(); - self.sender.start(); - self.reporter - } -} diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 27ef694950..8582c15d0d 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -5,12 +5,14 @@ use crate::config::Config; use crate::node::http::{ description::description, not_found, + stats::stats, verloc::{verloc, VerlocState}, }; use crate::node::listener::connection_handler::packet_processing::PacketProcessor; use crate::node::listener::connection_handler::ConnectionHandler; use crate::node::listener::Listener; use crate::node::node_description::NodeDescription; +use crate::node::node_statistics::NodeStatsWrapper; use crate::node::packet_delayforwarder::{DelayForwarder, PacketDelayForwardSender}; use crypto::asymmetric::{encryption, identity}; use log::{error, info, warn}; @@ -22,8 +24,9 @@ use version_checker::parse_version; pub(crate) mod http; mod listener; -mod metrics; +// mod metrics; pub(crate) mod node_description; +pub(crate) mod node_statistics; pub(crate) mod packet_delayforwarder; // the MixNode will live for whole duration of this program @@ -49,7 +52,11 @@ impl MixNode { } } - fn start_http_api(&self, atomic_verloc_result: AtomicVerlocResult) { + fn start_http_api( + &self, + atomic_verloc_result: AtomicVerlocResult, + node_stats_pointer: NodeStatsWrapper, + ) { info!("Starting HTTP API on http://localhost:8000"); let mut config = rocket::config::Config::release_default(); @@ -62,35 +69,38 @@ impl MixNode { tokio::spawn(async move { rocket::build() .configure(config) - .mount("/", routes![verloc, description]) + .mount("/", routes![verloc, description, stats]) .register("/", catchers![not_found]) .manage(verloc_state) .manage(descriptor) + .manage(node_stats_pointer) .launch() .await }); } - fn start_metrics_reporter(&self) -> metrics::MetricsReporter { - info!("Starting metrics reporter..."); - metrics::MetricsController::new( - self.config.get_metrics_server(), - self.identity_keypair.public_key().to_base58_string(), - self.config.get_metrics_running_stats_logging_delay(), - ) - .start() + fn start_node_stats_controller(&self) -> (NodeStatsWrapper, node_statistics::UpdateSender) { + info!("Starting node stats controller..."); + let controller = node_statistics::Controller::new( + self.config.get_node_stats_logging_delay(), + self.config.get_node_stats_updating_delay(), + ); + let node_stats_pointer = controller.get_node_stats_data_pointer(); + let update_sender = controller.start(); + + (node_stats_pointer, update_sender) } fn start_socket_listener( &self, - metrics_reporter: metrics::MetricsReporter, + node_stats_update_sender: node_statistics::UpdateSender, delay_forwarding_channel: PacketDelayForwardSender, ) { info!("Starting socket listener..."); let packet_processor = PacketProcessor::new( self.sphinx_keypair.private_key(), - metrics_reporter, + node_stats_update_sender, self.config.get_cache_entry_ttl(), ); @@ -103,7 +113,7 @@ impl MixNode { fn start_packet_delay_forwarder( &mut self, - metrics_reporter: metrics::MetricsReporter, + node_stats_update_sender: node_statistics::UpdateSender, ) -> PacketDelayForwardSender { info!("Starting packet delay-forwarder..."); @@ -112,7 +122,7 @@ impl MixNode { self.config.get_packet_forwarding_maximum_backoff(), self.config.get_initial_connection_timeout(), self.config.get_maximum_connection_buffer_size(), - metrics_reporter, + node_stats_update_sender, ); let packet_sender = packet_forwarder.sender(); @@ -212,12 +222,12 @@ impl MixNode { } } - let metrics_reporter = self.start_metrics_reporter(); - let delay_forwarding_channel = self.start_packet_delay_forwarder(metrics_reporter.clone()); - self.start_socket_listener(metrics_reporter, delay_forwarding_channel); + let (node_stats_pointer, node_stats_update_sender) = self.start_node_stats_controller(); + let delay_forwarding_channel = self.start_packet_delay_forwarder(node_stats_update_sender.clone()); + self.start_socket_listener(node_stats_update_sender, delay_forwarding_channel); let atomic_verloc_results= self.start_rtt_measurer(); - self.start_http_api(atomic_verloc_results); + self.start_http_api(atomic_verloc_results, node_stats_pointer); info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!"); self.wait_for_interrupt().await diff --git a/mixnode/src/node/node_statistics.rs b/mixnode/src/node/node_statistics.rs new file mode 100644 index 0000000000..41096c0e3e --- /dev/null +++ b/mixnode/src/node/node_statistics.rs @@ -0,0 +1,449 @@ +use futures::channel::mpsc; +use futures::lock::Mutex; +use futures::StreamExt; +use serde::Serialize; +use std::collections::HashMap; +use std::ops::DerefMut; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; +use tokio::sync::{RwLock, RwLockReadGuard}; + +// convenience aliases +type PacketsMap = HashMap; +type PacketDataReceiver = mpsc::UnboundedReceiver; +type PacketDataSender = mpsc::UnboundedSender; + +#[derive(Clone)] +pub(crate) struct NodeStatsWrapper { + inner: Arc>, +} + +impl NodeStatsWrapper { + pub(crate) fn new() -> Self { + let now = SystemTime::now(); + NodeStatsWrapper { + inner: Arc::new(RwLock::new(NodeStats { + update_time: now, + previous_update_time: now, + packets_received_since_startup: 0, + packets_sent_since_startup: HashMap::new(), + packets_explicitly_dropped_since_startup: HashMap::new(), + packets_received_since_last_update: 0, + packets_sent_since_last_update: HashMap::new(), + packets_explicitly_dropped_since_last_update: HashMap::new(), + })), + } + } + + pub(crate) async fn update( + &self, + new_received: u64, + new_sent: PacketsMap, + new_dropped: PacketsMap, + ) { + let mut guard = self.inner.write().await; + let snapshot_time = SystemTime::now(); + + guard.previous_update_time = guard.update_time; + guard.update_time = snapshot_time; + + guard.packets_received_since_startup += new_received; + for (mix, count) in new_sent.iter() { + *guard + .packets_sent_since_startup + .entry(mix.clone()) + .or_insert(0) += *count; + } + + for (mix, count) in new_dropped.iter() { + *guard + .packets_explicitly_dropped_since_last_update + .entry(mix.clone()) + .or_insert(0) += *count; + } + + guard.packets_received_since_last_update = new_received; + guard.packets_sent_since_last_update = new_sent; + guard.packets_explicitly_dropped_since_last_update = new_dropped; + } + + pub(crate) async fn clone_data(&self) -> NodeStats { + self.inner.read().await.clone() + } + + async fn read(&self) -> RwLockReadGuard<'_, NodeStats> { + self.inner.read().await + } +} + +#[derive(Serialize, Clone)] +pub(crate) struct NodeStats { + #[serde(serialize_with = "humantime_serde::serialize")] + update_time: SystemTime, + + #[serde(serialize_with = "humantime_serde::serialize")] + previous_update_time: SystemTime, + + packets_received_since_startup: u64, + + // note: sent does not imply forwarded. We don't know if it was delivered successfully + packets_sent_since_startup: PacketsMap, + + // we know for sure we dropped packets to those destinations + packets_explicitly_dropped_since_startup: PacketsMap, + + packets_received_since_last_update: u64, + + // note: sent does not imply forwarded. We don't know if it was delivered successfully + packets_sent_since_last_update: PacketsMap, + + // we know for sure we dropped packets to those destinations + packets_explicitly_dropped_since_last_update: PacketsMap, +} + +impl NodeStats { + pub(crate) fn simplify(&self) -> NodeStatsSimple { + NodeStatsSimple { + update_time: self.update_time, + previous_update_time: self.previous_update_time, + packets_received_since_startup: self.packets_received_since_startup, + packets_sent_since_startup: self.packets_sent_since_startup.values().sum(), + packets_explicitly_dropped_since_startup: self + .packets_explicitly_dropped_since_startup + .values() + .sum(), + packets_received_since_last_update: self.packets_received_since_last_update, + packets_sent_since_last_update: self.packets_sent_since_last_update.values().sum(), + packets_explicitly_dropped_since_last_update: self + .packets_explicitly_dropped_since_last_update + .values() + .sum(), + } + } +} + +#[derive(Serialize, Clone)] +pub(crate) struct NodeStatsSimple { + #[serde(serialize_with = "humantime_serde::serialize")] + update_time: SystemTime, + + #[serde(serialize_with = "humantime_serde::serialize")] + previous_update_time: SystemTime, + + packets_received_since_startup: u64, + + // note: sent does not imply forwarded. We don't know if it was delivered successfully + packets_sent_since_startup: u64, + + // we know for sure we dropped those packets + packets_explicitly_dropped_since_startup: u64, + + packets_received_since_last_update: u64, + + // note: sent does not imply forwarded. We don't know if it was delivered successfully + packets_sent_since_last_update: u64, + + // we know for sure we dropped those packets + packets_explicitly_dropped_since_last_update: u64, +} + +pub(crate) enum PacketEvent { + Sent(String), + Received, + Dropped(String), +} + +#[derive(Debug, Clone)] +struct CurrentPacketData { + inner: Arc, +} + +#[derive(Debug)] +struct PacketDataInner { + received: AtomicU64, + sent: Mutex, + dropped: Mutex, +} + +impl CurrentPacketData { + pub(crate) fn new() -> Self { + CurrentPacketData { + inner: Arc::new(PacketDataInner { + received: AtomicU64::new(0), + sent: Mutex::new(HashMap::new()), + dropped: Mutex::new(HashMap::new()), + }), + } + } + + fn increment_received(&self) { + self.inner.received.fetch_add(1, Ordering::SeqCst); + } + + async fn increment_sent(&self, destination: String) { + let mut unlocked = self.inner.sent.lock().await; + let receiver_count = unlocked.entry(destination).or_insert(0); + *receiver_count += 1; + } + + async fn increment_dropped(&self, destination: String) { + let mut unlocked = self.inner.dropped.lock().await; + let dropped_count = unlocked.entry(destination).or_insert(0); + *dropped_count += 1; + } + + async fn acquire_and_reset(&self) -> (u64, PacketsMap, PacketsMap) { + let mut unlocked_sent = self.inner.sent.lock().await; + let mut unlocked_dropped = self.inner.dropped.lock().await; + let received = self.inner.received.swap(0, Ordering::SeqCst); + + let sent = std::mem::take(unlocked_sent.deref_mut()); + let dropped = std::mem::take(unlocked_dropped.deref_mut()); + + (received, sent, dropped) + } +} + +struct UpdateHandler { + current_data: CurrentPacketData, + update_receiver: PacketDataReceiver, +} + +impl UpdateHandler { + fn new(current_data: CurrentPacketData, update_receiver: PacketDataReceiver) -> Self { + UpdateHandler { + current_data, + update_receiver, + } + } + + async fn run(&mut self) { + while let Some(packet_data) = self.update_receiver.next().await { + match packet_data { + PacketEvent::Received => self.current_data.increment_received(), + PacketEvent::Sent(destination) => { + self.current_data.increment_sent(destination).await + } + PacketEvent::Dropped(destination) => { + self.current_data.increment_dropped(destination).await + } + } + } + } +} + +#[derive(Clone)] +pub struct UpdateSender(PacketDataSender); + +impl UpdateSender { + pub(crate) fn new(update_sender: PacketDataSender) -> Self { + UpdateSender(update_sender) + } + + pub(crate) fn report_sent(&self, destination: String) { + // in unbounded_send() failed it means that the receiver channel was disconnected + // and hence something weird must have happened without a way of recovering + self.0 + .unbounded_send(PacketEvent::Sent(destination)) + .unwrap() + } + + // TODO: in the future this could be slightly optimised to get rid of the channel + // in favour of incrementing value directly + pub(crate) fn report_received(&self) { + // in unbounded_send() failed it means that the receiver channel was disconnected + // and hence something weird must have happened without a way of recovering + self.0.unbounded_send(PacketEvent::Received).unwrap() + } + + pub(crate) fn report_dropped(&self, destination: String) { + // in unbounded_send() failed it means that the receiver channel was disconnected + // and hence something weird must have happened without a way of recovering + self.0 + .unbounded_send(PacketEvent::Dropped(destination)) + .unwrap() + } +} + +struct StatsUpdater { + updating_delay: Duration, + current_packet_data: CurrentPacketData, + current_stats: NodeStatsWrapper, +} + +impl StatsUpdater { + fn new( + updating_delay: Duration, + current_packet_data: CurrentPacketData, + current_stats: NodeStatsWrapper, + ) -> Self { + StatsUpdater { + updating_delay, + current_packet_data, + current_stats, + } + } + + async fn update_stats(&self) { + // grab new data since last update + let (received, sent, dropped) = self.current_packet_data.acquire_and_reset().await; + self.current_stats.update(received, sent, dropped).await; + } + + async fn run(&self) { + loop { + tokio::time::sleep(self.updating_delay).await; + self.update_stats().await + } + } +} + +// TODO: question: should this data still be logged to the console or should we perhaps remove it +// since we have the http endpoint now? +struct PacketStatsConsoleLogger { + logging_delay: Duration, + stats: NodeStatsWrapper, +} + +impl PacketStatsConsoleLogger { + fn new(logging_delay: Duration, stats: NodeStatsWrapper) -> Self { + PacketStatsConsoleLogger { + logging_delay, + stats, + } + } + + async fn log_running_stats(&mut self) { + let stats = self.stats.read().await; + + // it's super unlikely this will ever fail, but anything involving time is super weird + // so let's just guard against it + if let Ok(time_difference) = stats.update_time.duration_since(stats.previous_update_time) { + // we honestly don't care if it was 30.000828427s or 30.002461449s, 30s is enough + let difference_secs = time_difference.as_secs(); + + info!( + "Since startup mixed {} packets! ({} in last {} seconds)", + stats.packets_sent_since_startup.values().sum::(), + stats.packets_sent_since_last_update.values().sum::(), + difference_secs, + ); + if !stats.packets_explicitly_dropped_since_startup.is_empty() { + info!( + "Since startup dropped {} packets! ({} in last {} seconds)", + stats + .packets_explicitly_dropped_since_startup + .values() + .sum::(), + stats + .packets_explicitly_dropped_since_last_update + .values() + .sum::(), + difference_secs, + ); + } + + debug!( + "Since startup received {} packets ({} in last {} seconds)", + stats.packets_received_since_startup, + stats.packets_received_since_last_update, + difference_secs, + ); + trace!( + "Since startup sent packets to the following: \n{:#?} \n And in last {} seconds: {:#?})", + stats.packets_sent_since_startup, + difference_secs, + stats.packets_sent_since_last_update + ); + } else { + info!( + "Since startup mixed {} packets!", + stats.packets_sent_since_startup.values().sum::(), + ); + if !stats.packets_explicitly_dropped_since_startup.is_empty() { + info!( + "Since startup dropped {} packets!", + stats + .packets_explicitly_dropped_since_startup + .values() + .sum::(), + ); + } + + debug!( + "Since startup received {} packets", + stats.packets_received_since_startup + ); + trace!( + "Since startup sent packets to the following: \n{:#?}", + stats.packets_sent_since_startup + ); + } + } + + async fn run(&mut self) { + loop { + tokio::time::sleep(self.logging_delay).await; + self.log_running_stats().await; + } + } +} + +// basically an easy single entry point to start all of the required tasks +pub struct Controller { + /// Responsible for handling data coming from UpdateSender + update_handler: UpdateHandler, + + /// Wrapper around channel sending information about new packet being received or sent + update_sender: UpdateSender, + + /// Responsible for logging stats to the console at given interval + console_logger: PacketStatsConsoleLogger, + + /// Responsible for updating stats at given interval + stats_updater: StatsUpdater, + + /// Pointer to the current node stats + node_stats: NodeStatsWrapper, +} + +impl Controller { + pub(crate) fn new(logging_delay: Duration, stats_updating_delay: Duration) -> Self { + let (sender, receiver) = mpsc::unbounded(); + let shared_packet_data = CurrentPacketData::new(); + let shared_node_stats = NodeStatsWrapper::new(); + + Controller { + update_handler: UpdateHandler::new(shared_packet_data.clone(), receiver), + update_sender: UpdateSender::new(sender), + console_logger: PacketStatsConsoleLogger::new(logging_delay, shared_node_stats.clone()), + stats_updater: StatsUpdater::new( + stats_updating_delay, + shared_packet_data, + shared_node_stats.clone(), + ), + node_stats: shared_node_stats, + } + } + + pub(crate) fn get_node_stats_data_pointer(&self) -> NodeStatsWrapper { + NodeStatsWrapper { + inner: Arc::clone(&self.node_stats.inner), + } + } + + // reporter is how node is going to be accessing the metrics data + pub(crate) fn start(self) -> UpdateSender { + // move out of self + let mut update_handler = self.update_handler; + let stats_updater = self.stats_updater; + let mut console_logger = self.console_logger; + + tokio::spawn(async move { update_handler.run().await }); + tokio::spawn(async move { stats_updater.run().await }); + tokio::spawn(async move { console_logger.run().await }); + + self.update_sender + } +} diff --git a/mixnode/src/node/packet_delayforwarder.rs b/mixnode/src/node/packet_delayforwarder.rs index 89c37e6e6c..ee469f3eb8 100644 --- a/mixnode/src/node/packet_delayforwarder.rs +++ b/mixnode/src/node/packet_delayforwarder.rs @@ -1,12 +1,12 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::metrics::MetricsReporter; +use crate::node::node_statistics::UpdateSender; use futures::channel::mpsc; use futures::StreamExt; -use log::debug; use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, TimerError}; use nymsphinx::forwarding::packet::MixPacket; +use std::io; use tokio::time::{Duration, Instant}; // Delay + MixPacket vs Instant + MixPacket @@ -22,7 +22,7 @@ pub(crate) struct DelayForwarder { mixnet_client: mixnet_client::Client, packet_sender: PacketDelayForwardSender, packet_receiver: PacketDelayForwardReceiver, - metrics_reporter: MetricsReporter, + node_stats_update_sender: UpdateSender, } impl DelayForwarder { @@ -31,7 +31,7 @@ impl DelayForwarder { maximum_reconnection_backoff: Duration, initial_connection_timeout: Duration, maximum_connection_buffer_size: usize, - metrics_reporter: MetricsReporter, + node_stats_update_sender: UpdateSender, ) -> Self { let client_config = mixnet_client::Config::new( initial_reconnection_backoff, @@ -47,7 +47,7 @@ impl DelayForwarder { mixnet_client: mixnet_client::Client::new(client_config), packet_sender, packet_receiver, - metrics_reporter, + node_stats_update_sender, } } @@ -64,9 +64,20 @@ impl DelayForwarder { self.mixnet_client .send_without_response(next_hop, sphinx_packet, packet_mode) { - debug!("failed to forward the packet to {} - {}", next_hop, err) + if err.kind() == io::ErrorKind::WouldBlock { + // we only know for sure if we dropped a packet if our sending queue was full + // in any other case the connection might still be re-established (or created for the first time) + // and the packet might get sent, but we won't know about it + self.node_stats_update_sender + .report_dropped(next_hop.to_string()) + } else if err.kind() == io::ErrorKind::NotConnected { + // let's give the benefit of the doubt and assume we manage to establish connection + self.node_stats_update_sender + .report_sent(next_hop.to_string()); + } } else { - self.metrics_reporter.report_sent(next_hop.to_string()); + self.node_stats_update_sender + .report_sent(next_hop.to_string()); } }