From 4f5118d3f48aaab171264e46b2b9afe19f48c5ef Mon Sep 17 00:00:00 2001 From: dynco-nym <173912580+dynco-nym@users.noreply.github.com> Date: Wed, 5 Mar 2025 23:30:00 +0100 Subject: [PATCH] Add /v3/nym-nodes - returns extended node info from local DB - endpoint caching - add bond_info & self_described to DB nym_nodes - update mixnode & gateway bond status on data refresh - add `active` column to DB nym_nodes - use only active & bonded nodes in scraping/testrun tasks --- Cargo.lock | 2 +- .../nym-node-status-agent/run.sh | 2 +- .../nym-node-status-api/Cargo.toml | 4 +- .../launch_node_status_api.sh | 5 - .../migrations/005_node_self_described.sql | 3 + .../nym-node-status-api/src/db/mod.rs | 10 +- .../nym-node-status-api/src/db/models.rs | 54 +++++- .../src/db/queries/gateways.rs | 21 +- .../src/db/queries/mixnodes.rs | 20 +- .../nym-node-status-api/src/db/queries/mod.rs | 11 +- .../src/db/queries/nym_nodes.rs | 181 ++++++++++++++---- .../src/db/queries/scraper.rs | 75 ++++---- .../nym-node-status-api/src/http/api/mod.rs | 2 + .../src/http/api/nym_nodes.rs | 42 ++++ .../nym-node-status-api/src/http/models.rs | 19 ++ .../nym-node-status-api/src/http/state.rs | 135 ++++++++++++- .../nym-node-status-api/src/monitor/mod.rs | 80 ++++++-- .../nym-node-status-api/src/testruns/queue.rs | 1 + 18 files changed, 547 insertions(+), 120 deletions(-) create mode 100644 nym-node-status-api/nym-node-status-api/migrations/005_node_self_described.sql create mode 100644 nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs diff --git a/Cargo.lock b/Cargo.lock index 00e49d7ca8..e55e95af49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6339,7 +6339,7 @@ dependencies = [ [[package]] name = "nym-node-status-api" -version = "1.0.2" +version = "1.0.3" dependencies = [ "ammonia", "anyhow", diff --git a/nym-node-status-api/nym-node-status-agent/run.sh b/nym-node-status-api/nym-node-status-agent/run.sh index a01e513a1c..544c6ca731 100755 --- a/nym-node-status-api/nym-node-status-agent/run.sh +++ b/nym-node-status-api/nym-node-status-agent/run.sh @@ -3,7 +3,7 @@ set -eu export ENVIRONMENT=${ENVIRONMENT:-"mainnet"} -probe_git_ref="nym-vpn-core-v1.3.2" +probe_git_ref="nym-vpn-core-v1.4.0" crate_root=$(dirname $(realpath "$0")) monorepo_root=$(realpath "${crate_root}/../..") diff --git a/nym-node-status-api/nym-node-status-api/Cargo.toml b/nym-node-status-api/nym-node-status-api/Cargo.toml index d3e70db0c6..2cd1ed4d72 100644 --- a/nym-node-status-api/nym-node-status-api/Cargo.toml +++ b/nym-node-status-api/nym-node-status-api/Cargo.toml @@ -3,9 +3,7 @@ [package] name = "nym-node-status-api" - - -version = "1.0.2" +version = "1.0.3" authors.workspace = true repository.workspace = true homepage.workspace = true diff --git a/nym-node-status-api/nym-node-status-api/launch_node_status_api.sh b/nym-node-status-api/nym-node-status-api/launch_node_status_api.sh index b8a16eb3ce..bd21bee53d 100755 --- a/nym-node-status-api/nym-node-status-api/launch_node_status_api.sh +++ b/nym-node-status-api/nym-node-status-api/launch_node_status_api.sh @@ -5,13 +5,8 @@ set -e user_rust_log_preference=$RUST_LOG export ENVIRONMENT=${ENVIRONMENT:-"mainnet"} export NYM_API_CLIENT_TIMEOUT=60 -export EXPLORER_CLIENT_TIMEOUT=60 export NODE_STATUS_API_TESTRUN_REFRESH_INTERVAL=120 -if [ "$ENVIRONMENT" = "mainnet" ]; then - export NYM_NODE_STATUS_API_HM_URL="https://harbourmaster.nymtech.net" -fi - # public counterpart of the agent's private key. # For TESTING only. NOT used in any other environment export NODE_STATUS_API_AGENT_KEY_LIST="H4z8kx5Kkf5JMQHhxaW1MwYndjKCDHC7HsVhHTFfBZ4J" diff --git a/nym-node-status-api/nym-node-status-api/migrations/005_node_self_described.sql b/nym-node-status-api/nym-node-status-api/migrations/005_node_self_described.sql new file mode 100644 index 0000000000..20acaba886 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/migrations/005_node_self_described.sql @@ -0,0 +1,3 @@ +ALTER TABLE nym_nodes ADD COLUMN self_described TEXT; +ALTER TABLE nym_nodes ADD COLUMN bond_info TEXT; +ALTER TABLE nym_nodes ADD COLUMN active INTEGER CHECK (active in (0, 1)) NOT NULL DEFAULT 0; diff --git a/nym-node-status-api/nym-node-status-api/src/db/mod.rs b/nym-node-status-api/nym-node-status-api/src/db/mod.rs index 86e25e9227..cc3398964c 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/mod.rs @@ -1,5 +1,9 @@ use anyhow::{anyhow, Result}; -use sqlx::{migrate::Migrator, sqlite::SqliteConnectOptions, ConnectOptions, SqlitePool}; +use sqlx::{ + migrate::Migrator, + sqlite::{SqliteAutoVacuum, SqliteConnectOptions, SqliteSynchronous}, + ConnectOptions, SqlitePool, +}; use std::str::FromStr; pub(crate) mod models; @@ -16,6 +20,10 @@ pub(crate) struct Storage { impl Storage { pub async fn init(connection_url: String) -> Result { let connect_options = SqliteConnectOptions::from_str(&connection_url)? + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .foreign_keys(true) .create_if_missing(true) .disable_statement_logging(); diff --git a/nym-node-status-api/nym-node-status-api/src/db/models.rs b/nym-node-status-api/nym-node-status-api/src/db/models.rs index 2bd0931ffa..5363040efc 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/models.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/models.rs @@ -2,20 +2,33 @@ use std::str::FromStr; use crate::{ http::{self, models::SummaryHistory}, - utils::NumericalCheckedCast, + utils::{decimal_to_i64, NumericalCheckedCast}, }; use anyhow::Context; use nym_contracts_common::Percent; use nym_crypto::asymmetric::{ed25519, x25519}; use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT; use nym_node_requests::api::v1::node::models::NodeDescription; -use nym_validator_client::nym_api::SkimmedNode; +use nym_validator_client::{ + client::NymNodeDetails, models::NymNodeDescription, nym_api::SkimmedNode, +}; use serde::{Deserialize, Serialize}; use sqlx::FromRow; use strum_macros::{EnumString, FromRepr}; use time::{Date, OffsetDateTime}; use utoipa::ToSchema; +macro_rules! serialize_opt_to_value { + ($var:expr) => {{ + match $var { + None => Ok(None), + Some(ref value) => serde_json::to_value(value).map(Some).map_err(|err| { + anyhow::anyhow!("Failed to serialize {}: {:?}", stringify!($var), err) + }), + } + }}; +} + pub(crate) struct GatewayInsertRecord { pub(crate) identity_key: String, pub(crate) bonded: bool, @@ -406,11 +419,11 @@ impl ScraperNodeInfo { } } +#[allow(dead_code)] // it's not dead code but clippy doesn't detect usage in sqlx macros #[derive(sqlx::Decode, Debug)] pub(crate) struct NymNodeDto { pub node_id: i64, pub ed25519_identity_pubkey: String, - #[allow(dead_code)] // it's not dead code but clippy doesn't detect usage in sqlx macros pub total_stake: i64, pub ip_addresses: serde_json::Value, pub mix_port: i64, @@ -419,8 +432,12 @@ pub(crate) struct NymNodeDto { pub supported_roles: serde_json::Value, pub entry: Option, pub performance: String, + pub self_described: Option, + pub bond_info: Option, + pub active: bool, } +#[allow(dead_code)] // it's not dead code but clippy doesn't detect usage in sqlx macros #[derive(Debug)] pub(crate) struct NymNodeInsertRecord { pub node_id: i64, @@ -433,13 +450,34 @@ pub(crate) struct NymNodeInsertRecord { pub supported_roles: serde_json::Value, pub performance: String, pub entry: Option, + pub self_described: Option, + pub bond_info: Option, + pub active: bool, pub last_updated_utc: String, } impl NymNodeInsertRecord { - pub fn new(skimmed_node: SkimmedNode, total_stake: i64) -> anyhow::Result { + pub fn new( + skimmed_node: SkimmedNode, + bond_info: Option<&NymNodeDetails>, + self_described: Option<&NymNodeDescription>, + ) -> anyhow::Result { let now = OffsetDateTime::now_utc().to_string(); + // if a node doesn't expose its self-described endpoint, it can't route traffic + // https://nym.com/docs/operators/nodes/nym-node/bonding + // same if it's not bonded in the mixnet smart contract + // https://nym.com/docs/operators/tokenomics/mixnet-rewards#rewarded-set-selection + let active = self_described.is_some() && bond_info.is_some(); + + // if bond info is missing, set stake to 0 + let total_stake = bond_info + .map(|info| decimal_to_i64(info.total_stake())) + .unwrap_or(0); + let entry = serialize_opt_to_value!(skimmed_node.entry)?; + let bond_info = serialize_opt_to_value!(bond_info)?; + let self_described = serialize_opt_to_value!(self_described)?; + let record = Self { node_id: skimmed_node.node_id.into(), ed25519_identity_pubkey: skimmed_node.ed25519_identity_pubkey.to_base58_string(), @@ -450,10 +488,10 @@ impl NymNodeInsertRecord { node_role: serde_json::to_value(&skimmed_node.role)?, supported_roles: serde_json::to_value(skimmed_node.supported_roles)?, performance: skimmed_node.performance.value().to_string(), - entry: match skimmed_node.entry { - Some(entry) => Some(serde_json::to_value(entry)?), - None => None, - }, + entry, + self_described, + bond_info, + active, last_updated_utc: now, }; diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs index cf23115ea1..92e9d7d849 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs @@ -30,11 +30,22 @@ pub(crate) async fn select_gateway_identity( Ok(record.gateway_identity_key) } -pub(crate) async fn insert_gateways( +pub(crate) async fn update_bonded_gateways( pool: &DbPool, gateways: Vec, ) -> anyhow::Result<()> { - let mut db = pool.acquire().await?; + let mut tx = pool.begin().await?; + + sqlx::query!( + r#"UPDATE + gateways + SET + bonded = false + "#, + ) + .execute(&mut *tx) + .await?; + for record in gateways { sqlx::query!( "INSERT INTO gateways @@ -55,10 +66,12 @@ pub(crate) async fn insert_gateways( record.last_updated_utc, record.performance ) - .execute(&mut *db) + .execute(&mut *tx) .await?; } + tx.commit().await?; + Ok(()) } @@ -101,7 +114,7 @@ pub(crate) async fn get_all_gateways(pool: &DbPool) -> anyhow::Result anyhow::Result> { +pub(crate) async fn get_bonded_gateway_id_keys(pool: &DbPool) -> anyhow::Result> { let mut conn = pool.acquire().await?; let items = sqlx::query!( r#" diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs index 67cb3d554f..3d2b3570bc 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs @@ -11,11 +11,21 @@ use crate::{ http::models::{DailyStats, Mixnode}, }; -pub(crate) async fn insert_mixnodes( +pub(crate) async fn update_bonded_mixnodes( pool: &DbPool, mixnodes: Vec, ) -> anyhow::Result<()> { - let mut conn = pool.acquire().await?; + let mut tx = pool.begin().await?; + + sqlx::query!( + r#"UPDATE + mixnodes + SET + bonded = false + "#, + ) + .execute(&mut *tx) + .await?; for record in mixnodes.iter() { // https://www.sqlite.org/lang_upsert.html @@ -43,10 +53,12 @@ pub(crate) async fn insert_mixnodes( record.last_updated_utc, record.is_dp_delegatee ) - .execute(&mut *conn) + .execute(&mut *tx) .await?; } + tx.commit().await?; + Ok(()) } @@ -126,7 +138,7 @@ pub(crate) async fn get_daily_stats(pool: &DbPool) -> anyhow::Result anyhow::Result> { +pub(crate) async fn get_bonded_mix_ids(pool: &DbPool) -> anyhow::Result> { let mut conn = pool.acquire().await?; let items = sqlx::query!( r#" diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs index 4103df77b4..3ef48accab 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs @@ -9,12 +9,17 @@ mod summary; pub(crate) mod testruns; pub(crate) use gateways::{ - get_all_gateway_id_keys, get_all_gateways, insert_gateways, select_gateway_identity, + get_all_gateways, get_bonded_gateway_id_keys, select_gateway_identity, update_bonded_gateways, }; pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_session_records}; pub(crate) use misc::insert_summaries; -pub(crate) use mixnodes::{get_all_mix_ids, get_all_mixnodes, get_daily_stats, insert_mixnodes}; -pub(crate) use nym_nodes::{get_nym_nodes, insert_nym_nodes}; +pub(crate) use mixnodes::{ + get_all_mixnodes, get_bonded_mix_ids, get_daily_stats, update_bonded_mixnodes, +}; +pub(crate) use nym_nodes::{ + get_active_node_bond_info, get_active_node_descriptions, get_active_nym_nodes, + get_all_nym_nodes, update_nym_nodes, +}; pub(crate) use packet_stats::{ get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats, }; diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs index fcfbe8d8e3..4d7394bc02 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs @@ -2,21 +2,21 @@ use std::collections::HashMap; use anyhow::Context; use futures_util::TryStreamExt; -use nym_validator_client::{client::NymNodeDetails, nym_api::SkimmedNode}; +use nym_validator_client::{ + client::{NodeId, NymNodeDetails}, + models::NymNodeDescription, +}; use tracing::instrument; -use crate::{ - db::{ - models::{NymNodeDto, NymNodeInsertRecord}, - DbPool, - }, - utils::decimal_to_i64, +use crate::db::{ + models::{NymNodeDto, NymNodeInsertRecord}, + DbPool, }; -pub(crate) async fn get_nym_nodes(pool: &DbPool) -> anyhow::Result> { +pub(crate) async fn get_all_nym_nodes(pool: &DbPool) -> anyhow::Result> { let mut conn = pool.acquire().await?; - let items = sqlx::query_as!( + sqlx::query_as!( NymNodeDto, r#"SELECT node_id, @@ -28,44 +28,72 @@ pub(crate) async fn get_nym_nodes(pool: &DbPool) -> anyhow::Result>() - .await?; + .await + .map_err(From::from) +} - let mut skimmed_nodes = Vec::new(); - for item in items { - let node_id = item.node_id; - match SkimmedNode::try_from(item) { - Ok(node) => skimmed_nodes.push(node), - Err(e) => { - tracing::warn!("Failed to decode node_id={}: {}", node_id, e); - } - } - } +pub(crate) async fn get_active_nym_nodes(pool: &DbPool) -> anyhow::Result> { + let mut conn = pool.acquire().await?; - Ok(skimmed_nodes) + sqlx::query_as!( + NymNodeDto, + r#"SELECT + node_id, + ed25519_identity_pubkey, + total_stake, + ip_addresses as "ip_addresses!: serde_json::Value", + mix_port, + x25519_sphinx_pubkey, + node_role as "node_role: serde_json::Value", + supported_roles as "supported_roles: serde_json::Value", + entry as "entry: serde_json::Value", + performance, + self_described as "self_described: serde_json::Value", + bond_info as "bond_info: serde_json::Value", + active as "active: bool" + FROM + nym_nodes + WHERE + active = true + "#, + ) + .fetch(&mut *conn) + .try_collect::>() + .await + .map_err(From::from) } #[instrument(level = "debug", skip_all)] -pub(crate) async fn insert_nym_nodes( +pub(crate) async fn update_nym_nodes( pool: &DbPool, - nym_nodes: Vec, - bonded_node_info: &HashMap, -) -> anyhow::Result<()> { - let mut conn = pool.acquire().await?; + node_records: Vec, +) -> anyhow::Result { + let mut tx = pool.begin().await?; - for nym_node in nym_nodes.into_iter() { - let total_stake = bonded_node_info - .get(&nym_node.node_id) - .map(|details| decimal_to_i64(details.total_stake())) - .unwrap_or(0); + // set inactive all nodes + sqlx::query!( + r#"UPDATE + nym_nodes + SET + active = false + "#, + ) + .execute(&mut *tx) + .await?; - let record = NymNodeInsertRecord::new(nym_node, total_stake)?; + // active nodes will get updated on insert + let inserted = node_records.len(); + for record in node_records { // https://www.sqlite.org/lang_upsert.html sqlx::query!( "INSERT INTO nym_nodes @@ -74,9 +102,12 @@ pub(crate) async fn insert_nym_nodes( ip_addresses, mix_port, x25519_sphinx_pubkey, node_role, supported_roles, entry, + self_described, + bond_info, + active, performance, last_updated_utc ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(node_id) DO UPDATE SET ed25519_identity_pubkey=excluded.ed25519_identity_pubkey, ip_addresses=excluded.ip_addresses, @@ -85,6 +116,9 @@ pub(crate) async fn insert_nym_nodes( node_role=excluded.node_role, supported_roles=excluded.supported_roles, entry=excluded.entry, + self_described=excluded.self_described, + bond_info=excluded.bond_info, + active=excluded.active, performance=excluded.performance, last_updated_utc=excluded.last_updated_utc ;", @@ -97,13 +131,88 @@ pub(crate) async fn insert_nym_nodes( record.node_role, record.supported_roles, record.entry, + record.self_described, + record.bond_info, + record.active, record.performance, record.last_updated_utc, ) - .execute(&mut *conn) + .execute(&mut *tx) .await .with_context(|| format!("node_id={}", record.node_id))?; } - Ok(()) + tx.commit().await?; + + Ok(inserted) +} + +pub(crate) async fn get_active_node_bond_info( + pool: &DbPool, +) -> anyhow::Result> { + let mut conn = pool.acquire().await?; + + sqlx::query!( + r#"SELECT + node_id, + bond_info as "bond_info: serde_json::Value" + FROM + nym_nodes + WHERE + bond_info IS NOT NULL + AND + active = true + "#, + ) + .fetch_all(&mut *conn) + .await + .map(|records| { + records + .into_iter() + .filter_map(|record| { + record + .bond_info + // only return details for nodes which have details stored + .and_then(|bond_info| serde_json::from_value::(bond_info).ok()) + .map(|res| (record.node_id as NodeId, res)) + }) + .collect::>() + }) + .map_err(From::from) +} + +pub(crate) async fn get_active_node_descriptions( + pool: &DbPool, +) -> anyhow::Result> { + let mut conn = pool.acquire().await?; + + sqlx::query!( + r#"SELECT + node_id, + self_described as "self_described: serde_json::Value" + FROM + nym_nodes + WHERE + self_described IS NOT NULL + AND + active = 1 + "#, + ) + .fetch_all(&mut *conn) + .await + .map(|records| { + records + .into_iter() + .filter_map(|record| { + record + .self_described + // only return details for nodes which have details stored + .and_then(|description| { + serde_json::from_value::(description).ok() + }) + .map(|res| (record.node_id as NodeId, res)) + }) + .collect::>() + }) + .map_err(From::from) } diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs index 52cff30a01..87d9fea229 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs @@ -7,45 +7,56 @@ use crate::{ }; use anyhow::Result; use chrono::Utc; +use nym_validator_client::nym_api::SkimmedNode; pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result> { let mut nodes_to_scrape = Vec::new(); - let mixnode_ids = queries::get_all_mix_ids(pool).await?; - let gateway_keys = queries::get_all_gateway_id_keys(pool).await?; + let mixnode_ids = queries::get_bonded_mix_ids(pool).await?; + let gateway_keys = queries::get_bonded_gateway_id_keys(pool).await?; let mut entry_exit_nodes = 0; - queries::get_nym_nodes(pool) - .await? - .into_iter() - .for_each(|node| { - // due to polyfilling, Nym nodes table might contain legacy mixnodes - // as well. Mark them as such here. - let node_kind = if mixnode_ids.contains(&node.node_id.into()) { - ScrapeNodeKind::LegacyMixnode { - mix_id: node.node_id.into(), + let skimmed_nodes = queries::get_active_nym_nodes(pool).await.map(|nodes_dto| { + nodes_dto.into_iter().filter_map(|node| { + let node_id = node.node_id; + match SkimmedNode::try_from(node) { + Ok(node) => Some(node), + Err(e) => { + tracing::error!("Failed to decode node_id={}: {}", node_id, e); + None } - } else if gateway_keys.contains(&node.ed25519_identity_pubkey.to_base58_string()) { - entry_exit_nodes += 1; - ScrapeNodeKind::EntryExitNymNode { - node_id: node.node_id.into(), - identity_key: node.ed25519_identity_pubkey.to_base58_string(), - } - } else { - ScrapeNodeKind::MixingNymNode { - node_id: node.node_id.into(), - } - }; - nodes_to_scrape.push(ScraperNodeInfo { - node_kind, - hosts: node - .ip_addresses - .into_iter() - .map(|ip| ip.to_string()) - .collect::>(), - http_api_port: node.mix_port.into(), - }) - }); + } + }) + })?; + + skimmed_nodes.for_each(|node| { + // TODO: relies on polyfilling: Nym nodes table might contain legacy mixnodes + // as well. Categorize them here. + let node_kind = if mixnode_ids.contains(&node.node_id.into()) { + ScrapeNodeKind::LegacyMixnode { + mix_id: node.node_id.into(), + } + } else if gateway_keys.contains(&node.ed25519_identity_pubkey.to_base58_string()) { + entry_exit_nodes += 1; + ScrapeNodeKind::EntryExitNymNode { + node_id: node.node_id.into(), + identity_key: node.ed25519_identity_pubkey.to_base58_string(), + } + } else { + ScrapeNodeKind::MixingNymNode { + node_id: node.node_id.into(), + } + }; + nodes_to_scrape.push(ScraperNodeInfo { + node_kind, + hosts: node + .ip_addresses + .into_iter() + .map(|ip| ip.to_string()) + .collect::>(), + http_api_port: node.mix_port.into(), + }) + }); tracing::debug!("Fetched {} 🌟 total nym nodes", nodes_to_scrape.len()); tracing::debug!("Fetched {} 🚪 entry/exit nodes", entry_exit_nodes); diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs b/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs index 6483a44519..1ba2711ec3 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs @@ -10,6 +10,7 @@ use crate::http::{server::HttpServer, state::AppState}; pub(crate) mod gateways; pub(crate) mod metrics; pub(crate) mod mixnodes; +pub(crate) mod nym_nodes; pub(crate) mod services; pub(crate) mod summary; pub(crate) mod testruns; @@ -38,6 +39,7 @@ impl RouterBuilder { .nest("/summary", summary::routes()) .nest("/metrics", metrics::routes()), ) + .nest("/v3", Router::new().nest("/nym-nodes", nym_nodes::routes())) .nest( "/internal", Router::new().nest("/testruns", testruns::routes()), diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs b/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs new file mode 100644 index 0000000000..0bcb3ceadc --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs @@ -0,0 +1,42 @@ +use axum::{ + extract::{Query, State}, + Json, Router, +}; +use tracing::instrument; + +use crate::http::{ + error::{HttpError, HttpResult}, + models::ExtendedNymNode, + state::AppState, + PagedResult, Pagination, +}; + +pub(crate) fn routes() -> Router { + Router::new().route("/", axum::routing::get(nym_nodes)) +} + +#[utoipa::path( + tag = "Nym Nodes", + get, + params( + Pagination + ), + path = "/v3/nym-nodes", + responses( + (status = 200, body = PagedResult) + ) +)] +#[instrument(level = tracing::Level::DEBUG, skip_all, fields(page=pagination.page, size=pagination.size))] +async fn nym_nodes( + Query(pagination): Query, + State(state): State, +) -> HttpResult>> { + let db = state.db_pool(); + + let nodes = state.cache().get_nym_nodes_list(db).await.map_err(|e| { + tracing::error!("{e}"); + HttpError::internal() + })?; + + Ok(Json(PagedResult::paginate(pagination, nodes))) +} diff --git a/nym-node-status-api/nym-node-status-api/src/http/models.rs b/nym-node-status-api/nym-node-status-api/src/http/models.rs index a3975b3a02..3ff313e450 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/models.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/models.rs @@ -1,4 +1,6 @@ +use cosmwasm_std::Decimal; use nym_node_requests::api::v1::node::models::NodeDescription; +use nym_validator_client::client::NodeId; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; @@ -45,6 +47,23 @@ pub struct Mixnode { pub last_updated_utc: String, } +#[derive(Clone, Debug, utoipa::ToSchema, Deserialize, Serialize)] +pub(crate) struct ExtendedNymNode { + pub(crate) node_id: NodeId, + pub(crate) identity_key: String, + pub(crate) uptime: f64, + #[schema(value_type = String)] + pub(crate) total_stake: Decimal, + pub(crate) original_pledge: u128, + pub(crate) bonding_address: String, + pub(crate) bonded: bool, + pub(crate) node_type: String, + pub(crate) ip_address: String, + pub(crate) accepted_tnc: bool, + pub(crate) description: serde_json::Value, + pub(crate) rewarding_details: serde_json::Value, +} + #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub struct DailyStats { pub date_utc: String, diff --git a/nym-node-status-api/nym-node-status-api/src/http/state.rs b/nym-node-status-api/nym-node-status-api/src/http/state.rs index f1aea18cb8..032ad49ff6 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/state.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/state.rs @@ -1,12 +1,16 @@ -use std::{sync::Arc, time::Duration}; +use std::{collections::HashMap, sync::Arc, time::Duration}; +use cosmwasm_std::Decimal; use moka::{future::Cache, Entry}; +use nym_contracts_common::NaiveFloat; use nym_crypto::asymmetric::ed25519::PublicKey; +use nym_validator_client::{models::DescribedNodeType, nym_api::SkimmedNode}; use tokio::sync::RwLock; +use tracing::instrument; use crate::{ - db::DbPool, - http::models::{DailyStats, Gateway, Mixnode, SummaryHistory}, + db::{queries, DbPool}, + http::models::{DailyStats, ExtendedNymNode, Gateway, Mixnode, SummaryHistory}, }; use super::models::SessionStats; @@ -53,6 +57,7 @@ impl AppState { static GATEWAYS_LIST_KEY: &str = "gateways"; static MIXNODES_LIST_KEY: &str = "mixnodes"; +static NYM_NODES_LIST_KEY: &str = "nym_nodes"; static MIXSTATS_LIST_KEY: &str = "mixstats"; static SUMMARY_HISTORY_LIST_KEY: &str = "summary-history"; static SESSION_STATS_LIST_KEY: &str = "session-stats"; @@ -63,6 +68,7 @@ const MIXNODE_STATS_HISTORY_DAYS: usize = 30; pub(crate) struct HttpCache { gateways: Cache>>>, mixnodes: Cache>>>, + nym_nodes: Cache>>>, mixstats: Cache>>>, history: Cache>>>, session_stats: Cache>>>, @@ -79,6 +85,10 @@ impl HttpCache { .max_capacity(2) .time_to_live(Duration::from_secs(ttl_seconds)) .build(), + nym_nodes: Cache::builder() + .max_capacity(2) + .time_to_live(Duration::from_secs(ttl_seconds)) + .build(), mixstats: Cache::builder() .max_capacity(2) .time_to_live(Duration::from_secs(ttl_seconds)) @@ -181,6 +191,47 @@ impl HttpCache { } } + pub async fn upsert_nym_node_list( + &self, + nym_node_list: Vec, + ) -> Entry>>> { + self.nym_nodes + .entry_by_ref(NYM_NODES_LIST_KEY) + .and_upsert_with(|maybe_entry| async { + if let Some(entry) = maybe_entry { + let v = entry.into_value(); + let mut guard = v.write().await; + *guard = nym_node_list; + v.clone() + } else { + Arc::new(RwLock::new(nym_node_list)) + } + }) + .await + } + + pub async fn get_nym_nodes_list(&self, db: &DbPool) -> anyhow::Result> { + match self.nym_nodes.get(NYM_NODES_LIST_KEY).await { + Some(guard) => { + tracing::trace!("Fetching from cache..."); + let read_lock = guard.read().await; + Ok(read_lock.clone()) + } + 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"); + } + })?; + self.upsert_nym_node_list(nym_nodes.clone()).await; + + Ok(nym_nodes) + } + } + } + pub async fn upsert_mixnode_stats( &self, mixnode_stats: Vec, @@ -293,3 +344,81 @@ impl HttpCache { .await } } + +#[instrument(level = "info", skip_all)] +async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result> { + let node_bond_info = queries::get_active_node_bond_info(pool).await?; + tracing::debug!("Active nodes with bond info: {}", node_bond_info.len()); + + let skimmed_nodes = queries::get_all_nym_nodes(pool).await.map(|records| { + records + .into_iter() + .filter_map(|dto| SkimmedNode::try_from(dto).ok()) + .map(|skimmed_node| (skimmed_node.node_id, skimmed_node)) + .collect::>() + })?; + tracing::debug!("Skimmed nodes: {}", skimmed_nodes.len()); + + let described_nodes = queries::get_active_node_descriptions(pool).await?; + tracing::debug!("Active described nodes: {}", described_nodes.len()); + + let mut parsed_nym_nodes = Vec::new(); + for (node_id, described_node) in described_nodes { + let bond_details = node_bond_info.get(&node_id); + let bonded = bond_details.is_some(); + let total_stake = bond_details + .map(|details| details.total_stake()) + .unwrap_or(Decimal::zero()); + let identity_key = described_node.ed25519_identity_key().to_string(); + + let original_pledge = bond_details + .map(|details| details.original_pledge().amount.u128()) + .unwrap_or(0u128); + let rewarding_details = &node_bond_info + .get(&node_id) + .map(|details| details.rewarding_details.clone()); + + let uptime = skimmed_nodes + .get(&node_id) + .map(|node| node.performance.naive_to_f64()) + .unwrap_or(0.0); + let node_type = match described_node.contract_node_type { + DescribedNodeType::NymNode => "nym_node".to_string(), + DescribedNodeType::LegacyMixnode => "legacy_mixnode".to_string(), + DescribedNodeType::LegacyGateway => "legacy_gateway".to_string(), + }; + let ip_address = described_node + .description + .host_information + .ip_address + .first() + .map(ToString::to_string) + .unwrap_or_default(); + let accepted_tnc = described_node + .description + .auxiliary_details + .accepted_operator_terms_and_conditions; + let description = described_node.description; + + let bonding_address = bond_details + .map(|details| details.bond_information.owner.to_string()) + .unwrap_or_default(); + + parsed_nym_nodes.push(ExtendedNymNode { + node_id, + identity_key, + total_stake, + uptime, + ip_address, + original_pledge, + bonding_address, + bonded, + node_type, + accepted_tnc, + description: serde_json::to_value(description).unwrap_or_default(), + rewarding_details: serde_json::to_value(rewarding_details).unwrap_or_default(), + }); + } + + Ok(parsed_nym_nodes) +} diff --git a/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs b/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs index 81711d7640..9c8dda42f5 100644 --- a/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs @@ -1,9 +1,10 @@ #![allow(deprecated)] use crate::db::models::{ - gateway, mixnode, GatewayInsertRecord, MixnodeRecord, NetworkSummary, ASSIGNED_ENTRY_COUNT, - ASSIGNED_EXIT_COUNT, ASSIGNED_MIXING_COUNT, GATEWAYS_BONDED_COUNT, GATEWAYS_HISTORICAL_COUNT, - MIXNODES_HISTORICAL_COUNT, MIXNODES_LEGACY_COUNT, NYMNODES_DESCRIBED_COUNT, NYMNODE_COUNT, + gateway, mixnode, GatewayInsertRecord, MixnodeRecord, NetworkSummary, NymNodeInsertRecord, + ASSIGNED_ENTRY_COUNT, ASSIGNED_EXIT_COUNT, ASSIGNED_MIXING_COUNT, GATEWAYS_BONDED_COUNT, + GATEWAYS_HISTORICAL_COUNT, MIXNODES_HISTORICAL_COUNT, MIXNODES_LEGACY_COUNT, + NYMNODES_DESCRIBED_COUNT, NYMNODE_COUNT, }; use crate::db::{queries, DbPool}; use crate::monitor::geodata::{Location, NodeGeoData}; @@ -11,7 +12,7 @@ use crate::utils::{decimal_to_i64, LogError, NumericalCheckedCast}; use anyhow::anyhow; use moka::future::Cache; use nym_network_defaults::NymNetworkDetails; -use nym_validator_client::client::{NodeId, NymApiClientExt}; +use nym_validator_client::client::{NodeId, NymApiClientExt, NymNodeDetails}; use nym_validator_client::models::{ LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription, }; @@ -107,19 +108,24 @@ impl Monitor { let described_nodes = api_client .get_all_described_nodes() .await - .log_error("get_all_described_nodes")?; + .log_error("get_all_described_nodes")? + .into_iter() + .map(|elem| (elem.node_id, elem)) + .collect::>(); tracing::info!("🟣 described nodes: {}", described_nodes.len()); let gateways = described_nodes .iter() - .filter(|node| { - node.description.declared_role.entry - || node.description.declared_role.exit_ipr - || node.description.declared_role.exit_nr + .filter_map(|(_, node)| { + if node.description.declared_role.entry || node.description.declared_role.exit_ipr { + Some(node) + } else { + None + } }) .collect::>(); - let bonded_node_info = api_client + let bonded_nym_nodes = api_client .get_all_bonded_nym_nodes() .await? .into_iter() @@ -127,22 +133,27 @@ impl Monitor { // for faster reads .collect::>(); - tracing::info!("🟣 bonded_nodes: {}", bonded_node_info.len()); + tracing::info!("🟣 bonded_nodes: {}", bonded_nym_nodes.len()); + // returns only bonded nodes let nym_nodes = api_client .get_all_basic_nodes() .await .log_error("get_all_basic_nodes")?; - queries::insert_nym_nodes(&self.db_pool, nym_nodes.clone(), &bonded_node_info) + tracing::info!("🟣 get_all_basic_nodes: {}", nym_nodes.len()); + + let nym_node_records = + self.prepare_nym_node_data(nym_nodes.clone(), &bonded_nym_nodes, &described_nodes); + queries::update_nym_nodes(&self.db_pool, nym_node_records) .await - .map(|_| { - tracing::debug!("{} nym nodes written to DB!", nym_nodes.len()); + .map(|inserted| { + 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_node_info.get(&gateway.node_id) { + 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(), @@ -170,6 +181,7 @@ impl Monitor { .map(|elem| elem.identity_key().to_owned()) .collect::>(); + // TODO this assumes polyfilling of legacy mixnodes on Nym API let mixnodes_legacy = nym_nodes .iter() .filter(|node| { @@ -208,11 +220,12 @@ 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)?; + let gateway_records = + self.prepare_gateway_data(&gateways, gateway_geodata, &nym_nodes, &bonded_nym_nodes)?; let pool = self.db_pool.clone(); let gateways_count = gateway_records.len(); - queries::insert_gateways(&pool, gateway_records) + queries::update_bonded_gateways(&pool, gateway_records) .await .map(|_| { tracing::debug!("{} gateway records written to DB!", gateways_count); @@ -224,7 +237,7 @@ impl Monitor { delegation_program_members, )?; let mixnodes_count = mixnode_records.len(); - queries::insert_mixnodes(&pool, mixnode_records) + queries::update_bonded_mixnodes(&pool, mixnode_records) .await .map(|_| { tracing::debug!("{} mixnode info written to DB!", mixnodes_count); @@ -317,17 +330,45 @@ impl Monitor { } } + fn prepare_nym_node_data( + &self, + skimmed_nodes: Vec, + bonded_node_info: &HashMap, + described_nodes: &HashMap, + ) -> Vec { + skimmed_nodes + .into_iter() + .filter_map(|skimmed_node| { + let node_id = skimmed_node.node_id; + let bond_info = bonded_node_info.get(&skimmed_node.node_id); + let self_described = described_nodes.get(&skimmed_node.node_id); + match NymNodeInsertRecord::new(skimmed_node, bond_info, self_described) { + Ok(record) => Some(record), + Err(err) => { + tracing::error!( + "Failed to create insert record for node {}: {}", + node_id, + err + ); + None + } + } + }) + .collect::>() + } + fn prepare_gateway_data( &self, described_gateways: &[&NymNodeDescription], gateway_geodata: Vec, skimmed_gateways: &[SkimmedNode], + bonded_nodes: &HashMap, ) -> anyhow::Result> { let mut gateway_records = Vec::new(); for gateway in described_gateways { let identity_key = gateway.ed25519_identity_key().to_base58_string(); - let bonded = true; + let bonded = bonded_nodes.contains_key(&gateway.node_id); let last_updated_utc = chrono::offset::Utc::now().timestamp(); let self_described = serde_json::to_string(&gateway.description)?; @@ -373,6 +414,7 @@ impl Monitor { for mixnode in mixnodes { let mix_id = mixnode.mix_id(); let identity_key = mixnode.identity_key(); + // only bonded nodes are given to this function let bonded = true; let total_stake = decimal_to_i64(mixnode.mixnode_details.total_stake()); let node_info = mixnode.mix_node(); diff --git a/nym-node-status-api/nym-node-status-api/src/testruns/queue.rs b/nym-node-status-api/nym-node-status-api/src/testruns/queue.rs index db1f72e3a7..6de0579741 100644 --- a/nym-node-status-api/nym-node-status-api/src/testruns/queue.rs +++ b/nym-node-status-api/nym-node-status-api/src/testruns/queue.rs @@ -24,6 +24,7 @@ pub(crate) async fn try_queue_testrun( explorer_pretty_bond as "explorer_pretty_bond?" FROM gateways WHERE gateway_identity_key = ? + AND bonded = true ORDER BY gateway_identity_key LIMIT 1"#, identity_key,