From fbdf31b8794643dde2f78d02eaed7235fa108584 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Thu, 9 Jun 2022 16:58:51 +0300 Subject: [PATCH] Feature/split stats service (#1328) * Replace client address with service address * Copy over the server parts of the statistics server * Separate API in module * Rename struct * Add insertion endpoint * Remove unused code from network requester * Box big rocket error * Remove the feature-specific code * Construct http req * Clippy + Cargo.lock update * Re-added needed sqlx feature * Wrap http req in ordered msg * Add http headers * Move common api functionality into the separate crate * Make stats server address configurable, especially for testing * Update changelog * Fix clippy * Help command update --- CHANGELOG.md | 4 +- Cargo.lock | 29 +++- Cargo.toml | 2 + common/socks5/requests/src/msg.rs | 20 +++ common/statistics/Cargo.toml | 15 ++ common/statistics/src/api.rs | 40 +++++ common/statistics/src/error.rs | 10 ++ common/statistics/src/lib.rs | 43 +++++ .../network-requester/Cargo.toml | 16 +- service-providers/network-requester/README.md | 9 +- .../network-requester/src/connection.rs | 9 +- .../network-requester/src/core.rs | 149 ++++++++--------- .../network-requester/src/main.rs | 33 ++-- .../network-requester/src/statistics/comm.rs | 150 +++++++++--------- .../network-requester/src/statistics/error.rs | 3 - .../network-requester/src/statistics/mod.rs | 2 +- .../network-requester/src/storage/routes.rs | 52 ------ .../network-statistics/Cargo.toml | 23 +++ .../network-statistics/Rocket.toml | 7 + .../build.rs | 2 +- .../20220512120000_mixnet_statistics.sql | 5 +- .../network-statistics/src/api/error.rs | 31 ++++ .../network-statistics/src/api/mod.rs | 48 ++++++ .../network-statistics/src/api/routes.rs | 61 +++++++ .../network-statistics/src/main.rs | 53 +++++++ .../src/storage/error.rs | 5 +- .../src/storage/manager.rs | 19 +-- .../src/storage/mod.rs | 42 ++--- .../src/storage/models.rs | 5 +- 29 files changed, 590 insertions(+), 297 deletions(-) create mode 100644 common/statistics/Cargo.toml create mode 100644 common/statistics/src/api.rs create mode 100644 common/statistics/src/error.rs create mode 100644 common/statistics/src/lib.rs delete mode 100644 service-providers/network-requester/src/storage/routes.rs create mode 100644 service-providers/network-statistics/Cargo.toml create mode 100644 service-providers/network-statistics/Rocket.toml rename service-providers/{network-requester => network-statistics}/build.rs (91%) rename service-providers/{network-requester => network-statistics}/migrations/20220512120000_mixnet_statistics.sql (74%) create mode 100644 service-providers/network-statistics/src/api/error.rs create mode 100644 service-providers/network-statistics/src/api/mod.rs create mode 100644 service-providers/network-statistics/src/api/routes.rs create mode 100644 service-providers/network-statistics/src/main.rs rename service-providers/{network-requester => network-statistics}/src/storage/error.rs (76%) rename service-providers/{network-requester => network-statistics}/src/storage/manager.rs (70%) rename service-providers/{network-requester => network-statistics}/src/storage/mod.rs (66%) rename service-providers/{network-requester => network-statistics}/src/storage/models.rs (78%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dd00a5c89..7d5127b964 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,6 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// - gateway: Added gateway coconut verifications and validator-api communication for double spending protection ([#1261]) - mixnet-contract: Added ClaimOperatorReward and ClaimDelegatorReward messages ([#1292]) - mixnet-contract: Replace all naked `-` with `saturating_sub`. -- network-requester: send traffic statistics from all network requesters and receive it in a special network-requester that aggregates the data and exposes it via a rest API ([#1267], [#1278]). - validator-api: add `estimated_node_profit` and `estimated_operator_cost` to `reward-estimate` endpoint ([#1284]) - validator-api: add detailed mixnode bond endpoints, and explorer-api makes use of that data to append stake saturation. - validator-api: add Swagger to document the REST API ([#1249]). @@ -22,6 +21,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// - wallet: compound and claim reward endpoints for operators and delegators ([#1302]) - wallet: require password to switch accounts - wallet: the wallet backend learned how to keep track of validator name, either hardcoded or by querying the status endpoint. +- network-statistics: a new mixnet service that aggregates and exposes anonymized data about mixnet services ([#1328]) ### Fixed @@ -39,6 +39,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// - validator-client: created internal `Coin` type that replaces coins from `cosmrs` and `cosmwasm` for API entrypoints [[#1295]] - all: updated all `cosmwasm`-related dependencies to `1.0.0` and `cw-storage-plus` to `0.13.4` [[#1318]] +- network-requester: allow to voluntarily store and send statistical data about the number of bytes the proxied server serves ([#1328]) [#1249]: https://github.com/nymtech/nym/pull/1249 [#1256]: https://github.com/nymtech/nym/pull/1256 @@ -56,6 +57,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// [#1302]: https://github.com/nymtech/nym/pull/1302 [#1318]: https://github.com/nymtech/nym/pull/1318 [#1322]: https://github.com/nymtech/nym/pull/1322 +[#1328]: https://github.com/nymtech/nym/pull/1328 ## [nym-wallet-v1.0.4](https://github.com/nymtech/nym/tree/nym-wallet-v1.0.4) (2022-05-04) diff --git a/Cargo.lock b/Cargo.lock index 363273a699..d800d74f15 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3129,7 +3129,6 @@ dependencies = [ name = "nym-network-requester" version = "1.0.1" dependencies = [ - "bincode", "clap 2.34.0", "dirs", "futures", @@ -3143,16 +3142,32 @@ dependencies = [ "publicsuffix", "rand 0.7.3", "reqwest", - "rocket", "serde", "socks5-requests", "sqlx", + "statistics", "thiserror", "tokio", "tokio-tungstenite", "websocket-requests", ] +[[package]] +name = "nym-network-statistics" +version = "0.1.0" +dependencies = [ + "dirs", + "log", + "pretty_env_logger", + "rocket", + "serde", + "sqlx", + "statistics", + "thiserror", + "tokio", + "tokio-tungstenite", +] + [[package]] name = "nym-socks5-client" version = "1.0.1" @@ -5303,6 +5318,16 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "statistics" +version = "1.0.1" +dependencies = [ + "reqwest", + "serde", + "serde_json", + "thiserror", +] + [[package]] name = "stdweb" version = "0.4.20" diff --git a/Cargo.toml b/Cargo.toml index d1acbb02a1..4ffc4cbdd2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ members = [ "common/nymsphinx/params", "common/nymsphinx/types", "common/pemstore", + "common/statistics", "common/socks5/proxy-helpers", "common/socks5/requests", "common/task", @@ -63,6 +64,7 @@ members = [ "gateway/gateway-requests", "mixnode", "service-providers/network-requester", + "service-providers/network-statistics", "validator-api", "validator-api/validator-api-requests", ] diff --git a/common/socks5/requests/src/msg.rs b/common/socks5/requests/src/msg.rs index 21a37b1190..68d1160b4e 100644 --- a/common/socks5/requests/src/msg.rs +++ b/common/socks5/requests/src/msg.rs @@ -32,6 +32,26 @@ impl Message { const REQUEST_FLAG: u8 = 0; const RESPONSE_FLAG: u8 = 1; + pub fn conn_id(&self) -> u64 { + match self { + Message::Request(req) => match req { + Request::Connect(c) => c.conn_id, + Request::Send(conn_id, _, _) => *conn_id, + }, + Message::Response(resp) => resp.connection_id, + } + } + + pub fn size(&self) -> usize { + match self { + Message::Request(req) => match req { + Request::Connect(_) => 0, + Request::Send(_, data, _) => data.len(), + }, + Message::Response(resp) => resp.data.len(), + } + } + pub fn try_from_bytes(b: &[u8]) -> Result { if b.is_empty() { return Err(MessageError::NoData); diff --git a/common/statistics/Cargo.toml b/common/statistics/Cargo.toml new file mode 100644 index 0000000000..efed3428a5 --- /dev/null +++ b/common/statistics/Cargo.toml @@ -0,0 +1,15 @@ +# Copyright 2022 - Nym Technologies SA +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "statistics" +version = "1.0.1" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +reqwest = "0.11" +serde = { version = "1.0", features = ["derive"] } +serde_json = "1" +thiserror = "1" \ No newline at end of file diff --git a/common/statistics/src/api.rs b/common/statistics/src/api.rs new file mode 100644 index 0000000000..6b9139adba --- /dev/null +++ b/common/statistics/src/api.rs @@ -0,0 +1,40 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::StatsError; +use crate::StatsMessage; + +pub const DEFAULT_STATISTICS_SERVICE_ADDRESS: &str = "127.0.0.1"; +pub const DEFAULT_STATISTICS_SERVICE_PORT: u16 = 8090; + +pub const STATISTICS_SERVICE_VERSION: &str = "/v1"; +pub const STATISTICS_SERVICE_API_STATISTICS: &str = "statistic"; + +pub fn build_statistics_request_bytes(msg: StatsMessage) -> Result, StatsError> { + let json_msg = msg.to_json()?; + + let req = reqwest::Request::new( + reqwest::Method::POST, + reqwest::Url::parse(&format!( + "http://{}:{}/{}/{}", + DEFAULT_STATISTICS_SERVICE_ADDRESS, + DEFAULT_STATISTICS_SERVICE_PORT, + STATISTICS_SERVICE_VERSION, + STATISTICS_SERVICE_API_STATISTICS + )) + .unwrap(), + ); + let data = format!( + "{} {} {:?}\n\ + Content-Type: application/json\n\ + Content-Length: {}\n\n\ + {}\n", + req.method().as_str(), + req.url().as_str(), + req.version(), + json_msg.len(), + json_msg + ); + + Ok(data.into_bytes()) +} diff --git a/common/statistics/src/error.rs b/common/statistics/src/error.rs new file mode 100644 index 0000000000..1c25807d98 --- /dev/null +++ b/common/statistics/src/error.rs @@ -0,0 +1,10 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum StatsError { + #[error("Serde JSON error: {0}")] + SerdeJsonError(#[from] serde_json::Error), +} diff --git a/common/statistics/src/lib.rs b/common/statistics/src/lib.rs new file mode 100644 index 0000000000..267ccafb83 --- /dev/null +++ b/common/statistics/src/lib.rs @@ -0,0 +1,43 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use serde::{Deserialize, Serialize}; + +use error::StatsError; + +pub mod api; +pub mod error; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct StatsMessage { + pub stats_data: Vec, + pub interval_seconds: u32, + pub timestamp: String, +} + +impl StatsMessage { + pub fn to_json(&self) -> Result { + Ok(serde_json::to_string(self)?) + } + + pub fn from_json(s: &str) -> Result { + Ok(serde_json::from_str(s)?) + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct StatsServiceData { + pub requested_service: String, + pub request_bytes: u32, + pub response_bytes: u32, +} + +impl StatsServiceData { + pub fn new(requested_service: String, request_bytes: u32, response_bytes: u32) -> Self { + StatsServiceData { + requested_service, + request_bytes, + response_bytes, + } + } +} diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index ba83a25179..b4825bb782 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -10,7 +10,6 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -bincode = "1.3" clap = "2.33.0" dirs = "3.0" futures = "0.3" @@ -18,10 +17,10 @@ ipnetwork = "0.17" log = "0.4" pretty_env_logger = "0.4" publicsuffix = "1.5" +rand = "0.7" reqwest = { version = "0.11", features = ["json"] } -rocket = { version = "0.5.0-rc.1", features = ["json"], optional = true } serde = { version = "1.0", features = ["derive"] } -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]} +sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "chrono"]} thiserror = "1" tokio = { version = "1.19.1", features = [ "net", "rt-multi-thread", "macros", "time" ] } tokio-tungstenite = "0.14" @@ -33,14 +32,5 @@ nymsphinx = { path = "../../common/nymsphinx" } ordered-buffer = {path = "../../common/socks5/ordered-buffer"} proxy-helpers = { path = "../../common/socks5/proxy-helpers" } socks5-requests = { path = "../../common/socks5/requests" } +statistics = { path = "../../common/statistics" } websocket-requests = { path = "../../clients/native/websocket-requests" } - -[dev-dependencies] -rand = "0.7" - -[build-dependencies] -sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } -tokio = { version = "1.19.1", features = ["rt-multi-thread", "macros"] } - -[features] -stats-service = ["rocket"] \ No newline at end of file diff --git a/service-providers/network-requester/README.md b/service-providers/network-requester/README.md index a5006561ef..fb414d7e75 100644 --- a/service-providers/network-requester/README.md +++ b/service-providers/network-requester/README.md @@ -19,8 +19,7 @@ Running in `open-proxy` mode allows any traffic to be proxied by the network requester. ### Statistics service -The network requester can be build and ran as a gatherer of statistics from all -the other network requesters on the mixnet. For that, build the binary with the -`stats-service` feature enabled. The native client address that corresponds to -this network requester would have to be built into the constants of all the -other network requesters that are sending the data. \ No newline at end of file +The network requester can be ran as a gatherer of statistics for all +the services it proxies. For that, run the binary with the +`enable-statistics` flag enabled. Anonymized statistics are then sent to +a central server, through the mixnet. \ No newline at end of file diff --git a/service-providers/network-requester/src/connection.rs b/service-providers/network-requester/src/connection.rs index ee3aff024e..9d4374f76e 100644 --- a/service-providers/network-requester/src/connection.rs +++ b/service-providers/network-requester/src/connection.rs @@ -5,7 +5,7 @@ use futures::channel::mpsc; use nymsphinx::addressing::clients::Recipient; use proxy_helpers::connection_controller::ConnectionReceiver; use proxy_helpers::proxy_runner::ProxyRunner; -use socks5_requests::{ConnectionId, RemoteAddress, Response}; +use socks5_requests::{ConnectionId, Message as Socks5Message, RemoteAddress, Response}; use std::io; use tokio::net::TcpStream; @@ -39,7 +39,7 @@ impl Connection { pub(crate) async fn run_proxy( &mut self, mix_receiver: ConnectionReceiver, - mix_sender: mpsc::UnboundedSender<(Response, Recipient)>, + mix_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>, ) { let stream = self.conn.take().unwrap(); let remote_source_address = "???".to_string(); // we don't know ip address of requester @@ -54,7 +54,10 @@ impl Connection { connection_id, ) .run(move |conn_id, read_data, socket_closed| { - (Response::new(conn_id, read_data, socket_closed), recipient) + ( + Socks5Message::Response(Response::new(conn_id, read_data, socket_closed)), + recipient, + ) }) .await .into_inner(); diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 315854a9da..74fd2fb509 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -3,22 +3,19 @@ use crate::allowed_hosts::{HostsStore, OutboundRequestFilter}; use crate::connection::Connection; -use crate::statistics::{Statistics, StatsData, Timer}; +use crate::statistics::{StatisticsCollector, StatisticsSender, Timer}; use crate::websocket; use crate::websocket::TSWebsocketStream; use futures::channel::mpsc; use futures::stream::{SplitSink, SplitStream}; use futures::{SinkExt, StreamExt}; use log::*; -use nymsphinx::addressing::clients::{ClientIdentity, Recipient}; +use nymsphinx::addressing::clients::Recipient; use nymsphinx::receiver::ReconstructedMessage; use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender}; use socks5_requests::{ConnectionId, Message as Socks5Message, Request, Response}; -use std::collections::HashMap; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; -use tokio::sync::RwLock; use tokio_tungstenite::tungstenite::protocol::Message; use websocket::WebsocketConnectionError; use websocket_requests::{requests::ClientRequest, responses::ServerResponse}; @@ -28,20 +25,18 @@ static ACTIVE_PROXIES: AtomicUsize = AtomicUsize::new(0); pub struct ServiceProvider { listening_address: String, - description: String, - #[cfg(feature = "stats-service")] - db_path: PathBuf, outbound_request_filter: OutboundRequestFilter, open_proxy: bool, enable_statistics: bool, + stats_provider_addr: Option, } impl ServiceProvider { pub fn new( listening_address: String, - description: String, open_proxy: bool, enable_statistics: bool, + stats_provider_addr: Option, ) -> ServiceProvider { let allowed_hosts = HostsStore::new( HostsStore::default_base_dir(), @@ -53,21 +48,13 @@ impl ServiceProvider { PathBuf::from("unknown.list"), ); - #[cfg(feature = "stats-service")] - let db_path = HostsStore::default_base_dir() - .join("service-providers") - .join("network-requester") - .join("db.sqlite"); - let outbound_request_filter = OutboundRequestFilter::new(allowed_hosts, unknown_hosts); ServiceProvider { listening_address, - description, - #[cfg(feature = "stats-service")] - db_path, outbound_request_filter, open_proxy, enable_statistics, + stats_provider_addr, } } @@ -75,21 +62,29 @@ impl ServiceProvider { /// via the `websocket_writer`. async fn mixnet_response_listener( mut websocket_writer: SplitSink, - mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>, - response_stats_data: &Option>>, + mut mix_reader: mpsc::UnboundedReceiver<(Socks5Message, Recipient)>, + stats_collector: Option, ) { // TODO: wire SURBs in here once they're available - while let Some((response, return_address)) = mix_reader.next().await { - if let Some(response_stats_data) = response_stats_data { - response_stats_data - .write() + while let Some((msg, return_address)) = mix_reader.next().await { + if let Some(stats_collector) = stats_collector.as_ref() { + if let Some(remote_addr) = stats_collector + .connected_services + .read() .await - .processed(return_address.identity(), response.data.len() as u32); + .get(&msg.conn_id()) + { + stats_collector + .response_stats_data + .write() + .await + .processed(remote_addr, msg.size() as u32); + } } // make 'request' to native-websocket client let response_message = ClientRequest::Send { recipient: return_address, - message: Socks5Message::Response(response).into_bytes(), + message: msg.into_bytes(), with_reply_surb: false, }; @@ -135,7 +130,7 @@ impl ServiceProvider { remote_addr: String, return_address: Recipient, controller_sender: ControllerSender, - mix_input_sender: mpsc::UnboundedSender<(Response, Recipient)>, + mix_input_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>, ) { let mut conn = match Connection::new(conn_id, remote_addr.clone(), return_address).await { Ok(conn) => conn, @@ -148,7 +143,10 @@ impl ServiceProvider { // inform the remote that the connection is closed before it even was established mix_input_sender - .unbounded_send((Response::new(conn_id, Vec::new(), true), return_address)) + .unbounded_send(( + Socks5Message::Response(Response::new(conn_id, Vec::new(), true)), + return_address, + )) .unwrap(); return; @@ -187,7 +185,7 @@ impl ServiceProvider { fn handle_proxy_connect( &mut self, controller_sender: &mut ControllerSender, - mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>, + mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>, conn_id: ConnectionId, remote_addr: String, return_address: Recipient, @@ -227,12 +225,10 @@ impl ServiceProvider { async fn handle_proxy_message( &mut self, - #[cfg(feature = "stats-service")] storage: &crate::storage::NetworkRequesterStorage, raw_request: &[u8], controller_sender: &mut ControllerSender, - mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>, - request_stats_data: &Option>>, - connected_clients: &mut HashMap, + mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>, + stats_collector: Option, ) { let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) { Ok(msg) => msg, @@ -244,8 +240,12 @@ impl ServiceProvider { match deserialized_msg { Socks5Message::Request(deserialized_request) => match deserialized_request { Request::Connect(req) => { - if self.enable_statistics { - connected_clients.insert(req.conn_id, *req.return_address.identity()); + if let Some(stats_collector) = stats_collector { + stats_collector + .connected_services + .write() + .await + .insert(req.conn_id, req.remote_addr.clone()); } self.handle_proxy_connect( controller_sender, @@ -257,29 +257,24 @@ impl ServiceProvider { } Request::Send(conn_id, data, closed) => { - if let Some(request_stats_data) = request_stats_data { - if let Some(client_identity) = connected_clients.get(&conn_id) { - request_stats_data + if let Some(stats_collector) = stats_collector { + if let Some(remote_addr) = stats_collector + .connected_services + .read() + .await + .get(&conn_id) + { + stats_collector + .request_stats_data .write() .await - .processed(client_identity, data.len() as u32); + .processed(remote_addr, data.len() as u32); } } self.handle_proxy_send(controller_sender, conn_id, data, closed) } }, - Socks5Message::Response(_deserialized_response) => - { - #[cfg(feature = "stats-service")] - match crate::statistics::StatsMessage::from_bytes(&_deserialized_response.data) { - Ok(data) => { - if let Err(e) = storage.insert_service_statistics(data).await { - error!("Could not store received statistics: {}", e); - } - } - Err(e) => error!("Malformed statistics received: {}", e), - } - } + Socks5Message::Response(_) => {} } } @@ -292,7 +287,8 @@ impl ServiceProvider { // channels responsible for managing messages that are to be sent to the mix network. The receiver is // going to be used by `mixnet_response_listener` - let (mix_input_sender, mix_input_receiver) = mpsc::unbounded::<(Response, Recipient)>(); + let (mix_input_sender, mix_input_receiver) = + mpsc::unbounded::<(Socks5Message, Recipient)>(); let (mut timer_sender, timer_receiver) = Timer::new(); let interval = timer_sender.interval(); @@ -306,53 +302,35 @@ impl ServiceProvider { active_connections_controller.run().await; }); - let mut request_stats_data = None; - let mut response_stats_data = None; + let stats_collector = if self.enable_statistics { + let mut stats_sender = + StatisticsSender::new(interval, timer_receiver, self.stats_provider_addr) + .await + .expect("Statistics controller could not be bootstrapped"); + let stats_collector = StatisticsCollector::from(&stats_sender); - if self.enable_statistics { - let mut stats = Statistics::new(self.description.clone(), interval, timer_receiver) - .await - .expect("Statistics controller could not be bootstrapped"); - request_stats_data = Some(Arc::clone(stats.request_data())); - response_stats_data = Some(Arc::clone(stats.response_data())); let mix_input_sender_clone = mix_input_sender.clone(); tokio::spawn(async move { - stats.run(&mix_input_sender_clone).await; + stats_sender.run(&mix_input_sender_clone).await; }); - } - - #[cfg(feature = "stats-service")] - let storage = crate::storage::NetworkRequesterStorage::init(self.db_path.as_path()) - .await - .expect("Could not create network requester storage"); - - #[cfg(feature = "stats-service")] - tokio::spawn( - rocket::build() - .mount( - "/v1", - rocket::routes![crate::storage::post_mixnet_statistics], - ) - .manage(storage.clone()) - .ignite() - .await - .expect("Could not ignite stats api service") - .launch(), - ); + Some(stats_collector) + } else { + None + }; + let stats_collector_clone = stats_collector.clone(); // start the listener for mix messages tokio::spawn(async move { Self::mixnet_response_listener( websocket_writer, mix_input_receiver, - &response_stats_data, + stats_collector_clone, ) .await; }); println!("\nAll systems go. Press CTRL-C to stop the server."); // for each incoming message from the websocket... (which in 99.99% cases is going to be a mix message) - let mut connected_clients = HashMap::new(); loop { let received = match Self::read_websocket_message(&mut websocket_reader).await { Some(msg) => msg, @@ -366,13 +344,10 @@ impl ServiceProvider { // TODO: here be potential SURB (i.e. received.reply_SURB) self.handle_proxy_message( - #[cfg(feature = "stats-service")] - &storage, &raw_message, &mut controller_sender, &mix_input_sender, - &request_stats_data, - &mut connected_clients, + stats_collector.clone(), ) .await; } diff --git a/service-providers/network-requester/src/main.rs b/service-providers/network-requester/src/main.rs index a7a988ebc2..8b6d8d5834 100644 --- a/service-providers/network-requester/src/main.rs +++ b/service-providers/network-requester/src/main.rs @@ -4,19 +4,18 @@ use clap::{App, Arg, ArgMatches}; use network_defaults::DEFAULT_WEBSOCKET_LISTENING_PORT; +use nymsphinx::addressing::clients::Recipient; mod allowed_hosts; mod connection; mod core; mod statistics; -#[cfg(feature = "stats-service")] -mod storage; mod websocket; const OPEN_PROXY_ARG: &str = "open-proxy"; const WS_PORT: &str = "websocket-port"; -const DESCRIPTION: &str = "description"; const ENABLE_STATISTICS: &str = "enable-statistics"; +const STATISTICS_RECIPIENT: &str = "statistics-recipient"; fn parse_args<'a>() -> ArgMatches<'a> { App::new("Nym Network Requester") @@ -37,15 +36,14 @@ fn parse_args<'a>() -> ArgMatches<'a> { ) .arg( Arg::with_name(ENABLE_STATISTICS) - .help("enable mixnet statistics that get sent to a Nym server") - .long(ENABLE_STATISTICS) - .requires(DESCRIPTION), + .help("enable service statistics that get sent to a statistics aggregator server") + .long(ENABLE_STATISTICS), ) .arg( - Arg::with_name(DESCRIPTION) - .help("service description") - .long(DESCRIPTION) - .short("d") + Arg::with_name(STATISTICS_RECIPIENT) + .help("mixnet client address where a statistics aggregator is running. The default value is a Nym aggregator client") + .long(STATISTICS_RECIPIENT) + .requires(ENABLE_STATISTICS) .takes_value(true), ) .get_matches() @@ -63,9 +61,15 @@ async fn main() { let enable_statistics = matches.is_present(ENABLE_STATISTICS); if enable_statistics { - println!("\n\nTHE NETWORK REQUESTER STATISTICS ARE ENABLED. IT WILL COLLECT AND SEND STATISTICS TO A NYM SERVER. PLEASE QUIT IF YOU DON'T WANT THIS TO HAPPEN AND START WITHOUT THE {} FLAG .\n\n", ENABLE_STATISTICS); + println!("\n\nTHE NETWORK REQUESTER STATISTICS ARE ENABLED. IT WILL COLLECT AND SEND ANONYMIZED STATISTICS TO A CENTRAL SERVER. PLEASE QUIT IF YOU DON'T WANT THIS TO HAPPEN AND START WITHOUT THE {} FLAG .\n\n", ENABLE_STATISTICS); } + let stats_provider_addr = matches + .value_of(STATISTICS_RECIPIENT) + .map(Recipient::try_from_base58_string) + .transpose() + .unwrap_or(None); + let uri = format!( "ws://localhost:{}", matches @@ -73,12 +77,9 @@ async fn main() { .unwrap_or(&DEFAULT_WEBSOCKET_LISTENING_PORT.to_string()) ); - let description = matches - .value_of(DESCRIPTION) - .unwrap_or("undefined") - .to_string(); println!("Starting socks5 service provider:"); - let mut server = core::ServiceProvider::new(uri, description, open_proxy, enable_statistics); + let mut server = + core::ServiceProvider::new(uri, open_proxy, enable_statistics, stats_provider_addr); server.run().await; } diff --git a/service-providers/network-requester/src/statistics/comm.rs b/service-providers/network-requester/src/statistics/comm.rs index d92ac44d99..79f619b0db 100644 --- a/service-providers/network-requester/src/statistics/comm.rs +++ b/service-providers/network-requester/src/statistics/comm.rs @@ -4,7 +4,8 @@ use futures::channel::mpsc; use futures::StreamExt; use log::*; -use serde::{Deserialize, Serialize}; +use rand::RngCore; +use serde::Deserialize; use sqlx::types::chrono::{DateTime, Utc}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -12,50 +13,20 @@ use std::time::Duration; use tokio::sync::RwLock; use network_defaults::DEFAULT_NETWORK; -use nymsphinx::addressing::clients::{ClientIdentity, Recipient}; -use socks5_requests::Response; +use nymsphinx::addressing::clients::Recipient; +use ordered_buffer::OrderedMessageSender; +use socks5_requests::{ConnectionId, Message as Socks5Message, RemoteAddress, Request}; +use statistics::api::{ + build_statistics_request_bytes, DEFAULT_STATISTICS_SERVICE_ADDRESS, + DEFAULT_STATISTICS_SERVICE_PORT, +}; +use statistics::{StatsMessage, StatsServiceData}; use super::error::StatsError; const REMOTE_SOURCE_OF_STATS_PROVIDER_CONFIG: &str = "https://nymtech.net/.wellknown/network-requester/stats-provider.json"; -#[derive(Clone, Debug, Deserialize, Serialize)] -pub struct StatsMessage { - pub description: String, - pub stats_data: Vec, - pub interval_seconds: u32, - pub timestamp: String, -} - -impl StatsMessage { - pub fn to_bytes(&self) -> Result, StatsError> { - Ok(bincode::serialize(self)?) - } - - #[cfg(feature = "stats-service")] - pub fn from_bytes(b: &[u8]) -> Result { - Ok(bincode::deserialize(b)?) - } -} - -#[derive(Clone, Debug, Deserialize, Serialize)] -pub struct StatsClientData { - pub client_identity: String, - pub request_bytes: u32, - pub response_bytes: u32, -} - -impl StatsClientData { - pub fn new(client_identity: String, request_bytes: u32, response_bytes: u32) -> Self { - StatsClientData { - client_identity, - request_bytes, - response_bytes, - } - } -} - #[derive(Clone, Debug)] pub struct StatsData { client_processed_bytes: HashMap, @@ -68,13 +39,12 @@ impl StatsData { } } - pub fn processed(&mut self, client_identity: &ClientIdentity, bytes: u32) { - let client_identity_bs58 = client_identity.to_base58_string(); - if let Some(curr_bytes) = self.client_processed_bytes.get_mut(&client_identity_bs58) { + pub fn processed(&mut self, remote_addr: &str, bytes: u32) { + if let Some(curr_bytes) = self.client_processed_bytes.get_mut(remote_addr) { *curr_bytes += bytes; } else { self.client_processed_bytes - .insert(client_identity_bs58, bytes); + .insert(remote_addr.to_string(), bytes); } } } @@ -103,8 +73,24 @@ impl OptionalStatsProviderConfig { } } -pub struct Statistics { - description: String, +#[derive(Clone)] +pub struct StatisticsCollector { + pub(crate) request_stats_data: Arc>, + pub(crate) response_stats_data: Arc>, + pub(crate) connected_services: Arc>>, +} + +impl StatisticsCollector { + pub fn from(stats: &StatisticsSender) -> Self { + Self { + request_stats_data: Arc::clone(&stats.request_data), + response_stats_data: Arc::clone(&stats.response_data), + connected_services: Arc::new(RwLock::new(HashMap::new())), + } + } +} + +pub struct StatisticsSender { request_data: Arc>, response_data: Arc>, interval_seconds: u32, @@ -113,11 +99,11 @@ pub struct Statistics { stats_provider_addr: Recipient, } -impl Statistics { +impl StatisticsSender { pub async fn new( - description: String, interval_seconds: Duration, timer_receiver: mpsc::Receiver<()>, + stats_provider_addr: Option, ) -> Result { let client = reqwest::Client::builder() .timeout(Duration::from_secs(3)) @@ -128,15 +114,16 @@ impl Statistics { .await? .json() .await?; - let stats_provider_addr = Recipient::try_from_base58_string( - stats_provider_config - .stats_client_address() - .ok_or(StatsError::InvalidClientAddress)?, - ) - .map_err(|_| StatsError::InvalidClientAddress)?; + let stats_provider_addr = stats_provider_addr.unwrap_or( + Recipient::try_from_base58_string( + stats_provider_config + .stats_client_address() + .ok_or(StatsError::InvalidClientAddress)?, + ) + .map_err(|_| StatsError::InvalidClientAddress)?, + ); - Ok(Statistics { - description, + Ok(StatisticsSender { request_data: Arc::new(RwLock::new(StatsData::new())), response_data: Arc::new(RwLock::new(StatsData::new())), timestamp: Utc::now(), @@ -146,15 +133,10 @@ impl Statistics { }) } - pub fn request_data(&self) -> &Arc> { - &self.request_data - } - - pub fn response_data(&self) -> &Arc> { - &self.response_data - } - - pub async fn run(&mut self, mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>) { + pub async fn run( + &mut self, + mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>, + ) { loop { if self.timer_receiver.next().await == None { error!("Timer thread has died. No more statistics will be sent"); @@ -162,42 +144,62 @@ impl Statistics { let stats_data = { let request_data_bytes = self.request_data.read().await; let response_data_bytes = self.response_data.read().await; - let clients: HashSet = request_data_bytes + let services: HashSet = request_data_bytes .client_processed_bytes .keys() .chain(response_data_bytes.client_processed_bytes.keys()) .cloned() .collect(); - clients + services .into_iter() - .map(|client_identity| { + .map(|requested_service| { let request_bytes = request_data_bytes .client_processed_bytes - .get(&client_identity) + .get(&requested_service) .copied() .unwrap_or(0); let response_bytes = response_data_bytes .client_processed_bytes - .get(&client_identity) + .get(&requested_service) .copied() .unwrap_or(0); - StatsClientData::new(client_identity, request_bytes, response_bytes) + StatsServiceData::new(requested_service, request_bytes, response_bytes) }) .collect() }; let stats_message = StatsMessage { - description: self.description.clone(), stats_data, interval_seconds: self.interval_seconds, timestamp: self.timestamp.to_rfc3339(), }; - match stats_message.to_bytes() { + match build_statistics_request_bytes(stats_message) { Ok(data) => { - trace!("Sending data to statistics service"); + trace!("Connecting to statistics service"); + let mut rng = rand::rngs::OsRng; + let conn_id = rng.next_u64(); + let connect_req = Request::new_connect( + conn_id, + format!( + "{}:{}", + DEFAULT_STATISTICS_SERVICE_ADDRESS, DEFAULT_STATISTICS_SERVICE_PORT + ), + self.stats_provider_addr, + ); mix_input_sender .unbounded_send(( - Response::new(0, data, false), + Socks5Message::Request(connect_req), + self.stats_provider_addr, + )) + .unwrap(); + + trace!("Sending data to statistics service"); + let mut message_sender = OrderedMessageSender::new(); + let ordered_msg = message_sender.wrap_message(data).into_bytes(); + let send_req = Request::new_send(conn_id, ordered_msg, true); + mix_input_sender + .unbounded_send(( + Socks5Message::Request(send_req), self.stats_provider_addr, )) .unwrap(); diff --git a/service-providers/network-requester/src/statistics/error.rs b/service-providers/network-requester/src/statistics/error.rs index 7ca0d1f22b..eadc188c5a 100644 --- a/service-providers/network-requester/src/statistics/error.rs +++ b/service-providers/network-requester/src/statistics/error.rs @@ -5,9 +5,6 @@ use thiserror::Error; #[derive(Debug, Error)] pub enum StatsError { - #[error("Bincode error: {0}")] - BincodeError(#[from] bincode::Error), - #[error("Reqwuest error {0}")] ReqwestError(#[from] reqwest::Error), diff --git a/service-providers/network-requester/src/statistics/mod.rs b/service-providers/network-requester/src/statistics/mod.rs index 257c405915..8284283801 100644 --- a/service-providers/network-requester/src/statistics/mod.rs +++ b/service-providers/network-requester/src/statistics/mod.rs @@ -5,5 +5,5 @@ mod comm; mod error; mod timer; -pub use comm::{Statistics, StatsClientData, StatsData, StatsMessage}; +pub use comm::{StatisticsCollector, StatisticsSender, StatsData}; pub use timer::Timer; diff --git a/service-providers/network-requester/src/storage/routes.rs b/service-providers/network-requester/src/storage/routes.rs deleted file mode 100644 index 2c76e24c38..0000000000 --- a/service-providers/network-requester/src/storage/routes.rs +++ /dev/null @@ -1,52 +0,0 @@ -// Copyright 2022 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use rocket::serde::json::Json; -use rocket::State; -use serde::{Deserialize, Serialize}; - -use crate::storage::NetworkRequesterStorage; - -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct MixnetStatisticsRequest { - // date, RFC 3339 format - since: String, - // date, RFC 3339 format - until: String, -} - -#[derive(Clone, Serialize, Deserialize, Debug)] -pub struct MixnetStatisticsResponse { - pub service_description: String, - pub client_identity: String, - pub request_processed_bytes: u32, - pub response_processed_bytes: u32, - pub interval_seconds: u32, - pub timestamp: String, -} - -#[rocket::post("/mixnet-statistics", data = "")] -pub(crate) async fn post_mixnet_statistics( - mixnet_statistics_request: Json, - storage: &State, -) -> Json> { - let mixnet_statistics = storage - .get_service_statistics_in_interval( - &mixnet_statistics_request.since, - &mixnet_statistics_request.until, - ) - .await - .unwrap() - .into_iter() - .map(|data| MixnetStatisticsResponse { - service_description: data.service_description, - client_identity: data.client_identity, - request_processed_bytes: data.request_processed_bytes as u32, - response_processed_bytes: data.response_processed_bytes as u32, - interval_seconds: data.interval_seconds as u32, - timestamp: data.timestamp.to_string(), - }) - .collect(); - - Json(mixnet_statistics) -} diff --git a/service-providers/network-statistics/Cargo.toml b/service-providers/network-statistics/Cargo.toml new file mode 100644 index 0000000000..de08cf0785 --- /dev/null +++ b/service-providers/network-statistics/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "nym-network-statistics" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +dirs = "3.0" +log = "0.4" +pretty_env_logger = "0.4" +rocket = { version = "0.5.0-rc.1", features = ["json"] } +serde = { version = "1.0", features = ["derive"] } +sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]} +thiserror = "1" +tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros", "time" ] } +tokio-tungstenite = "0.14" + +statistics = { path = "../../common/statistics" } + +[build-dependencies] +sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } +tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] } diff --git a/service-providers/network-statistics/Rocket.toml b/service-providers/network-statistics/Rocket.toml new file mode 100644 index 0000000000..5828739b6f --- /dev/null +++ b/service-providers/network-statistics/Rocket.toml @@ -0,0 +1,7 @@ +[default] +limits = { forms = "64 kB", json = "1 MiB" } +port = 8090 +address = "127.0.0.1" + +[release] +address = "0.0.0.0" \ No newline at end of file diff --git a/service-providers/network-requester/build.rs b/service-providers/network-statistics/build.rs similarity index 91% rename from service-providers/network-requester/build.rs rename to service-providers/network-statistics/build.rs index 73e5fdf775..b8baded838 100644 --- a/service-providers/network-requester/build.rs +++ b/service-providers/network-statistics/build.rs @@ -7,7 +7,7 @@ use std::env; #[tokio::main] async fn main() { let out_dir = env::var("OUT_DIR").unwrap(); - let database_path = format!("{}/network-requester-example.sqlite", out_dir); + let database_path = format!("{}/network-statistics-example.sqlite", out_dir); let mut conn = SqliteConnection::connect(&*format!("sqlite://{}?mode=rwc", database_path)) .await diff --git a/service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql b/service-providers/network-statistics/migrations/20220512120000_mixnet_statistics.sql similarity index 74% rename from service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql rename to service-providers/network-statistics/migrations/20220512120000_mixnet_statistics.sql index 5370884e27..51514df755 100644 --- a/service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql +++ b/service-providers/network-statistics/migrations/20220512120000_mixnet_statistics.sql @@ -3,11 +3,10 @@ * SPDX-License-Identifier: Apache-2.0 */ -CREATE TABLE mixnet_statistics +CREATE TABLE service_statistics ( id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, - service_description VARCHAR NOT NULL, - client_identity VARCHAR NOT NULL, + requested_service VARCHAR NOT NULL, request_processed_bytes INTEGER NOT NULL, response_processed_bytes INTEGER NOT NULL, interval_seconds INTEGER NOT NULL, diff --git a/service-providers/network-statistics/src/api/error.rs b/service-providers/network-statistics/src/api/error.rs new file mode 100644 index 0000000000..d2b190840e --- /dev/null +++ b/service-providers/network-statistics/src/api/error.rs @@ -0,0 +1,31 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use rocket::http::{ContentType, Status}; +use rocket::response::Responder; +use rocket::{response, Request, Response}; +use std::io::Cursor; + +use crate::storage::error::NetworkStatisticsStorageError; + +pub type Result = std::result::Result; + +#[derive(Debug, thiserror::Error)] +pub enum NetworkStatisticsAPIError { + #[error("{0}")] + RocketError(#[from] Box), + + #[error("{0}")] + StorageError(#[from] NetworkStatisticsStorageError), +} + +impl<'r, 'o: 'r> Responder<'r, 'o> for NetworkStatisticsAPIError { + fn respond_to(self, _: &'r Request<'_>) -> response::Result<'o> { + let err_msg = self.to_string(); + Response::build() + .header(ContentType::Plain) + .sized_body(err_msg.len(), Cursor::new(err_msg)) + .status(Status::BadRequest) + .ok() + } +} diff --git a/service-providers/network-statistics/src/api/mod.rs b/service-providers/network-statistics/src/api/mod.rs new file mode 100644 index 0000000000..3a30d8ebb8 --- /dev/null +++ b/service-providers/network-statistics/src/api/mod.rs @@ -0,0 +1,48 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use log::*; +use rocket::{Ignite, Rocket}; + +use crate::storage::NetworkStatisticsStorage; +use error::Result; +use routes::{post_all_statistics, post_statistic}; + +use statistics::api::STATISTICS_SERVICE_VERSION; + +mod error; +mod routes; + +pub(crate) struct NetworkStatisticsAPI { + rocket: Rocket, +} + +impl NetworkStatisticsAPI { + pub async fn init(storage: NetworkStatisticsStorage) -> Result { + let rocket = rocket::build() + .mount( + STATISTICS_SERVICE_VERSION, + rocket::routes![post_all_statistics, post_statistic], + ) + .manage(storage.clone()) + .ignite() + .await + .map_err(Box::new)?; + + Ok(NetworkStatisticsAPI { rocket }) + } + + pub async fn run(self) { + let shutdown_handle = self.rocket.shutdown(); + tokio::spawn(self.rocket.launch()); + + if let Err(e) = tokio::signal::ctrl_c().await { + error!( + "There was an error while capturing SIGINT - {:?}. We will terminate regardless", + e + ); + } + info!("Received SIGINT - the network statistics API will terminate now"); + shutdown_handle.notify(); + } +} diff --git a/service-providers/network-statistics/src/api/routes.rs b/service-providers/network-statistics/src/api/routes.rs new file mode 100644 index 0000000000..5f5c9bb3b1 --- /dev/null +++ b/service-providers/network-statistics/src/api/routes.rs @@ -0,0 +1,61 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use rocket::serde::json::Json; +use rocket::State; +use serde::{Deserialize, Serialize}; + +use statistics::StatsMessage; + +use crate::api::error::Result; +use crate::storage::NetworkStatisticsStorage; + +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct ServiceStatisticsRequest { + // date, RFC 3339 format + since: String, + // date, RFC 3339 format + until: String, +} + +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct ServiceStatistic { + pub requested_service: String, + pub request_processed_bytes: u32, + pub response_processed_bytes: u32, + pub interval_seconds: u32, + pub timestamp: String, +} + +#[rocket::post("/all-statistics", data = "")] +pub(crate) async fn post_all_statistics( + all_statistics_request: Json, + storage: &State, +) -> Result>> { + let service_statistics = storage + .get_service_statistics_in_interval( + &all_statistics_request.since, + &all_statistics_request.until, + ) + .await? + .into_iter() + .map(|data| ServiceStatistic { + requested_service: data.requested_service, + request_processed_bytes: data.request_processed_bytes as u32, + response_processed_bytes: data.response_processed_bytes as u32, + interval_seconds: data.interval_seconds as u32, + timestamp: data.timestamp.to_string(), + }) + .collect(); + + Ok(Json(service_statistics)) +} + +#[rocket::post("/statistic", data = "")] +pub(crate) async fn post_statistic( + statistic: Json, + storage: &State, +) -> Result> { + storage.insert_service_statistics(statistic.0).await?; + Ok(Json(())) +} diff --git a/service-providers/network-statistics/src/main.rs b/service-providers/network-statistics/src/main.rs new file mode 100644 index 0000000000..0a32438a80 --- /dev/null +++ b/service-providers/network-statistics/src/main.rs @@ -0,0 +1,53 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::path::PathBuf; + +use api::NetworkStatisticsAPI; + +mod api; +mod storage; + +#[tokio::main] +async fn main() { + setup_logging(); + + let base_dir = default_base_dir(); + let storage = storage::NetworkStatisticsStorage::init(&base_dir) + .await + .expect("Could not create network statistics storage"); + + let api = NetworkStatisticsAPI::init(storage) + .await + .expect("Could not ignite stats api service"); + api.run().await; +} + +fn setup_logging() { + let mut log_builder = pretty_env_logger::formatted_timed_builder(); + if let Ok(s) = ::std::env::var("RUST_LOG") { + log_builder.parse_filters(&s); + } else { + // default to 'Info' + log_builder.filter(None, log::LevelFilter::Info); + } + + log_builder + .filter_module("hyper", log::LevelFilter::Warn) + .filter_module("tokio_reactor", log::LevelFilter::Warn) + .filter_module("reqwest", log::LevelFilter::Warn) + .filter_module("mio", log::LevelFilter::Warn) + .filter_module("want", log::LevelFilter::Warn) + .init(); +} + +/// Returns the default base directory for the storefile. +/// +/// This is split out so we can easily inject our own base_dir for unit tests. +fn default_base_dir() -> PathBuf { + dirs::home_dir() + .expect("no home directory known for this OS") + .join(".nym") + .join("service-providers") + .join("network-statistics") +} diff --git a/service-providers/network-requester/src/storage/error.rs b/service-providers/network-statistics/src/storage/error.rs similarity index 76% rename from service-providers/network-requester/src/storage/error.rs rename to service-providers/network-statistics/src/storage/error.rs index 1df83a46fd..48253a261b 100644 --- a/service-providers/network-requester/src/storage/error.rs +++ b/service-providers/network-statistics/src/storage/error.rs @@ -2,7 +2,10 @@ // SPDX-License-Identifier: Apache-2.0 #[derive(Debug, thiserror::Error)] -pub enum NetworkRequesterStorageError { +pub enum NetworkStatisticsStorageError { + #[error("File system error - {0}")] + FSError(#[from] std::io::Error), + #[error("SQL error - {0}")] InternalDatabaseError(#[from] sqlx::Error), diff --git a/service-providers/network-requester/src/storage/manager.rs b/service-providers/network-statistics/src/storage/manager.rs similarity index 70% rename from service-providers/network-requester/src/storage/manager.rs rename to service-providers/network-statistics/src/storage/manager.rs index a60712de11..4fda9d7bd0 100644 --- a/service-providers/network-requester/src/storage/manager.rs +++ b/service-providers/network-statistics/src/storage/manager.rs @@ -3,7 +3,7 @@ use sqlx::types::chrono::{DateTime, Utc}; -use crate::storage::models::MixnetStatistics; +use crate::storage::models::ServiceStatistics; #[derive(Clone)] pub(crate) struct StorageManager { @@ -16,24 +16,21 @@ impl StorageManager { /// /// # Arguments /// - /// * `service_description`: Description of the service that gathered the data. - /// * `client_identity`: Client that connected to the service. + /// * `requested_service`: Address of the service requested. /// * `request_processed_bytes`: Number of bytes for socks5 requests. /// * `response_processed_bytes`: Number of bytes for socks5 responses. /// * `interval_seconds`: Duration in seconds in which the data was gathered. pub(super) async fn insert_service_statistics( &self, - service_description: String, - client_identity: String, + requested_service: String, request_processed_bytes: u32, response_processed_bytes: u32, interval_seconds: u32, timestamp: DateTime, ) -> Result<(), sqlx::Error> { sqlx::query!( - "INSERT INTO mixnet_statistics(service_description, client_identity, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?, ?)", - service_description, - client_identity, + "INSERT INTO service_statistics(requested_service, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?)", + requested_service, request_processed_bytes, response_processed_bytes, interval_seconds, @@ -55,10 +52,10 @@ impl StorageManager { &self, since: DateTime, until: DateTime, - ) -> Result, sqlx::Error> { + ) -> Result, sqlx::Error> { sqlx::query_as!( - MixnetStatistics, - "SELECT * FROM mixnet_statistics WHERE timestamp BETWEEN ? AND ?", + ServiceStatistics, + "SELECT * FROM service_statistics WHERE timestamp BETWEEN ? AND ?", since, until ) diff --git a/service-providers/network-requester/src/storage/mod.rs b/service-providers/network-statistics/src/storage/mod.rs similarity index 66% rename from service-providers/network-requester/src/storage/mod.rs rename to service-providers/network-statistics/src/storage/mod.rs index 2e35565d24..ac7cdb8bdd 100644 --- a/service-providers/network-requester/src/storage/mod.rs +++ b/service-providers/network-statistics/src/storage/mod.rs @@ -4,27 +4,28 @@ use log::*; use sqlx::types::chrono::{DateTime, Utc}; use sqlx::ConnectOptions; -use std::path::Path; +use std::path::PathBuf; -use crate::statistics::StatsMessage; -use crate::storage::error::NetworkRequesterStorageError; +use statistics::StatsMessage; + +use crate::storage::error::NetworkStatisticsStorageError; use crate::storage::manager::StorageManager; -use crate::storage::models::MixnetStatistics; -pub(crate) use crate::storage::routes::post_mixnet_statistics; +use crate::storage::models::ServiceStatistics; -mod error; +pub(crate) mod error; mod manager; mod models; -mod routes; // note that clone here is fine as upon cloning the same underlying pool will be used #[derive(Clone)] -pub(crate) struct NetworkRequesterStorage { +pub(crate) struct NetworkStatisticsStorage { manager: StorageManager, } -impl NetworkRequesterStorage { - pub async fn init(database_path: &Path) -> Result { +impl NetworkStatisticsStorage { + pub async fn init(base_dir: &PathBuf) -> Result { + std::fs::create_dir_all(base_dir)?; + let database_path = base_dir.join("db.sqlite"); let mut opts = sqlx::sqlite::SqliteConnectOptions::new() .filename(database_path) .create_if_missing(true); @@ -36,7 +37,7 @@ impl NetworkRequesterStorage { sqlx::migrate!("./migrations").run(&connection_pool).await?; info!("Database migration finished!"); - let storage = NetworkRequesterStorage { + let storage = NetworkStatisticsStorage { manager: StorageManager { connection_pool }, }; @@ -51,17 +52,16 @@ impl NetworkRequesterStorage { pub(super) async fn insert_service_statistics( &self, msg: StatsMessage, - ) -> Result<(), NetworkRequesterStorageError> { + ) -> Result<(), NetworkStatisticsStorageError> { let timestamp: DateTime = DateTime::parse_from_rfc3339(&msg.timestamp) - .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .map_err(|_| NetworkStatisticsStorageError::TimestampParse)? .into(); - for client_data in msg.stats_data { + for service_data in msg.stats_data { self.manager .insert_service_statistics( - msg.description.clone(), - client_data.client_identity.clone(), - client_data.request_bytes, - client_data.response_bytes, + service_data.requested_service.clone(), + service_data.request_bytes, + service_data.response_bytes, msg.interval_seconds, timestamp, ) @@ -81,12 +81,12 @@ impl NetworkRequesterStorage { &self, since: &str, until: &str, - ) -> Result, NetworkRequesterStorageError> { + ) -> Result, NetworkStatisticsStorageError> { let since = DateTime::parse_from_rfc3339(since) - .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .map_err(|_| NetworkStatisticsStorageError::TimestampParse)? .into(); let until = DateTime::parse_from_rfc3339(until) - .map_err(|_| NetworkRequesterStorageError::TimestampParse)? + .map_err(|_| NetworkStatisticsStorageError::TimestampParse)? .into(); Ok(self .manager diff --git a/service-providers/network-requester/src/storage/models.rs b/service-providers/network-statistics/src/storage/models.rs similarity index 78% rename from service-providers/network-requester/src/storage/models.rs rename to service-providers/network-statistics/src/storage/models.rs index 7c08292a2c..1f3ee3e1de 100644 --- a/service-providers/network-requester/src/storage/models.rs +++ b/service-providers/network-statistics/src/storage/models.rs @@ -4,11 +4,10 @@ use sqlx::types::chrono::NaiveDateTime; // Internally used struct to catch results from the database to get mixnet statistics -pub(crate) struct MixnetStatistics { +pub(crate) struct ServiceStatistics { #[allow(dead_code)] pub(crate) id: i64, - pub(crate) service_description: String, - pub(crate) client_identity: String, + pub(crate) requested_service: String, pub(crate) request_processed_bytes: i64, pub(crate) response_processed_bytes: i64, pub(crate) interval_seconds: i64,