From b2ec19ece477c7a6eac8e59e6f21b83edb34e86b Mon Sep 17 00:00:00 2001 From: Mark Sinclair Date: Wed, 6 Oct 2021 20:47:25 +0100 Subject: [PATCH] Split geolcation into its own task the picks off mixnodes that don't have locations or ones where the cache TTL has expired --- .../src/country_statistics/geolocate.rs | 139 ++++++++++++++++++ explorer-api/src/country_statistics/mod.rs | 104 +------------ 2 files changed, 141 insertions(+), 102 deletions(-) create mode 100644 explorer-api/src/country_statistics/geolocate.rs diff --git a/explorer-api/src/country_statistics/geolocate.rs b/explorer-api/src/country_statistics/geolocate.rs new file mode 100644 index 0000000000..8e23a1efa0 --- /dev/null +++ b/explorer-api/src/country_statistics/geolocate.rs @@ -0,0 +1,139 @@ +use log::{info, warn}; +use reqwest::Error as ReqwestError; +use thiserror::Error; + +use crate::mix_nodes::{GeoLocation, Location}; +use crate::state::ExplorerApiStateContext; + +pub(crate) struct GeoLocateTask { + state: ExplorerApiStateContext, +} + +impl GeoLocateTask { + pub(crate) fn new(state: ExplorerApiStateContext) -> Self { + GeoLocateTask { state } + } + + pub(crate) fn start(mut self) { + info!("Spawning mix node locator task runner..."); + tokio::spawn(async move { + let mut interval_timer = tokio::time::interval(std::time::Duration::from_millis(50)); + loop { + // wait for the next interval tick + interval_timer.tick().await; + self.locate_mix_nodes().await; + } + }); + } + + async fn locate_mix_nodes(&mut self) { + let mixnode_bonds = self.state.inner.mix_nodes.get().await.value; + + for (i, cache_item) in mixnode_bonds.values().enumerate() { + if self + .state + .inner + .mix_nodes + .is_location_valid(&cache_item.mix_node.identity_key) + .await + { + // when the cached location is valid, don't locate and continue to next mix node + continue; + } + + // the mix node has not been located or is the cache time has expired + match locate(&cache_item.mix_node.host).await { + Ok(geo_location) => { + let location = Location::new(geo_location); + + trace!( + "{} mix nodes already located. Ip {} is located in {:#?}", + i, + cache_item.mix_node.host, + location.three_letter_iso_country_code, + ); + + if i > 0 && (i % 100) == 0 { + info!( + "Located {} mixnodes...", + i + 1, + ); + } + + self.state + .inner + .mix_nodes + .set_location(&cache_item.mix_node.identity_key, Some(location)) + .await; + + // one node has been located, so break out of the loop + return; + } + Err(e) => match e { + LocateError::ReqwestError(e) => warn!( + "❌ Oh no! Location for {} failed {}", + cache_item.mix_node.host, e + ), + LocateError::NotFound(e) => { + error!( + "❌ Location for {} not found. Response body: {}", + cache_item.mix_node.host, e + ); + self.state + .inner + .mix_nodes + .set_location(&cache_item.mix_node.identity_key, None) + .await; + }, + LocateError::RateLimited(e) => error!( + "❌ Oh no, we've been rate limited! Location for {} failed. Response body: {}", + cache_item.mix_node.host, e + ), + }, + } + } + + trace!("All mix nodes located"); + } +} + +#[derive(Debug, Error)] +enum LocateError { + #[error("Oops, we have made too many requests and are being rate limited. Request body: {0}")] + RateLimited(String), + + #[error("Geolocation not found. Request body: {0}")] + NotFound(String), + + #[error(transparent)] + ReqwestError(#[from] ReqwestError), +} + +async fn locate(ip: &str) -> Result { + let api_key = ::std::env::var("GEO_IP_SERVICE_API_KEY") + .expect("Env var GEO_IP_SERVICE_API_KEY is not set"); + let uri = format!("{}/{}?apikey={}", crate::GEO_IP_SERVICE, ip, api_key); + match reqwest::get(uri.clone()).await { + Ok(response) => { + if response.status() == 429 { + return Err(LocateError::RateLimited( + response + .text() + .await + .unwrap_or_else(|_| "(the response body is empty)".to_string()), + )); + } + if response.status() == 404 { + return Err(LocateError::NotFound( + response + .text() + .await + .unwrap_or_else(|_| "(the response body is empty)".to_string()), + )); + } + let location = response.json::().await?; + Ok(location) + } + Err(e) => Err(LocateError::ReqwestError(e)), + } +} diff --git a/explorer-api/src/country_statistics/mod.rs b/explorer-api/src/country_statistics/mod.rs index 681cc7cc32..43cbaadeba 100644 --- a/explorer-api/src/country_statistics/mod.rs +++ b/explorer-api/src/country_statistics/mod.rs @@ -1,104 +1,4 @@ -use log::{info, trace, warn}; -use reqwest::Error as ReqwestError; - -use crate::country_statistics::country_nodes_distribution::CountryNodesDistribution; -use crate::mix_nodes::{GeoLocation, Location}; -use crate::state::ExplorerApiStateContext; - pub mod country_nodes_distribution; +pub mod distribution; +pub mod geolocate; pub mod http; - -pub(crate) struct CountryStatistics { - state: ExplorerApiStateContext, -} - -impl CountryStatistics { - pub(crate) fn new(state: ExplorerApiStateContext) -> Self { - CountryStatistics { state } - } - - pub(crate) fn start(mut self) { - info!("Spawning task runner..."); - tokio::spawn(async move { - let mut interval_timer = tokio::time::interval(std::time::Duration::from_secs(60 * 60)); - loop { - // wait for the next interval tick - interval_timer.tick().await; - - info!("Running task..."); - self.calculate_nodes_per_country().await; - info!("Done"); - } - }); - } - - /// Retrieves the current list of mixnodes from the validators and calculates how many nodes are in each country - async fn calculate_nodes_per_country(&mut self) { - // force the mixnode cache to invalidate - let mixnode_bonds = self.state.inner.mix_nodes.refresh_and_get().await.value; - - let mut distribution = CountryNodesDistribution::new(); - - info!("Locating mixnodes..."); - for (i, cache_item) in mixnode_bonds.values().enumerate() { - match locate(&cache_item.bond.mix_node.host).await { - Ok(geo_location) => { - let location = Location::new(geo_location); - - *(distribution.entry(location.three_letter_iso_country_code.to_string())) - .or_insert(0) += 1; - - trace!( - "Ip {} is located in {:#?}", - cache_item.bond.mix_node.host, - location.three_letter_iso_country_code, - ); - - self.state - .inner - .mix_nodes - .set_location(&cache_item.bond.mix_node.identity_key, location) - .await; - - if (i % 100) == 0 { - info!( - "Located {} mixnodes in {} countries", - i + 1, - distribution.len() - ); - } - } - Err(e) => warn!("❌ Oh no! Location failed {}", e), - } - } - - // replace the shared distribution to be the new distribution - self.state - .inner - .country_node_distribution - .set_all(distribution) - .await; - - info!( - "Locating mixnodes done: {:?}", - self.state.inner.country_node_distribution.get_all().await - ); - - // keep state on disk, so that when this process dies it can start up again and users get some data - self.state.write_to_file().await; - } -} - -async fn locate(ip: &str) -> Result { - let api_key = ::std::env::var("GEO_IP_SERVICE_API_KEY") - .expect("Env var GEO_IP_SERVICE_API_KEY is not set"); - let response = reqwest::get(format!( - "{}{}?apikey={}", - crate::GEO_IP_SERVICE, - ip, - api_key - )) - .await?; - let location = response.json::().await?; - Ok(location) -}