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
This commit is contained in:
Bogdan-Ștefan Neacşu
committed by
GitHub
GitHub
parent
e9aa75ccac
commit
fbdf31b879
+3
-1
@@ -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)
|
||||
|
||||
|
||||
Generated
+27
-2
@@ -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"
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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<Message, MessageError> {
|
||||
if b.is_empty() {
|
||||
return Err(MessageError::NoData);
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
# Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
# 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"
|
||||
@@ -0,0 +1,40 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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<Vec<u8>, 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())
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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),
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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<StatsServiceData>,
|
||||
pub interval_seconds: u32,
|
||||
pub timestamp: String,
|
||||
}
|
||||
|
||||
impl StatsMessage {
|
||||
pub fn to_json(&self) -> Result<String, StatsError> {
|
||||
Ok(serde_json::to_string(self)?)
|
||||
}
|
||||
|
||||
pub fn from_json(s: &str) -> Result<Self, StatsError> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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"]
|
||||
@@ -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.
|
||||
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.
|
||||
@@ -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();
|
||||
|
||||
@@ -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<Recipient>,
|
||||
}
|
||||
|
||||
impl ServiceProvider {
|
||||
pub fn new(
|
||||
listening_address: String,
|
||||
description: String,
|
||||
open_proxy: bool,
|
||||
enable_statistics: bool,
|
||||
stats_provider_addr: Option<Recipient>,
|
||||
) -> 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<TSWebsocketStream, Message>,
|
||||
mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>,
|
||||
response_stats_data: &Option<Arc<RwLock<StatsData>>>,
|
||||
mut mix_reader: mpsc::UnboundedReceiver<(Socks5Message, Recipient)>,
|
||||
stats_collector: Option<StatisticsCollector>,
|
||||
) {
|
||||
// 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<Arc<RwLock<StatsData>>>,
|
||||
connected_clients: &mut HashMap<ConnectionId, ClientIdentity>,
|
||||
mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>,
|
||||
stats_collector: Option<StatisticsCollector>,
|
||||
) {
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<StatsClientData>,
|
||||
pub interval_seconds: u32,
|
||||
pub timestamp: String,
|
||||
}
|
||||
|
||||
impl StatsMessage {
|
||||
pub fn to_bytes(&self) -> Result<Vec<u8>, StatsError> {
|
||||
Ok(bincode::serialize(self)?)
|
||||
}
|
||||
|
||||
#[cfg(feature = "stats-service")]
|
||||
pub fn from_bytes(b: &[u8]) -> Result<Self, StatsError> {
|
||||
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<String, u32>,
|
||||
@@ -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<RwLock<StatsData>>,
|
||||
pub(crate) response_stats_data: Arc<RwLock<StatsData>>,
|
||||
pub(crate) connected_services: Arc<RwLock<HashMap<ConnectionId, RemoteAddress>>>,
|
||||
}
|
||||
|
||||
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<RwLock<StatsData>>,
|
||||
response_data: Arc<RwLock<StatsData>>,
|
||||
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<Recipient>,
|
||||
) -> Result<Self, StatsError> {
|
||||
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<RwLock<StatsData>> {
|
||||
&self.request_data
|
||||
}
|
||||
|
||||
pub fn response_data(&self) -> &Arc<RwLock<StatsData>> {
|
||||
&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<String> = request_data_bytes
|
||||
let services: HashSet<String> = 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();
|
||||
|
||||
@@ -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),
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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 = "<mixnet_statistics_request>")]
|
||||
pub(crate) async fn post_mixnet_statistics(
|
||||
mixnet_statistics_request: Json<MixnetStatisticsRequest>,
|
||||
storage: &State<NetworkRequesterStorage>,
|
||||
) -> Json<Vec<MixnetStatisticsResponse>> {
|
||||
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)
|
||||
}
|
||||
@@ -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"] }
|
||||
@@ -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"
|
||||
+1
-1
@@ -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
|
||||
+2
-3
@@ -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,
|
||||
@@ -0,0 +1,31 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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<T> = std::result::Result<T, NetworkStatisticsAPIError>;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum NetworkStatisticsAPIError {
|
||||
#[error("{0}")]
|
||||
RocketError(#[from] Box<rocket::Error>),
|
||||
|
||||
#[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()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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<Ignite>,
|
||||
}
|
||||
|
||||
impl NetworkStatisticsAPI {
|
||||
pub async fn init(storage: NetworkStatisticsStorage) -> Result<Self> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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 = "<all_statistics_request>")]
|
||||
pub(crate) async fn post_all_statistics(
|
||||
all_statistics_request: Json<ServiceStatisticsRequest>,
|
||||
storage: &State<NetworkStatisticsStorage>,
|
||||
) -> Result<Json<Vec<ServiceStatistic>>> {
|
||||
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 = "<statistic>")]
|
||||
pub(crate) async fn post_statistic(
|
||||
statistic: Json<StatsMessage>,
|
||||
storage: &State<NetworkStatisticsStorage>,
|
||||
) -> Result<Json<()>> {
|
||||
storage.insert_service_statistics(statistic.0).await?;
|
||||
Ok(Json(()))
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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")
|
||||
}
|
||||
+4
-1
@@ -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),
|
||||
|
||||
+8
-11
@@ -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<Utc>,
|
||||
) -> 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<Utc>,
|
||||
until: DateTime<Utc>,
|
||||
) -> Result<Vec<MixnetStatistics>, sqlx::Error> {
|
||||
) -> Result<Vec<ServiceStatistics>, 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
|
||||
)
|
||||
+21
-21
@@ -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<Self, NetworkRequesterStorageError> {
|
||||
impl NetworkStatisticsStorage {
|
||||
pub async fn init(base_dir: &PathBuf) -> Result<Self, NetworkStatisticsStorageError> {
|
||||
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<Utc> = 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<Vec<MixnetStatistics>, NetworkRequesterStorageError> {
|
||||
) -> Result<Vec<ServiceStatistics>, 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
|
||||
+2
-3
@@ -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,
|
||||
Reference in New Issue
Block a user