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

Expand /v3/nym-nodes with geodata (#5686)

Browse Source 
* Expand /v3/nym-nodes
- includes node description and geodata
- expanded scope of included geodata

* Fetch geodata for all nodes

* Bump package version
This commit is contained in:
dynco-nym
2025-04-10 21:12:33 +02:00
committed by GitHub
parent 0be844e015 33c783bb7c
commit 660463908d
11 changed files with 189 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -6221,7 +6221,7 @@ dependencies = [
[[package]]
name = "nym-node-status-api"
version = "2.0.0"
version = "2.1.0"
dependencies = [
"ammonia",
"anyhow",
nym-node-status-api/nym-node-status-api/Cargo.toml
+1 -1
View File
@@ -3,7 +3,7 @@
[package]
name = "nym-node-status-api"
version = "2.0.0"
version = "2.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs
+2 -2
View File
@@ -15,8 +15,8 @@ pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_s
pub(crate) use misc::insert_summaries;
pub(crate) use mixnodes::{get_all_mixnodes, get_bonded_mix_ids, get_daily_stats, update_mixnodes};
pub(crate) use nym_nodes::{
get_all_nym_nodes, get_described_bonded_nym_nodes, get_described_node_bond_info,
get_node_descriptions, update_nym_nodes,
get_all_nym_nodes, get_bonded_node_description, get_described_bonded_nym_nodes,
get_described_node_bond_info, get_node_self_description, update_nym_nodes,
};
pub(crate) use packet_stats::{
get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats,
nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs
+44 -1
View File
@@ -1,4 +1,5 @@
use futures_util::TryStreamExt;
use nym_node_requests::api::v1::node::models::NodeDescription;
use nym_validator_client::{
client::{NodeId, NymNodeDetails},
models::NymNodeDescription,
@@ -180,7 +181,7 @@ pub(crate) async fn get_described_node_bond_info(
.map_err(From::from)
}
pub(crate) async fn get_node_descriptions(
pub(crate) async fn get_node_self_description(
pool: &DbPool,
) -> anyhow::Result<HashMap<NodeId, NymNodeDescription>> {
let mut conn = pool.acquire().await?;
@@ -213,3 +214,45 @@ pub(crate) async fn get_node_descriptions(
})
.map_err(From::from)
}
pub(crate) async fn get_bonded_node_description(
pool: &DbPool,
) -> anyhow::Result<HashMap<NodeId, NodeDescription>> {
let mut conn = pool.acquire().await?;
sqlx::query!(
r#"SELECT
nd.node_id,
moniker,
website,
security_contact,
details
FROM
nym_node_descriptions nd
INNER JOIN
nym_nodes
WHERE
bond_info IS NOT NULL
"#,
)
.fetch_all(&mut *conn)
.await
.map(|records| {
records
.into_iter()
.map(|elem| {
let node_id: NodeId = elem.node_id.try_into().unwrap_or_default();
(
node_id,
NodeDescription {
moniker: elem.moniker.unwrap_or_default(),
website: elem.website.unwrap_or_default(),
security_contact: elem.security_contact.unwrap_or_default(),
details: elem.details.unwrap_or_default(),
},
)
})
.collect::<HashMap<NodeId, NodeDescription>>()
})
.map_err(From::from)
}
nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs
+9 -4
View File
@@ -32,11 +32,16 @@ async fn nym_nodes(
State(state): State<AppState>,
) -> HttpResult<Json<PagedResult<ExtendedNymNode>>> {
let db = state.db_pool();
let node_geocache = state.node_geocache();
let nodes = state.cache().get_nym_nodes_list(db).await.map_err(|e| {
tracing::error!("{e}");
HttpError::internal()
})?;
let nodes = state
.cache()
.get_nym_nodes_list(db, node_geocache)
.await
.map_err(|e| {
tracing::error!("{e}");
HttpError::internal()
})?;
Ok(Json(PagedResult::paginate(pagination, nodes)))
}
nym-node-status-api/nym-node-status-api/src/http/models.rs
+16 -1
View File
@@ -60,8 +60,23 @@ pub(crate) struct ExtendedNymNode {
pub(crate) node_type: String,
pub(crate) ip_address: String,
pub(crate) accepted_tnc: bool,
pub(crate) description: serde_json::Value,
pub(crate) self_description: serde_json::Value,
pub(crate) rewarding_details: serde_json::Value,
pub(crate) description: NodeDescription,
pub(crate) geoip: Option<NodeGeoData>,
}
#[derive(Clone, Debug, utoipa::ToSchema, Deserialize, Serialize)]
pub(crate) struct NodeGeoData {
pub(crate) city: String,
pub(crate) country: String,
pub(crate) ip_address: String,
pub(crate) latitude: String,
pub(crate) longitude: String,
pub(crate) org: String,
pub(crate) postal: String,
pub(crate) region: String,
pub(crate) timezone: String,
}
#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)]
nym-node-status-api/nym-node-status-api/src/http/server.rs
+10 -1
View File
@@ -7,6 +7,7 @@ use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
use crate::{
db::DbPool,
http::{api::RouterBuilder, state::AppState},
monitor::NodeGeoCache,
};
/// Return handles that allow for graceful shutdown of server + awaiting its
@@ -17,10 +18,18 @@ pub(crate) async fn start_http_api(
nym_http_cache_ttl: u64,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
node_geocache: NodeGeoCache,
) -> anyhow::Result<ShutdownHandles> {
let router_builder = RouterBuilder::with_default_routes();
let state = AppState::new(db_pool, nym_http_cache_ttl, agent_key_list, agent_max_count).await;
let state = AppState::new(
db_pool,
nym_http_cache_ttl,
agent_key_list,
agent_max_count,
node_geocache,
)
.await;
let router = router_builder.with_state(state);
let bind_addr = format!("0.0.0.0:{}", http_port);
nym-node-status-api/nym-node-status-api/src/http/state.rs
+46 -10
View File
@@ -10,7 +10,8 @@ use tracing::instrument;
use crate::{
db::{queries, DbPool},
http::models::{DailyStats, ExtendedNymNode, Gateway, Mixnode, SummaryHistory},
http::models::{DailyStats, ExtendedNymNode, Gateway, Mixnode, NodeGeoData, SummaryHistory},
monitor::NodeGeoCache,
};
use super::models::SessionStats;
@@ -21,6 +22,7 @@ pub(crate) struct AppState {
cache: HttpCache,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
node_geocache: NodeGeoCache,
}
impl AppState {
@@ -29,12 +31,14 @@ impl AppState {
cache_ttl: u64,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
node_geocache: NodeGeoCache,
) -> Self {
Self {
db_pool,
cache: HttpCache::new(cache_ttl).await,
agent_key_list,
agent_max_count,
node_geocache,
}
}
@@ -53,6 +57,10 @@ impl AppState {
pub(crate) fn agent_max_count(&self) -> i64 {
self.agent_max_count
}
pub(crate) fn node_geocache(&self) -> NodeGeoCache {
self.node_geocache.clone()
}
}
static GATEWAYS_LIST_KEY: &str = "gateways";
@@ -210,7 +218,11 @@ impl HttpCache {
.await
}
pub async fn get_nym_nodes_list(&self, db: &DbPool) -> anyhow::Result<Vec<ExtendedNymNode>> {
pub async fn get_nym_nodes_list(
&self,
db: &DbPool,
node_geocache: NodeGeoCache,
) -> anyhow::Result<Vec<ExtendedNymNode>> {
match self.nym_nodes.get(NYM_NODES_LIST_KEY).await {
Some(guard) => {
tracing::trace!("Fetching from cache...");
@@ -220,11 +232,13 @@ impl HttpCache {
None => {
tracing::trace!("No nym nodes in cache, refreshing cache from DB...");
let nym_nodes = aggregate_node_info_from_db(db).await.inspect(|nym_nodes| {
if nym_nodes.is_empty() {
tracing::warn!("Database contains 0 nym nodes");
}
})?;
let nym_nodes = aggregate_node_info_from_db(db, node_geocache)
.await
.inspect(|nym_nodes| {
if nym_nodes.is_empty() {
tracing::warn!("Database contains 0 nym nodes");
}
})?;
self.upsert_nym_node_list(nym_nodes.clone()).await;
Ok(nym_nodes)
@@ -346,7 +360,10 @@ impl HttpCache {
}
#[instrument(level = "info", skip_all)]
async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<ExtendedNymNode>> {
async fn aggregate_node_info_from_db(
pool: &DbPool,
node_geocache: NodeGeoCache,
) -> anyhow::Result<Vec<ExtendedNymNode>> {
let node_bond_info = queries::get_described_node_bond_info(pool).await?;
tracing::debug!("Described nodes with bond info: {}", node_bond_info.len());
@@ -359,9 +376,11 @@ async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<Extend
})?;
tracing::debug!("Skimmed nodes: {}", skimmed_nodes.len());
let described_nodes = queries::get_node_descriptions(pool).await?;
let described_nodes = queries::get_node_self_description(pool).await?;
tracing::debug!("Described nodes: {}", described_nodes.len());
let node_descriptions = queries::get_bonded_node_description(pool).await?;
let mut parsed_nym_nodes = Vec::new();
for (node_id, described_node) in described_nodes {
let bond_details = node_bond_info.get(&node_id);
@@ -404,6 +423,21 @@ async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<Extend
.map(|details| details.bond_information.owner.to_string())
.unwrap_or_default();
let node_description = node_descriptions.get(&node_id).cloned().unwrap_or_default();
let geoip = {
node_geocache.get(&node_id).await.map(|data| NodeGeoData {
city: data.city,
country: data.two_letter_iso_country_code,
ip_address: data.ip_address,
latitude: data.location.latitude.to_string(),
longitude: data.location.longitude.to_string(),
org: data.org,
postal: data.postal,
region: data.region,
timezone: data.timezone,
})
};
parsed_nym_nodes.push(ExtendedNymNode {
node_id,
identity_key,
@@ -415,8 +449,10 @@ async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<Extend
bonded,
node_type,
accepted_tnc,
description: serde_json::to_value(description).unwrap_or_default(),
self_description: serde_json::to_value(description).unwrap_or_default(),
rewarding_details: serde_json::to_value(rewarding_details).unwrap_or_default(),
description: node_description,
geoip,
});
}
nym-node-status-api/nym-node-status-api/src/main.rs
+8 -1
View File
@@ -37,8 +37,14 @@ async fn main() -> anyhow::Result<()> {
scraper.start().await;
});
// node geocache is shared between node monitor and HTTP server
let geocache = moka::future::Cache::builder()
.time_to_live(args.geodata_ttl)
.build();
// Start the monitor
let args_clone = args.clone();
let geocache_clone = geocache.clone();
tokio::spawn(async move {
monitor::spawn_in_background(
@@ -47,7 +53,7 @@ async fn main() -> anyhow::Result<()> {
args_clone.nyxd_addr,
args_clone.monitor_refresh_interval,
args_clone.ipinfo_api_token,
args_clone.geodata_ttl,
geocache_clone,
)
.await;
tracing::info!("Started monitor task");
@@ -67,6 +73,7 @@ async fn main() -> anyhow::Result<()> {
args.nym_http_cache_ttl,
agent_key_list.to_owned(),
args.max_agent_count,
geocache,
)
.await
.expect("Failed to start server");
nym-node-status-api/nym-node-status-api/src/monitor/geodata.rs
+28 -5
View File
@@ -29,6 +29,8 @@ impl IpInfoClient {
}
anyhow::Error::from(err)
})?;
// extracting text, then deserializing produces better error messages than response.json()
let raw_response = response.text().await?;
let response: LocationResponse =
serde_json::from_str(&raw_response).inspect_err(|e| tracing::error!("{e}"))?;
@@ -62,7 +64,7 @@ impl IpInfoClient {
}
#[derive(Debug, Clone, Serialize)]
pub(crate) struct NodeGeoData {
pub(crate) struct ExplorerPrettyBond {
pub(crate) identity_key: String,
pub(crate) owner: Addr,
pub(crate) pledge_amount: Coin,
@@ -74,6 +76,12 @@ pub(crate) struct Location {
pub(crate) two_letter_iso_country_code: String,
#[serde(flatten)]
pub(crate) location: Coordinates,
pub(crate) ip_address: String,
pub(crate) city: String,
pub(crate) region: String,
pub(crate) org: String,
pub(crate) postal: String,
pub(crate) timezone: String,
}
impl From<LocationResponse> for Location {
@@ -81,6 +89,12 @@ impl From<LocationResponse> for Location {
Self {
two_letter_iso_country_code: value.two_letter_iso_country_code,
location: value.loc,
ip_address: value.ip,
city: value.city,
region: value.region,
org: value.org,
postal: value.postal,
timezone: value.timezone,
}
}
}
@@ -91,6 +105,18 @@ pub(crate) struct LocationResponse {
pub(crate) two_letter_iso_country_code: String,
#[serde(deserialize_with = "deserialize_loc")]
pub(crate) loc: Coordinates,
#[serde(default = "String::default")]
pub(crate) ip: String,
#[serde(default = "String::default")]
pub(crate) city: String,
#[serde(default = "String::default")]
pub(crate) region: String,
#[serde(default = "String::default")]
pub(crate) org: String,
#[serde(default = "String::default")]
pub(crate) postal: String,
#[serde(default = "String::default")]
pub(crate) timezone: String,
}
fn deserialize_loc<'de, D>(deserializer: D) -> Result<Coordinates, D::Error>
@@ -115,10 +141,7 @@ pub(crate) struct Coordinates {
impl Location {
pub(crate) fn empty() -> Self {
Self {
two_letter_iso_country_code: String::new(),
location: Coordinates::default(),
}
Self::default()
}
}
nym-node-status-api/nym-node-status-api/src/monitor/mod.rs
+24 -27
View File
@@ -7,7 +7,7 @@ use crate::db::models::{
NYMNODES_DESCRIBED_COUNT, NYMNODE_COUNT,
};
use crate::db::{queries, DbPool};
use crate::monitor::geodata::{Location, NodeGeoData};
use crate::monitor::geodata::{ExplorerPrettyBond, Location};
use crate::utils::{decimal_to_i64, LogError, NumericalCheckedCast};
use anyhow::anyhow;
use moka::future::Cache;
@@ -31,8 +31,8 @@ pub(crate) use geodata::IpInfoClient;
mod geodata;
const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60);
static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw";
pub(crate) type NodeGeoCache = Cache<NodeId, Location>;
struct Monitor {
db_pool: DbPool,
@@ -40,7 +40,7 @@ struct Monitor {
nym_api_client_timeout: Duration,
nyxd_addr: Url,
ipinfo: IpInfoClient,
geocache: Cache<NodeId, Location>,
geocache: NodeGeoCache,
}
// TODO dz: query many NYM APIs:
@@ -52,9 +52,8 @@ pub(crate) async fn spawn_in_background(
nyxd_addr: Url,
refresh_interval: Duration,
ipinfo_api_token: String,
geodata_ttl: Duration,
geocache: NodeGeoCache,
) {
let geocache = Cache::builder().time_to_live(geodata_ttl).build();
let ipinfo = IpInfoClient::new(ipinfo_api_token.clone());
let mut monitor = Monitor {
db_pool,
@@ -151,18 +150,9 @@ impl Monitor {
tracing::debug!("{} nym nodes written to DB!", inserted);
})?;
let mut gateway_geodata = Vec::new();
for gateway in gateways.iter() {
if let Some(node_details) = bonded_nym_nodes.get(&gateway.node_id) {
let bond_info = &node_details.bond_information;
let gw_geodata = NodeGeoData {
identity_key: bond_info.node.identity_key.to_owned(),
owner: bond_info.owner.to_owned(),
pledge_amount: bond_info.original_pledge.to_owned(),
location: self.location_cached(gateway).await,
};
gateway_geodata.push(gw_geodata);
}
// refresh geodata for all nodes
for (_, node_description) in described_nodes.iter() {
self.location_cached(node_description).await;
}
let mixnodes_detailed = api_client
@@ -181,7 +171,6 @@ impl Monitor {
.map(|elem| elem.identity_key().to_owned())
.collect::<HashSet<_>>();
// TODO this assumes polyfilling of legacy mixnodes on Nym API
let mixnodes_legacy = nym_nodes
.iter()
.filter(|node| {
@@ -220,8 +209,9 @@ impl Monitor {
let assigned_mixing_count = mixing_assigned_nodes.len();
let count_legacy_mixnodes = mixnodes_legacy.len();
let gateway_records =
self.prepare_gateway_data(&gateways, gateway_geodata, &nym_nodes, &bonded_nym_nodes)?;
let gateway_records = self
.prepare_gateway_data(&gateways, &nym_nodes, &bonded_nym_nodes)
.await?;
let pool = self.db_pool.clone();
let gateways_count = gateway_records.len();
@@ -357,10 +347,9 @@ impl Monitor {
.collect::<Vec<_>>()
}
fn prepare_gateway_data(
&self,
async fn prepare_gateway_data(
&mut self,
described_gateways: &[&NymNodeDescription],
gateway_geodata: Vec<NodeGeoData>,
skimmed_gateways: &[SkimmedNode],
bonded_nodes: &HashMap<NodeId, NymNodeDetails>,
) -> anyhow::Result<Vec<GatewayInsertRecord>> {
@@ -373,11 +362,19 @@ impl Monitor {
let self_described = serde_json::to_string(&gateway.description)?;
let explorer_pretty_bond = gateway_geodata
.iter()
.find(|g| g.identity_key.eq(&identity_key));
let explorer_pretty_bond = {
let location = self.location_cached(gateway).await;
bonded_nodes
.get(&gateway.node_id)
.map(|details| ExplorerPrettyBond {
identity_key: gateway.ed25519_identity_key().to_base58_string(),
owner: details.bond_information.owner.to_owned(),
pledge_amount: details.bond_information.original_pledge.to_owned(),
location,
})
};
let explorer_pretty_bond =
explorer_pretty_bond.and_then(|g| serde_json::to_string(g).ok());
explorer_pretty_bond.and_then(|g| serde_json::to_string(&g).ok());
let performance = skimmed_gateways
.iter()