From a9422a32ed74f365afded44ef64fee3ae753ce7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Thu, 26 May 2022 11:33:25 +0300 Subject: [PATCH] Make stats in network requester opt-in (#1288) --- .../network-requester/src/core.rs | 56 ++++++++++++------- .../network-requester/src/main.rs | 22 ++++++-- 2 files changed, 53 insertions(+), 25 deletions(-) diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index dd17a7cf38..315854a9da 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -33,6 +33,7 @@ pub struct ServiceProvider { db_path: PathBuf, outbound_request_filter: OutboundRequestFilter, open_proxy: bool, + enable_statistics: bool, } impl ServiceProvider { @@ -40,6 +41,7 @@ impl ServiceProvider { listening_address: String, description: String, open_proxy: bool, + enable_statistics: bool, ) -> ServiceProvider { let allowed_hosts = HostsStore::new( HostsStore::default_base_dir(), @@ -65,6 +67,7 @@ impl ServiceProvider { db_path, outbound_request_filter, open_proxy, + enable_statistics, } } @@ -73,14 +76,16 @@ impl ServiceProvider { async fn mixnet_response_listener( mut websocket_writer: SplitSink, mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>, - response_stats_data: &Arc>, + response_stats_data: &Option>>, ) { // TODO: wire SURBs in here once they're available while let Some((response, return_address)) = mix_reader.next().await { - response_stats_data - .write() - .await - .processed(return_address.identity(), response.data.len() as u32); + if let Some(response_stats_data) = response_stats_data { + response_stats_data + .write() + .await + .processed(return_address.identity(), response.data.len() as u32); + } // make 'request' to native-websocket client let response_message = ClientRequest::Send { recipient: return_address, @@ -226,7 +231,7 @@ impl ServiceProvider { raw_request: &[u8], controller_sender: &mut ControllerSender, mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>, - request_stats_data: &Arc>, + request_stats_data: &Option>>, connected_clients: &mut HashMap, ) { let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) { @@ -239,7 +244,9 @@ impl ServiceProvider { match deserialized_msg { Socks5Message::Request(deserialized_request) => match deserialized_request { Request::Connect(req) => { - connected_clients.insert(req.conn_id, *req.return_address.identity()); + if self.enable_statistics { + connected_clients.insert(req.conn_id, *req.return_address.identity()); + } self.handle_proxy_connect( controller_sender, mix_input_sender, @@ -250,11 +257,13 @@ impl ServiceProvider { } Request::Send(conn_id, data, closed) => { - if let Some(client_identity) = connected_clients.get(&conn_id) { - request_stats_data - .write() - .await - .processed(client_identity, data.len() as u32); + if let Some(request_stats_data) = request_stats_data { + if let Some(client_identity) = connected_clients.get(&conn_id) { + request_stats_data + .write() + .await + .processed(client_identity, data.len() as u32); + } } self.handle_proxy_send(controller_sender, conn_id, data, closed) } @@ -297,15 +306,20 @@ impl ServiceProvider { active_connections_controller.run().await; }); - let mut stats = Statistics::new(self.description.clone(), interval, timer_receiver) - .await - .expect("Statistics controller could not be bootstrapped"); - let request_stats_data = Arc::clone(stats.request_data()); - let response_stats_data = 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; - }); + let mut request_stats_data = None; + let mut response_stats_data = None; + + 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; + }); + } #[cfg(feature = "stats-service")] let storage = crate::storage::NetworkRequesterStorage::init(self.db_path.as_path()) diff --git a/service-providers/network-requester/src/main.rs b/service-providers/network-requester/src/main.rs index 935f37b6c1..a7a988ebc2 100644 --- a/service-providers/network-requester/src/main.rs +++ b/service-providers/network-requester/src/main.rs @@ -16,6 +16,7 @@ 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"; fn parse_args<'a>() -> ArgMatches<'a> { App::new("Nym Network Requester") @@ -34,13 +35,18 @@ fn parse_args<'a>() -> ArgMatches<'a> { .short("p") .takes_value(true), ) + .arg( + Arg::with_name(ENABLE_STATISTICS) + .help("enable mixnet statistics that get sent to a Nym server") + .long(ENABLE_STATISTICS) + .requires(DESCRIPTION), + ) .arg( Arg::with_name(DESCRIPTION) .help("service description") .long(DESCRIPTION) .short("d") - .takes_value(true) - .default_value("undefined"), + .takes_value(true), ) .get_matches() } @@ -55,6 +61,11 @@ async fn main() { println!("\n\nYOU HAVE STARTED IN 'OPEN PROXY' MODE. ANYONE WITH YOUR CLIENT ADDRESS CAN MAKE REQUESTS FROM YOUR MACHINE. PLEASE QUIT IF YOU DON'T UNDERSTAND WHAT YOU'RE DOING.\n\n"); } + 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); + } + let uri = format!( "ws://localhost:{}", matches @@ -62,9 +73,12 @@ async fn main() { .unwrap_or(&DEFAULT_WEBSOCKET_LISTENING_PORT.to_string()) ); - let description = matches.value_of(DESCRIPTION).unwrap().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); + let mut server = core::ServiceProvider::new(uri, description, open_proxy, enable_statistics); server.run().await; }