From 7ca4c3b7b1a815a806e9537174e5e9907ea7989c Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Thu, 3 Aug 2023 14:50:35 +0200 Subject: [PATCH] query nym-api from the nym-api --- nym-api/src/main.rs | 11 +++++++++- nym-api/src/network_monitor/mod.rs | 18 ++++++++++++++++- nym-api/src/network_monitor/monitor/sender.rs | 20 +++++++++++++++++-- 3 files changed, 45 insertions(+), 4 deletions(-) diff --git a/nym-api/src/main.rs b/nym-api/src/main.rs index 39719ba0fe..a72be730df 100644 --- a/nym-api/src/main.rs +++ b/nym-api/src/main.rs @@ -22,6 +22,7 @@ use nym_bin_common::logging::setup_logging; use nym_contract_cache::cache::NymContractCache; use nym_sphinx::receiver::SphinxMessageReceiver; use nym_task::TaskManager; +use nym_validator_client::NymApiClient; use rand::rngs::OsRng; use std::error::Error; use support::{http, nyxd}; @@ -124,12 +125,20 @@ async fn start_nym_api_tasks( if config.network_monitor.enabled { // if network monitor is enabled, the storage MUST BE available let storage = maybe_storage.unwrap(); - + let url = format!( + "http://{}:{}", + rocket.config().address, + rocket.config().port + ) + .parse() + .unwrap(); + let nym_api_client = NymApiClient::new(url); network_monitor::start::( &config.network_monitor, nym_contract_cache_state, storage, nyxd_client.clone(), + nym_api_client, &shutdown, ) .await; diff --git a/nym-api/src/network_monitor/mod.rs b/nym-api/src/network_monitor/mod.rs index da1ff1088e..88b724e6ea 100644 --- a/nym-api/src/network_monitor/mod.rs +++ b/nym-api/src/network_monitor/mod.rs @@ -22,6 +22,7 @@ use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::params::PacketType; use nym_sphinx::receiver::MessageReceiver; use nym_task::TaskManager; +use nym_validator_client::NymApiClient; use std::sync::Arc; pub(crate) mod gateways_reader; @@ -36,10 +37,12 @@ pub(crate) fn setup<'a>( nym_contract_cache_state: &NymContractCache, storage: &NymApiStorage, nyxd_client: nyxd::Client, + nym_api_client: NymApiClient, ) -> NetworkMonitorBuilder<'a> { NetworkMonitorBuilder::new( config, nyxd_client, + nym_api_client, storage.to_owned(), nym_contract_cache_state.to_owned(), ) @@ -48,6 +51,7 @@ pub(crate) fn setup<'a>( pub(crate) struct NetworkMonitorBuilder<'a> { config: &'a config::NetworkMonitor, nyxd_client: nyxd::Client, + nym_api_client: NymApiClient, node_status_storage: NymApiStorage, validator_cache: NymContractCache, } @@ -56,12 +60,14 @@ impl<'a> NetworkMonitorBuilder<'a> { pub(crate) fn new( config: &'a config::NetworkMonitor, nyxd_client: nyxd::Client, + nym_api_client: NymApiClient, node_status_storage: NymApiStorage, validator_cache: NymContractCache, ) -> Self { NetworkMonitorBuilder { config, nyxd_client, + nym_api_client, node_status_storage, validator_cache, } @@ -108,6 +114,7 @@ impl<'a> NetworkMonitorBuilder<'a> { Arc::clone(&encryption_keypair), self.config.debug.gateway_sending_rate, bandwidth_controller, + self.nym_api_client, self.config.debug.disabled_credentials_mode, ); @@ -181,6 +188,7 @@ fn new_packet_sender( local_sphinx: Arc, max_sending_rate: usize, bandwidth_controller: BandwidthController, + nym_api_client: nym_validator_client::NymApiClient, disabled_credentials_mode: bool, ) -> PacketSender { PacketSender::new( @@ -192,6 +200,7 @@ fn new_packet_sender( config.debug.max_concurrent_gateway_clients, max_sending_rate, bandwidth_controller, + nym_api_client, disabled_credentials_mode, ) } @@ -224,9 +233,16 @@ pub(crate) async fn start( nym_contract_cache_state: &NymContractCache, storage: &NymApiStorage, nyxd_client: nyxd::Client, + nym_api_client: NymApiClient, shutdown: &TaskManager, ) { - let monitor_builder = setup(config, nym_contract_cache_state, storage, nyxd_client); + let monitor_builder = setup( + config, + nym_contract_cache_state, + storage, + nyxd_client, + nym_api_client, + ); info!("Starting network monitor..."); let runnables: NetworkMonitorRunnables = monitor_builder.build().await; runnables.spawn_tasks(shutdown); diff --git a/nym-api/src/network_monitor/monitor/sender.rs b/nym-api/src/network_monitor/monitor/sender.rs index dc12109f14..ff0d42de84 100644 --- a/nym-api/src/network_monitor/monitor/sender.rs +++ b/nym-api/src/network_monitor/monitor/sender.rs @@ -21,6 +21,7 @@ use nym_gateway_client::error::GatewayClientError; use nym_gateway_client::{AcknowledgementReceiver, GatewayClient, MixnetMessageReceiver}; use nym_sphinx::forwarding::packet::MixPacket; use nym_task::TaskClient; +use nym_validator_client::NymApiClient; use pin_project::pin_project; use std::mem; use std::num::NonZeroUsize; @@ -133,6 +134,7 @@ pub(crate) struct PacketSender { active_gateway_clients: ActiveGatewayClients, fresh_gateway_client_data: Arc, + nym_api_client: NymApiClient, gateway_connection_timeout: Duration, max_concurrent_clients: usize, max_sending_rate: usize, @@ -151,6 +153,7 @@ impl PacketSender { max_concurrent_clients: usize, max_sending_rate: usize, bandwidth_controller: BandwidthController, + nym_api_client: NymApiClient, disabled_credentials_mode: bool, ) -> Self { PacketSender { @@ -163,6 +166,7 @@ impl PacketSender { bandwidth_controller, disabled_credentials_mode, }), + nym_api_client, gateway_connection_timeout, max_concurrent_clients, max_sending_rate, @@ -186,6 +190,7 @@ impl PacketSender { identity: identity::PublicKey, sphinx: encryption::PublicKey, fresh_gateway_client_data: &FreshGatewayClientData, + nym_api_client: NymApiClient, ) -> ( GatewayClientHandle, (MixnetMessageReceiver, AcknowledgementReceiver), @@ -209,6 +214,7 @@ impl PacketSender { ack_sender, fresh_gateway_client_data.gateway_response_timeout, Some(fresh_gateway_client_data.bandwidth_controller.clone()), + nym_api_client, nym_task::TaskClient::dummy(), ); @@ -285,13 +291,19 @@ impl PacketSender { identity: identity::PublicKey, sphinx: encryption::PublicKey, fresh_gateway_client_data: &FreshGatewayClientData, + nym_api_client: NymApiClient, gateway_connection_timeout: Duration, ) -> Option<( GatewayClientHandle, (MixnetMessageReceiver, AcknowledgementReceiver), )> { - let (new_client, (message_receiver, ack_receiver)) = - Self::new_gateway_client_handle(address, identity, sphinx, fresh_gateway_client_data); + let (new_client, (message_receiver, ack_receiver)) = Self::new_gateway_client_handle( + address, + identity, + sphinx, + fresh_gateway_client_data, + nym_api_client, + ); // Put this in timeout in case the gateway has incorrectly set their ulimit and our connection // gets stuck in their TCP queue and just hangs on our end but does not terminate @@ -349,6 +361,7 @@ impl PacketSender { gateway_connection_timeout: Duration, packets: GatewayPackets, fresh_gateway_client_data: Arc, + nym_api_client: NymApiClient, client: Option, max_sending_rate: usize, ) -> Option { @@ -370,6 +383,7 @@ impl PacketSender { packets.pub_key, packets.encryption_key, &fresh_gateway_client_data, + nym_api_client, gateway_connection_timeout, ) .await?; @@ -497,6 +511,7 @@ impl PacketSender { // we're not interacting with right now) drop(guard); + let nym_api_client = &self.nym_api_client; // can't chain it all nicely together as there's no adapter method defined on Stream directly // for ForEachConcurrentClientUse let used_clients = ForEachConcurrentClientUse::new( @@ -507,6 +522,7 @@ impl PacketSender { gateway_connection_timeout, packets, fresh_data, + nym_api_client.clone(), client, max_sending_rate, )