Split geolcation into its own task the picks off mixnodes that don't have locations or ones where the cache TTL has expired
This commit is contained in:
Mark Sinclair
@@ -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<GeoLocation, LocateError> {
|
||||
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::<GeoLocation>().await?;
|
||||
Ok(location)
|
||||
}
|
||||
Err(e) => Err(LocateError::ReqwestError(e)),
|
||||
}
|
||||
}
|
||||
@@ -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<GeoLocation, ReqwestError> {
|
||||
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::<GeoLocation>().await?;
|
||||
Ok(location)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user