From dd3dcfa7fe0ffe9f98cc62ab3377e42365fd8b21 Mon Sep 17 00:00:00 2001 From: dynco-nym <173912580+dynco-nym@users.noreply.github.com> Date: Fri, 21 Feb 2025 20:32:39 +0100 Subject: [PATCH] Treat gateways as Nym Nodes (#5504) * Generate GW moniker if missing Beside that: - clear up gw nomenclature - adjust counting when legacy nodes are present in nym node APIs - create utils module * Store gatewy descriptions * Clippy & version --- Cargo.lock | 6 +- .../nym-node-status-api/Cargo.toml | 6 +- .../nym-node-status-api/src/db/models.rs | 28 +++-- .../src/db/queries/gateways.rs | 24 +++- .../src/db/queries/mixnodes.rs | 20 ++++ .../nym-node-status-api/src/db/queries/mod.rs | 8 +- .../src/db/queries/nym_nodes.rs | 6 +- .../src/db/queries/packet_stats.rs | 34 +++--- .../src/db/queries/scraper.rs | 98 ++++++++++++----- .../nym-node-status-api/src/main.rs | 1 + .../src/mixnet_scraper/helpers.rs | 57 +++++++--- .../src/mixnet_scraper/mod.rs | 44 ++++---- .../nym-node-status-api/src/monitor/mod.rs | 90 +++------------ .../src/node_scraper/mod.rs | 7 +- .../nym-node-status-api/src/utils.rs | 104 ++++++++++++++++++ nyx-chain-watcher/build.rs | 5 +- 16 files changed, 358 insertions(+), 180 deletions(-) create mode 100644 nym-node-status-api/nym-node-status-api/src/utils.rs diff --git a/Cargo.lock b/Cargo.lock index a1b81d40ca..4178cfb4de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6365,16 +6365,18 @@ dependencies = [ [[package]] name = "nym-node-status-api" -version = "1.0.0-rc.10" +version = "1.0.0" dependencies = [ "ammonia", "anyhow", "axum 0.7.9", + "bip39", "chrono", "clap", "cosmwasm-std", "envy", "futures-util", + "itertools 0.13.0", "moka", "nym-bin-common", "nym-contracts-common", @@ -6388,6 +6390,8 @@ dependencies = [ "nym-statistics-common", "nym-task", "nym-validator-client", + "rand 0.8.5", + "rand_chacha 0.3.1", "regex", "reqwest 0.12.4", "serde", 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 7123e12efa..8cca7e3829 100644 --- a/nym-node-status-api/nym-node-status-api/Cargo.toml +++ b/nym-node-status-api/nym-node-status-api/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "nym-node-status-api" -version = "1.0.0-rc.10" +version = "1.0.0" authors.workspace = true repository.workspace = true homepage.workspace = true @@ -16,11 +16,13 @@ rust-version.workspace = true ammonia = { workspace = true } anyhow = { workspace = true } axum = { workspace = true, features = ["tokio", "macros"] } +bip39 = { workspace = true } chrono = { workspace = true } clap = { workspace = true, features = ["cargo", "derive", "env", "string"] } cosmwasm-std = { workspace = true } envy = { workspace = true } futures-util = { workspace = true } +itertools = { workspace = true } moka = { workspace = true, features = ["future"] } nym-contracts-common = { path = "../../common/cosmwasm-smart-contracts/contracts-common" } nym-bin-common = { path = "../../common/bin-common", features = ["models"] } @@ -33,6 +35,8 @@ nym-statistics-common = { path = "../../common/statistics" } nym-validator-client = { path = "../../common/client-libs/validator-client" } nym-task = { path = "../../common/task" } nym-node-requests = { path = "../../nym-node/nym-node-requests", features = ["openapi"] } +rand = { workspace = true } +rand_chacha = { workspace = true } regex = { workspace = true } reqwest = { workspace = true } serde = { workspace = true, features = ["derive"] } 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 deb55a14fd..2bd0931ffa 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,7 +2,7 @@ use std::str::FromStr; use crate::{ http::{self, models::SummaryHistory}, - monitor::NumericalCheckedCast, + utils::NumericalCheckedCast, }; use anyhow::Context; use nym_contracts_common::Percent; @@ -16,7 +16,7 @@ use strum_macros::{EnumString, FromRepr}; use time::{Date, OffsetDateTime}; use utoipa::ToSchema; -pub(crate) struct GatewayRecord { +pub(crate) struct GatewayInsertRecord { pub(crate) identity_key: String, pub(crate) bonded: bool, pub(crate) self_described: String, @@ -360,14 +360,24 @@ impl TryFrom for http::models::SessionStats { } } -pub(crate) enum MixingNodeKind { - LegacyMixnode, - NymNode, +pub(crate) enum ScrapeNodeKind { + LegacyMixnode { mix_id: i64 }, + MixingNymNode { node_id: i64 }, + EntryExitNymNode { node_id: i64, identity_key: String }, +} + +impl ScrapeNodeKind { + pub(crate) fn node_id(&self) -> &i64 { + match self { + ScrapeNodeKind::LegacyMixnode { mix_id } => mix_id, + ScrapeNodeKind::MixingNymNode { node_id } => node_id, + ScrapeNodeKind::EntryExitNymNode { node_id, .. } => node_id, + } + } } pub(crate) struct ScraperNodeInfo { - pub node_id: i64, - pub node_kind: MixingNodeKind, + pub node_kind: ScrapeNodeKind, pub hosts: Vec, pub http_api_port: i64, } @@ -390,6 +400,10 @@ impl ScraperNodeInfo { urls } + + pub(crate) fn node_id(&self) -> &i64 { + self.node_kind.node_id() + } } #[derive(sqlx::Decode, Debug)] 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 9dc3ddbb11..cf23115ea1 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 @@ -1,6 +1,8 @@ +use std::collections::HashSet; + use crate::{ db::{ - models::{GatewayDto, GatewayRecord}, + models::{GatewayDto, GatewayInsertRecord}, DbPool, }, http::models::Gateway, @@ -30,7 +32,7 @@ pub(crate) async fn select_gateway_identity( pub(crate) async fn insert_gateways( pool: &DbPool, - gateways: Vec, + gateways: Vec, ) -> anyhow::Result<()> { let mut db = pool.acquire().await?; for record in gateways { @@ -98,3 +100,21 @@ pub(crate) async fn get_all_gateways(pool: &DbPool) -> anyhow::Result anyhow::Result> { + let mut conn = pool.acquire().await?; + let items = sqlx::query!( + r#" + SELECT gateway_identity_key + FROM gateways + WHERE bonded = true + "# + ) + .fetch_all(&mut *conn) + .await? + .into_iter() + .map(|record| record.gateway_identity_key) + .collect::>(); + + Ok(items) +} 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 7044941001..fb051ae4ad 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 @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use futures_util::TryStreamExt; use tracing::error; @@ -124,3 +126,21 @@ pub(crate) async fn get_daily_stats(pool: &DbPool) -> anyhow::Result anyhow::Result> { + let mut conn = pool.acquire().await?; + let items = sqlx::query!( + r#" + SELECT mix_id + FROM mixnodes + WHERE bonded = true + "# + ) + .fetch_all(&mut *conn) + .await? + .into_iter() + .map(|record| record.mix_id) + .collect::>(); + + Ok(items) +} 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 86b9475e25..4103df77b4 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 @@ -8,13 +8,15 @@ pub(crate) mod scraper; mod summary; pub(crate) mod testruns; -pub(crate) use gateways::{get_all_gateways, insert_gateways, select_gateway_identity}; +pub(crate) use gateways::{ + get_all_gateway_id_keys, get_all_gateways, insert_gateways, select_gateway_identity, +}; 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_mixnodes, get_daily_stats, insert_mixnodes}; +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 packet_stats::{ get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats, }; -pub(crate) use scraper::{get_mixing_nodes_for_scraping, insert_scraped_node_description}; +pub(crate) use scraper::{get_nodes_for_scraping, insert_scraped_node_description}; pub(crate) use summary::{get_summary, get_summary_history}; 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 c7dbf82ca9..fcfbe8d8e3 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 @@ -1,5 +1,6 @@ use std::collections::HashMap; +use anyhow::Context; use futures_util::TryStreamExt; use nym_validator_client::{client::NymNodeDetails, nym_api::SkimmedNode}; use tracing::instrument; @@ -9,7 +10,7 @@ use crate::{ models::{NymNodeDto, NymNodeInsertRecord}, DbPool, }, - monitor::decimal_to_i64, + utils::decimal_to_i64, }; pub(crate) async fn get_nym_nodes(pool: &DbPool) -> anyhow::Result> { @@ -100,7 +101,8 @@ pub(crate) async fn insert_nym_nodes( record.last_updated_utc, ) .execute(&mut *conn) - .await?; + .await + .with_context(|| format!("node_id={}", record.node_id))?; } Ok(()) diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/packet_stats.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/packet_stats.rs index fd9195ec0c..6c4fedc04a 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/packet_stats.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/packet_stats.rs @@ -1,27 +1,26 @@ use crate::db::{ - models::{MixingNodeKind, NodeStats, ScraperNodeInfo}, + models::{NodeStats, ScrapeNodeKind, ScraperNodeInfo}, DbPool, }; use anyhow::Result; pub(crate) async fn insert_node_packet_stats( pool: &DbPool, - node_id: i64, - node_kind: &MixingNodeKind, + node_kind: &ScrapeNodeKind, stats: &NodeStats, timestamp_utc: i64, ) -> Result<()> { let mut conn = pool.acquire().await?; match node_kind { - MixingNodeKind::LegacyMixnode => { + ScrapeNodeKind::LegacyMixnode { mix_id } => { sqlx::query!( r#" INSERT INTO mixnode_packet_stats_raw ( mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped ) VALUES (?, ?, ?, ?, ?) "#, - node_id, + mix_id, timestamp_utc, stats.packets_received, stats.packets_sent, @@ -30,7 +29,8 @@ pub(crate) async fn insert_node_packet_stats( .execute(&mut *conn) .await?; } - MixingNodeKind::NymNode => { + ScrapeNodeKind::MixingNymNode { node_id } + | ScrapeNodeKind::EntryExitNymNode { node_id, .. } => { sqlx::query!( r#" INSERT INTO nym_nodes_packet_stats_raw ( @@ -60,7 +60,7 @@ pub(crate) async fn get_raw_node_stats( let packets = match node.node_kind { // if no packets are found, it's fine to assume 0 because that's also // SQL default value if none provided - MixingNodeKind::LegacyMixnode => { + ScrapeNodeKind::LegacyMixnode { mix_id } => { sqlx::query_as!( NodeStats, r#" @@ -73,12 +73,13 @@ pub(crate) async fn get_raw_node_stats( ORDER BY timestamp_utc DESC LIMIT 1 OFFSET 1 "#, - node.node_id + mix_id ) .fetch_optional(&mut *conn) .await? } - MixingNodeKind::NymNode => { + ScrapeNodeKind::MixingNymNode { node_id } + | ScrapeNodeKind::EntryExitNymNode { node_id, .. } => { sqlx::query_as!( NodeStats, r#" @@ -91,7 +92,7 @@ pub(crate) async fn get_raw_node_stats( ORDER BY timestamp_utc DESC LIMIT 1 OFFSET 1 "#, - node.node_id + node_id ) .fetch_optional(&mut *conn) .await? @@ -110,7 +111,7 @@ pub(crate) async fn insert_daily_node_stats( let mut conn = pool.acquire().await?; match node.node_kind { - MixingNodeKind::LegacyMixnode => { + ScrapeNodeKind::LegacyMixnode { mix_id } => { let total_stake = sqlx::query_scalar!( r#" SELECT @@ -118,7 +119,7 @@ pub(crate) async fn insert_daily_node_stats( FROM mixnodes WHERE mix_id = ? "#, - node.node_id + mix_id ) .fetch_one(&mut *conn) .await?; @@ -136,7 +137,7 @@ pub(crate) async fn insert_daily_node_stats( packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent, packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped "#, - node.node_id, + mix_id, date_utc, total_stake, packets.packets_received, @@ -146,7 +147,8 @@ pub(crate) async fn insert_daily_node_stats( .execute(&mut *conn) .await?; } - MixingNodeKind::NymNode => { + ScrapeNodeKind::MixingNymNode { node_id } + | ScrapeNodeKind::EntryExitNymNode { node_id, .. } => { let total_stake = sqlx::query_scalar!( r#" SELECT @@ -154,7 +156,7 @@ pub(crate) async fn insert_daily_node_stats( FROM nym_nodes WHERE node_id = ? "#, - node.node_id + node_id ) .fetch_one(&mut *conn) .await?; @@ -172,7 +174,7 @@ pub(crate) async fn insert_daily_node_stats( packets_sent = nym_node_daily_mixing_stats.packets_sent + excluded.packets_sent, packets_dropped = nym_node_daily_mixing_stats.packets_dropped + excluded.packets_dropped "#, - node.node_id, + node_id, date_utc, total_stake, packets.packets_received, 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 2da881964f..52cff30a01 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 @@ -1,6 +1,6 @@ use crate::{ db::{ - models::{MixingNodeKind, ScraperNodeInfo}, + models::{ScrapeNodeKind, ScraperNodeInfo}, queries, DbPool, }, mixnet_scraper::helpers::NodeDescriptionResponse, @@ -8,16 +8,36 @@ use crate::{ use anyhow::Result; use chrono::Utc; -pub(crate) async fn get_mixing_nodes_for_scraping(pool: &DbPool) -> Result> { +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 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(), + } + } 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_id: node.node_id.into(), - node_kind: MixingNodeKind::NymNode, + node_kind, hosts: node .ip_addresses .into_iter() @@ -27,7 +47,8 @@ pub(crate) async fn get_mixing_nodes_for_scraping(pool: &DbPool) -> Result Result Result Result Result<()> { let timestamp = Utc::now().timestamp(); let mut conn = pool.acquire().await?; match node_kind { - MixingNodeKind::LegacyMixnode => { + ScrapeNodeKind::LegacyMixnode { mix_id } => { sqlx::query!( r#" INSERT INTO mixnode_description ( @@ -110,7 +124,7 @@ pub(crate) async fn insert_scraped_node_description( details = excluded.details, last_updated_utc = excluded.last_updated_utc "#, - node_id, + mix_id, description.moniker, description.website, description.security_contact, @@ -120,7 +134,7 @@ pub(crate) async fn insert_scraped_node_description( .execute(&mut *conn) .await?; } - MixingNodeKind::NymNode => { + ScrapeNodeKind::MixingNymNode { node_id } => { sqlx::query!( r#" INSERT INTO nym_node_descriptions ( @@ -143,6 +157,34 @@ pub(crate) async fn insert_scraped_node_description( .execute(&mut *conn) .await?; } + ScrapeNodeKind::EntryExitNymNode { identity_key, .. } => { + sqlx::query!( + r#" + INSERT INTO gateway_description ( + gateway_identity_key, + moniker, + website, + security_contact, + details, + last_updated_utc + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (gateway_identity_key) DO UPDATE SET + moniker = excluded.moniker, + website = excluded.website, + security_contact = excluded.security_contact, + details = excluded.details, + last_updated_utc = excluded.last_updated_utc + "#, + identity_key, + description.moniker, + description.website, + description.security_contact, + description.details, + timestamp, + ) + .execute(&mut *conn) + .await?; + } } Ok(()) diff --git a/nym-node-status-api/nym-node-status-api/src/main.rs b/nym-node-status-api/nym-node-status-api/src/main.rs index 459ae02dda..122a3b713f 100644 --- a/nym-node-status-api/nym-node-status-api/src/main.rs +++ b/nym-node-status-api/nym-node-status-api/src/main.rs @@ -10,6 +10,7 @@ mod mixnet_scraper; mod monitor; mod node_scraper; mod testruns; +mod utils; #[tokio::main] async fn main() -> anyhow::Result<()> { diff --git a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs index 6190978268..dfce27ad51 100644 --- a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs +++ b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs @@ -1,12 +1,15 @@ -use crate::db::{ - models::{NodeStats, ScraperNodeInfo}, - queries::{ - get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats, - insert_scraped_node_description, +use crate::{ + db::{ + models::{NodeStats, ScraperNodeInfo}, + queries::{ + get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats, + insert_scraped_node_description, + }, }, + utils::generate_node_name, }; use ammonia::Builder; -use anyhow::Result; +use anyhow::{anyhow, Result}; use chrono::{DateTime, Datelike, Utc}; use reqwest; use serde::{Deserialize, Serialize}; @@ -80,22 +83,33 @@ pub fn build_client() -> Result { .map_err(|e| anyhow::anyhow!("Failed to build HTTP client: {}", e)) } -pub fn sanitize_description(description: NodeDescriptionResponse) -> NodeDescriptionResponse { +pub fn sanitize_description( + description: NodeDescriptionResponse, + node_id: i64, +) -> NodeDescriptionResponse { let mut sanitizer = Builder::new(); sanitizer .tags(std::collections::HashSet::new()) .generic_attributes(std::collections::HashSet::new()) .url_schemes(std::collections::HashSet::new()); + const UNKNOWN: &str = "N/A"; let sanitize_field = |opt: Option| -> Option { Some( opt.filter(|s| !s.trim().is_empty()) - .map_or_else(|| "N/A".to_string(), |s| sanitizer.clean(&s).to_string()), + .map_or_else(|| UNKNOWN.to_string(), |s| sanitizer.clean(&s).to_string()), ) }; + let mut moniker = sanitize_field(description.moniker); + if let Some(sanitized) = &moniker { + if sanitized == UNKNOWN { + moniker = Some(generate_node_name(node_id)); + } + }; + NodeDescriptionResponse { - moniker: sanitize_field(description.moniker), + moniker, website: sanitize_field(description.website), security_contact: sanitize_field(description.security_contact), details: sanitize_field(description.details), @@ -108,18 +122,26 @@ pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeI let mut description = None; let mut error = None; + let mut tried_url_list = Vec::new(); for mut url in urls { url = format!("{}{}", url.trim_end_matches('/'), DESCRIPTION_URL); + tried_url_list.push(url.clone()); - match client.get(&url).send().await { + match client + .get(&url) + .send() + .await + // convert 404 and similar to error + .and_then(|res| res.error_for_status()) + { Ok(response) => { if let Ok(desc) = response.json::().await { description = Some(desc); break; } } - Err(e) => error = Some(e), + Err(e) => error = Some(anyhow!("{:?} ({})", tried_url_list, e)), } } @@ -128,9 +150,8 @@ pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeI anyhow::anyhow!("Failed to fetch description from any URL: {}", err_msg) })?; - let sanitized_description = sanitize_description(description); - insert_scraped_node_description(pool, &node.node_kind, node.node_id, &sanitized_description) - .await?; + let sanitized_description = sanitize_description(description, *node.node_id()); + insert_scraped_node_description(pool, &node.node_kind, &sanitized_description).await?; Ok(()) } @@ -144,9 +165,11 @@ pub async fn scrape_and_store_packet_stats( let mut stats = None; let mut error = None; + let mut tried_url_list = Vec::new(); for mut url in urls { url = format!("{}{}", url.trim_end_matches('/'), PACKET_STATS_URL); + tried_url_list.push(url.clone()); match client.get(&url).send().await { Ok(response) => { @@ -155,18 +178,18 @@ pub async fn scrape_and_store_packet_stats( break; } } - Err(e) => error = Some(e), + Err(e) => error = Some(anyhow!("{:?} ({})", tried_url_list, e)), } } let stats = stats.ok_or_else(|| { let err_msg = error.map_or_else(|| "Unknown error".to_string(), |e| e.to_string()); - anyhow::anyhow!("Failed to fetch stats from any URL: {}", err_msg) + anyhow::anyhow!("Failed to fetch description from any URL: {}", err_msg) })?; let timestamp = Utc::now(); let timestamp_utc = timestamp.timestamp(); - insert_node_packet_stats(pool, node.node_id, &node.node_kind, &stats, timestamp_utc).await?; + insert_node_packet_stats(pool, &node.node_kind, &stats, timestamp_utc).await?; // Update daily stats update_daily_stats(pool, node, timestamp, &stats).await?; diff --git a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs index 73169c2804..983775eec9 100644 --- a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs @@ -8,7 +8,7 @@ use sqlx::SqlitePool; use tracing::{debug, error, instrument, warn}; use crate::db::models::ScraperNodeInfo; -use crate::db::queries::get_mixing_nodes_for_scraping; +use crate::db::queries::get_nodes_for_scraping; const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); @@ -74,7 +74,7 @@ impl Scraper { pool: &SqlitePool, queue: Arc>>, ) -> Result<()> { - let nodes = get_mixing_nodes_for_scraping(pool).await?; + let nodes = get_nodes_for_scraping(pool).await?; if let Ok(mut queue_lock) = queue.lock() { queue_lock.extend(nodes); } else { @@ -82,7 +82,7 @@ impl Scraper { return Ok(()); } - Self::process_description_queue(pool, queue).await?; + Self::process_description_queue(pool, queue).await; Ok(()) } @@ -91,7 +91,7 @@ impl Scraper { pool: &SqlitePool, queue: Arc>>, ) -> Result<()> { - let nodes = get_mixing_nodes_for_scraping(pool).await?; + let nodes = get_nodes_for_scraping(pool).await?; tracing::info!("Querying {} mixing nodes", nodes.len()); if let Ok(mut queue_lock) = queue.lock() { queue_lock.extend(nodes); @@ -100,14 +100,11 @@ impl Scraper { return Ok(()); } - Self::process_packet_queue(pool, queue).await?; + Self::process_packet_queue(pool, queue).await; Ok(()) } - async fn process_description_queue( - pool: &SqlitePool, - queue: Arc>>, - ) -> Result<()> { + async fn process_description_queue(pool: &SqlitePool, queue: Arc>>) { loop { let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); @@ -132,12 +129,15 @@ impl Scraper { tokio::spawn(async move { match scrape_and_store_description(&pool, &node).await { Ok(_) => debug!( - "✅ Description task #{} for node {} complete", - task_id, node.node_id + "📝 ✅ Description task #{} for node {} complete", + task_id, + node.node_id() ), Err(e) => debug!( - "❌ Description task #{} for node {} failed: {}", - task_id, node.node_id, e + "📝 ❌ Description task #{} for node {} failed: {}", + task_id, + node.node_id(), + e ), } TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); @@ -146,13 +146,9 @@ impl Scraper { tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; } } - Ok(()) } - async fn process_packet_queue( - pool: &SqlitePool, - queue: Arc>>, - ) -> Result<()> { + async fn process_packet_queue(pool: &SqlitePool, queue: Arc>>) { loop { let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); @@ -177,12 +173,15 @@ impl Scraper { tokio::spawn(async move { match scrape_and_store_packet_stats(&pool, &node).await { Ok(_) => debug!( - "✅ Packet stats task #{} for node {} complete", - task_id, node.node_id + "📊 ✅ Packet stats task #{} for node {} complete", + task_id, + node.node_id() ), Err(e) => debug!( - "❌ Packet stats task #{} for node {} failed: {}", - task_id, node.node_id, e + "📊 ❌ Packet stats task #{} for node {} failed: {}", + task_id, + node.node_id(), + e ), } TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); @@ -191,6 +190,5 @@ impl Scraper { tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; } } - Ok(()) } } 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 afb77ed8d7..d771b6bf51 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,14 +1,14 @@ #![allow(deprecated)] use crate::db::models::{ - gateway, mixnode, GatewayRecord, MixnodeRecord, NetworkSummary, ASSIGNED_ENTRY_COUNT, + 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, }; use crate::db::{queries, DbPool}; use crate::monitor::geodata::{Location, NodeGeoData}; +use crate::utils::{decimal_to_i64, LogError, NumericalCheckedCast}; use anyhow::anyhow; -use cosmwasm_std::Decimal; use moka::future::Cache; use nym_network_defaults::NymNetworkDetails; use nym_validator_client::client::{NodeId, NymApiClientExt}; @@ -29,7 +29,6 @@ pub(crate) use geodata::IpInfoClient; mod geodata; -// TODO dz should be configurable const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw"; @@ -109,7 +108,11 @@ impl Monitor { let gateways = described_nodes .iter() - .filter(|node| node.description.declared_role.entry) + .filter(|node| { + node.description.declared_role.entry + || node.description.declared_role.exit_ipr + || node.description.declared_role.exit_nr + }) .collect::>(); let bonded_node_info = api_client @@ -120,6 +123,8 @@ impl Monitor { // for faster reads .collect::>(); + tracing::info!("🟣 bonded_nodes: {}", bonded_node_info.len()); + let nym_nodes = api_client .get_all_basic_nodes() .await @@ -198,10 +203,11 @@ impl Monitor { let gateway_records = self.prepare_gateway_data(&gateways, gateway_geodata, &nym_nodes)?; let pool = self.db_pool.clone(); + let gateways_count = gateway_records.len(); queries::insert_gateways(&pool, gateway_records) .await .map(|_| { - tracing::debug!("Gateway info written to DB!"); + tracing::debug!("{} gateway records written to DB!", gateways_count); })?; let mixnode_records = self.prepare_mixnode_data( @@ -209,10 +215,11 @@ impl Monitor { mixnodes_described, delegation_program_members, )?; + let mixnodes_count = mixnode_records.len(); queries::insert_mixnodes(&pool, mixnode_records) .await .map(|_| { - tracing::debug!("Mixnode info written to DB!"); + tracing::debug!("{} mixnode info written to DB!", mixnodes_count); })?; let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(&pool).await?; @@ -299,13 +306,13 @@ impl Monitor { fn prepare_gateway_data( &self, - gateways: &[&NymNodeDescription], + described_gateways: &[&NymNodeDescription], gateway_geodata: Vec, skimmed_gateways: &[SkimmedNode], - ) -> anyhow::Result> { + ) -> anyhow::Result> { let mut gateway_records = Vec::new(); - for gateway in gateways { + for gateway in described_gateways { let identity_key = gateway.ed25519_identity_key().to_base58_string(); let bonded = true; let last_updated_utc = chrono::offset::Utc::now().timestamp(); @@ -329,7 +336,7 @@ impl Monitor { .unwrap_or_default() .round_to_integer(); - gateway_records.push(GatewayRecord { + gateway_records.push(GatewayInsertRecord { identity_key: identity_key.to_owned(), bonded, self_described, @@ -400,33 +407,6 @@ impl Monitor { } } -// TODO dz is there a common monorepo place this can be put? -pub trait NumericalCheckedCast -where - T: TryFrom, - >::Error: std::error::Error, - Self: std::fmt::Display + Copy, -{ - fn cast_checked(self) -> anyhow::Result { - T::try_from(self).map_err(|e| { - anyhow::anyhow!( - "Couldn't cast {} to {}: {}", - self, - std::any::type_name::(), - e - ) - }) - } -} - -impl NumericalCheckedCast for T -where - U: TryFrom, - >::Error: std::error::Error, - T: std::fmt::Display + Copy, -{ -} - async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> { let mut conn = pool.acquire().await?; @@ -464,39 +444,3 @@ async fn get_delegation_program_details( Ok(mix_ids) } - -pub(crate) fn decimal_to_i64(decimal: Decimal) -> i64 { - // Convert the underlying Uint128 to a u128 - let atomics = decimal.atomics().u128(); - let precision = 1_000_000_000_000_000_000u128; - - // Get the fractional part - let fractional = atomics % precision; - - // Get the integer part - let integer = atomics / precision; - - // Combine them into a float - let float_value = integer as f64 + (fractional as f64 / 1_000_000_000_000_000_000_f64); - - // Limit to 6 decimal places - let rounded_value = (float_value * 1_000_000.0).round() / 1_000_000.0; - - rounded_value as i64 -} - -trait LogError { - fn log_error(self, msg: &str) -> Result; -} - -impl LogError for anyhow::Result -where - E: std::error::Error, -{ - fn log_error(self, msg: &str) -> Result { - if let Err(e) = &self { - tracing::error!("[{msg}]:\t{e}"); - } - self - } -} diff --git a/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs index 76f68a6549..04dd897098 100644 --- a/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs @@ -17,15 +17,14 @@ use tracing::instrument; mod error; const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); -const REFRESH_INTERVAL: Duration = Duration::from_secs(60 * 60 * 6); //6h, data only update once a day +const REFRESH_INTERVAL: Duration = Duration::from_secs(60 * 60 * 6); const STALE_DURATION: Duration = Duration::from_secs(86400 * 365); //one year -#[instrument(level = "debug", name = "node_scraper", skip_all)] +#[instrument(level = "info", name = "metrics_scraper", skip_all)] pub(crate) async fn spawn_in_background(db_pool: DbPool, nym_api_client_timeout: Duration) { let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env(); loop { - //No graceful shutdown? tracing::info!("Refreshing node self-described metrics..."); if let Err(e) = run(&db_pool, &network_defaults, nym_api_client_timeout).await { @@ -123,7 +122,7 @@ impl MetricsScrapingData { } } - #[instrument(level = "debug", name = "metrics_scraper", skip_all)] + #[instrument(level = "info", name = "metrics_scraper", skip_all)] async fn try_scrape_metrics(&self) -> Option { match self.try_get_client().await { Ok(client) => { diff --git a/nym-node-status-api/nym-node-status-api/src/utils.rs b/nym-node-status-api/nym-node-status-api/src/utils.rs new file mode 100644 index 0000000000..f315cee744 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/utils.rs @@ -0,0 +1,104 @@ +use cosmwasm_std::Decimal; +use itertools::Itertools; +use rand::prelude::SliceRandom; +use rand::SeedableRng; + +// pub(crate) fn generate_node_name(identity: ed25519::PublicKey) -> String { +pub(crate) fn generate_node_name(node_id: i64) -> String { + let seed = { + let node_id_bytes = node_id.to_le_bytes(); + let mut seed = [0u8; 32]; + for i in 0..4 { + seed[i * 8..(i + 1) * 8].copy_from_slice(&node_id_bytes); + } + seed + }; + let mut rng = rand_chacha::ChaCha20Rng::from_seed(seed); + let words = bip39::Language::English.word_list(); + words.choose_multiple(&mut rng, 3).join(" ") +} + +#[allow(clippy::items_after_test_module)] +#[cfg(test)] +mod test { + use rand::Rng; + + use super::*; + + #[test] + fn generate_node_name_should_be_deterministic() { + let mut rng = rand::thread_rng(); + + let node_id: i64 = rng.gen(); + let different_node_id: i64 = rng.gen(); + + let node_name = generate_node_name(node_id); + let node_name_different = generate_node_name(different_node_id); + assert_ne!(node_name, node_name_different); + + let node_name_same = generate_node_name(node_id); + assert_eq!(node_name, node_name_same); + } +} + +pub trait NumericalCheckedCast +where + T: TryFrom, + >::Error: std::error::Error, + Self: std::fmt::Display + Copy, +{ + fn cast_checked(self) -> anyhow::Result { + T::try_from(self).map_err(|e| { + anyhow::anyhow!( + "Couldn't cast {} to {}: {}", + self, + std::any::type_name::(), + e + ) + }) + } +} + +impl NumericalCheckedCast for T +where + U: TryFrom, + >::Error: std::error::Error, + T: std::fmt::Display + Copy, +{ +} + +pub(crate) fn decimal_to_i64(decimal: Decimal) -> i64 { + // Convert the underlying Uint128 to a u128 + let atomics = decimal.atomics().u128(); + let precision = 1_000_000_000_000_000_000u128; + + // Get the fractional part + let fractional = atomics % precision; + + // Get the integer part + let integer = atomics / precision; + + // Combine them into a float + let float_value = integer as f64 + (fractional as f64 / 1_000_000_000_000_000_000_f64); + + // Limit to 6 decimal places + let rounded_value = (float_value * 1_000_000.0).round() / 1_000_000.0; + + rounded_value as i64 +} + +pub(crate) trait LogError { + fn log_error(self, msg: &str) -> Result; +} + +impl LogError for anyhow::Result +where + E: std::error::Error, +{ + fn log_error(self, msg: &str) -> Result { + if let Err(e) = &self { + tracing::error!("[{msg}]:\t{e}"); + } + self + } +} diff --git a/nyx-chain-watcher/build.rs b/nyx-chain-watcher/build.rs index 5cfb413716..5cf16e56f8 100644 --- a/nyx-chain-watcher/build.rs +++ b/nyx-chain-watcher/build.rs @@ -1,13 +1,12 @@ use anyhow::Result; use sqlx::{sqlite::SqliteConnectOptions, Connection, SqliteConnection}; +use std::env::var; use std::io::Write; use std::{collections::HashMap, fs::File, path::PathBuf, str::FromStr}; #[tokio::main] async fn main() -> Result<()> { - let db_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join(".build") - .join("nyx_chain_watcher.sqlite"); + let db_path = PathBuf::from(var("OUT_DIR").unwrap()).join("nyx_chain_watcher.sqlite"); // Create the database directory if it doesn't exist if let Some(parent) = db_path.parent() {