GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make stats in network requester opt-in (#1288)

Browse Source
This commit is contained in:
Bogdan-Ștefan Neacşu
2022-05-26 11:33:25 +03:00
committed by GitHub
parent 177f1deefe
commit a9422a32ed
2 changed files with 53 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service-providers/network-requester/src/core.rs
+35 -21
View File
@@ -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<TSWebsocketStream, Message>,
mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>,
response_stats_data: &Arc<RwLock<StatsData>>,
response_stats_data: &Option<Arc<RwLock<StatsData>>>,
) {
// 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<RwLock<StatsData>>,
request_stats_data: &Option<Arc<RwLock<StatsData>>>,
connected_clients: &mut HashMap<ConnectionId, ClientIdentity>,
) {
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())
service-providers/network-requester/src/main.rs
+18 -4
View File
@@ -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;
}