diff --git a/Cargo.lock b/Cargo.lock index 790f8856f9..d43f524e7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6300,7 +6300,7 @@ dependencies = [ [[package]] name = "nym-node-status-api" -version = "1.0.0-rc.7" +version = "1.0.0-rc.8" dependencies = [ "ammonia", "anyhow", @@ -6312,6 +6312,7 @@ dependencies = [ "futures-util", "moka", "nym-bin-common", + "nym-contracts-common", "nym-crypto", "nym-explorer-client", "nym-network-defaults", diff --git a/common/http-api-client/src/lib.rs b/common/http-api-client/src/lib.rs index bd359b199d..bcd78136d8 100644 --- a/common/http-api-client/src/lib.rs +++ b/common/http-api-client/src/lib.rs @@ -768,7 +768,7 @@ where E: DeserializeOwned + Display, { let status = res.status(); - tracing::debug!("Status: {} (success: {})", &status, status.is_success()); + tracing::trace!("Status: {} (success: {})", &status, status.is_success()); if !allow_empty { if let Some(0) = res.content_length() { diff --git a/nym-node-status-api/.gitignore b/nym-node-status-api/.gitignore index db7ea3635a..4b2375049e 100644 --- a/nym-node-status-api/.gitignore +++ b/nym-node-status-api/.gitignore @@ -1,6 +1,7 @@ nym-node-status-agent/nym-gateway-probe nym-node-status-agent/keys/ data/ +settings.sql enter_db.sh nym-gateway-probe *.sqlite 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 81460981ec..400ac4cdbc 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.7" +version = "1.0.0-rc.8" authors.workspace = true repository.workspace = true homepage.workspace = true @@ -22,6 +22,7 @@ cosmwasm-std = { workspace = true } envy = { workspace = true } futures-util = { 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"] } nym-node-status-client = { path = "../nym-node-status-client" } nym-crypto = { path = "../../common/crypto", features = ["asymmetric", "serde"] } @@ -50,10 +51,6 @@ tracing-log = { workspace = true } tower-http = { workspace = true, features = ["cors", "trace"] } utoipa = { workspace = true, features = ["axum_extras", "time"] } utoipa-swagger-ui = { workspace = true, features = ["axum"] } -# TODO dz `cargo update async-trait` -# for automatic schema detection, which was merged, but not released yet -# https://github.com/ProbablyClem/utoipauto/pull/38 -# utoipauto = { git = "https://github.com/ProbablyClem/utoipauto", rev = "eb04cba" } utoipauto = { workspace = true } nym-node-metrics = { path = "../../nym-node/nym-node-metrics" } diff --git a/nym-node-status-api/nym-node-status-api/build.rs b/nym-node-status-api/nym-node-status-api/build.rs index 3a1b933bc4..1032fcb691 100644 --- a/nym-node-status-api/nym-node-status-api/build.rs +++ b/nym-node-status-api/nym-node-status-api/build.rs @@ -36,10 +36,18 @@ fn read_env_var(var: &str) -> Result { /// use `./enter_db.sh` to inspect DB async fn write_db_path_to_file(out_dir: &str, db_filename: &str) -> anyhow::Result<()> { + let mut file = File::create("settings.sql").await?; + let settings = ".mode columns +.headers on"; + file.write_all(settings.as_bytes()).await?; + let mut file = File::create("enter_db.sh").await?; - let _ = file.write(b"#!/bin/bash\n").await?; - file.write_all(format!("sqlite3 {}/{}", out_dir, db_filename).as_bytes()) - .await?; + let contents = format!( + "#!/bin/bash\n\ + sqlite3 -init settings.sql {}/{}", + out_dir, db_filename, + ); + file.write_all(contents.as_bytes()).await?; #[cfg(target_family = "unix")] file.set_permissions(Permissions::from_mode(0o755)) 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 9735f873d7..eca23c67a8 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 @@ -3,16 +3,19 @@ set -e user_rust_log_preference=$RUST_LOG +export ENVIRONMENT=${ENVIRONMENT:-"sandbox"} 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" -export ENVIRONMENT=${ENVIRONMENT:-"sandbox"} - script_dir=$(dirname $(realpath "$0")) monorepo_root=$(realpath "${script_dir}/../..") diff --git a/nym-node-status-api/nym-node-status-api/migrations/004_obsolete_fields.sql b/nym-node-status-api/nym-node-status-api/migrations/004_obsolete_fields.sql new file mode 100644 index 0000000000..f275ff0c33 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/migrations/004_obsolete_fields.sql @@ -0,0 +1,54 @@ +ALTER TABLE mixnodes DROP COLUMN blacklisted; +ALTER TABLE gateways DROP COLUMN blacklisted; + +CREATE TABLE nym_nodes ( + node_id INTEGER PRIMARY KEY, + ed25519_identity_pubkey VARCHAR NOT NULL UNIQUE, + total_stake INTEGER NOT NULL, + ip_addresses TEXT NOT NULL, + mix_port INTEGER NOT NULL, + x25519_sphinx_pubkey VARCHAR NOT NULL UNIQUE, + node_role TEXT NOT NULL, + supported_roles TEXT NOT NULL, + performance VARCHAR NOT NULL, + entry TEXT, + last_updated_utc INTEGER NOT NULL +); + +CREATE INDEX idx_nym_nodes_node_id ON nym_nodes (node_id); +CREATE INDEX idx_nym_nodes_ed25519_identity_pubkey ON nym_nodes (ed25519_identity_pubkey); + +CREATE TABLE nym_node_descriptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + node_id INTEGER UNIQUE NOT NULL, + moniker VARCHAR, + website VARCHAR, + security_contact VARCHAR, + details VARCHAR, + last_updated_utc INTEGER NOT NULL, + FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id) +); + +CREATE TABLE nym_nodes_packet_stats_raw ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + node_id INTEGER NOT NULL, + timestamp_utc INTEGER NOT NULL, + packets_received INTEGER, + packets_sent INTEGER, + packets_dropped INTEGER, + FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id) +); + +CREATE INDEX idx_nym_nodes_packet_stats_raw_node_id_timestamp_utc ON nym_nodes_packet_stats_raw (node_id, timestamp_utc); + +CREATE TABLE nym_node_daily_mixing_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + node_id INTEGER NOT NULL, + total_stake BIGINT NOT NULL, + date_utc VARCHAR NOT NULL, + packets_received INTEGER DEFAULT 0, + packets_sent INTEGER DEFAULT 0, + packets_dropped INTEGER DEFAULT 0, + FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id), + UNIQUE (node_id, date_utc) -- This constraint automatically creates an index +); diff --git a/nym-node-status-api/nym-node-status-api/src/cli/mod.rs b/nym-node-status-api/nym-node-status-api/src/cli/mod.rs index e9d76a1f75..291597bb7c 100644 --- a/nym-node-status-api/nym-node-status-api/src/cli/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/cli/mod.rs @@ -83,6 +83,9 @@ pub(crate) struct Cli { env = "NYM_NODE_STATUS_API_MAX_AGENT_COUNT" )] pub(crate) max_agent_count: i64, + + #[clap(long, default_value = "", env = "NYM_NODE_STATUS_API_HM_URL")] + pub(crate) hm_url: String, } fn parse_duration(arg: &str) -> Result { 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 7901b0f13c..deb55a14fd 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 @@ -1,18 +1,24 @@ +use std::str::FromStr; + use crate::{ http::{self, models::SummaryHistory}, monitor::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 serde::{Deserialize, Serialize}; use sqlx::FromRow; use strum_macros::{EnumString, FromRepr}; -use time::Date; +use time::{Date, OffsetDateTime}; use utoipa::ToSchema; pub(crate) struct GatewayRecord { pub(crate) identity_key: String, pub(crate) bonded: bool, - pub(crate) blacklisted: bool, pub(crate) self_described: String, // TODO dz shouldn't be an option pub(crate) explorer_pretty_bond: Option, @@ -24,7 +30,6 @@ pub(crate) struct GatewayRecord { pub(crate) struct GatewayDto { pub(crate) gateway_identity_key: String, pub(crate) bonded: bool, - pub(crate) blacklisted: bool, pub(crate) performance: i64, pub(crate) self_described: Option, pub(crate) explorer_pretty_bond: Option, @@ -69,7 +74,6 @@ impl TryFrom for http::models::Gateway { let last_probe_result = serde_json::from_str(&last_probe_result).unwrap_or(None); let bonded = value.bonded; - let blacklisted = value.blacklisted; let performance = value.performance as u8; let description = NodeDescription { @@ -82,7 +86,6 @@ impl TryFrom for http::models::Gateway { Ok(http::models::Gateway { gateway_identity_key: value.gateway_identity_key.clone(), bonded, - blacklisted, performance, self_described, explorer_pretty_bond, @@ -109,7 +112,6 @@ pub(crate) struct MixnodeRecord { pub(crate) total_stake: i64, pub(crate) host: String, pub(crate) http_port: u16, - pub(crate) blacklisted: bool, pub(crate) full_details: String, pub(crate) self_described: Option, pub(crate) last_updated_utc: i64, @@ -120,7 +122,6 @@ pub(crate) struct MixnodeRecord { pub(crate) struct MixnodeDto { pub(crate) mix_id: i64, pub(crate) bonded: bool, - pub(crate) blacklisted: bool, pub(crate) is_dp_delegatee: bool, pub(crate) total_stake: i64, pub(crate) full_details: String, @@ -147,7 +148,6 @@ impl TryFrom for http::models::Mixnode { let last_updated_utc = timestamp_as_utc(value.last_updated_utc.cast_checked()?).to_rfc3339(); - let blacklisted = value.blacklisted; let is_dp_delegatee = value.is_dp_delegatee; let moniker = value.moniker.clone(); let website = value.website.clone(); @@ -157,7 +157,6 @@ impl TryFrom for http::models::Mixnode { Ok(http::models::Mixnode { mix_id, bonded: value.bonded, - blacklisted, is_dp_delegatee, total_stake: value.total_stake, full_details, @@ -211,25 +210,27 @@ impl TryFrom for SummaryHistory { } } -pub(crate) const MIXNODES_BONDED_COUNT: &str = "mixnodes.bonded.count"; -pub(crate) const MIXNODES_BONDED_ACTIVE: &str = "mixnodes.bonded.active"; -pub(crate) const MIXNODES_BONDED_INACTIVE: &str = "mixnodes.bonded.inactive"; -pub(crate) const MIXNODES_BONDED_RESERVE: &str = "mixnodes.bonded.reserve"; -pub(crate) const MIXNODES_BLACKLISTED_COUNT: &str = "mixnodes.blacklisted.count"; +pub(crate) const MIXNODES_LEGACY_COUNT: &str = "mixnodes.legacy.count"; +pub(crate) const NYMNODES_DESCRIBED_COUNT: &str = "nymnode.described.count"; + +pub(crate) const NYMNODE_COUNT: &str = "nymnode.total.count"; +pub(crate) const ASSIGNED_ENTRY_COUNT: &str = "assigned.entry.count"; +pub(crate) const ASSIGNED_EXIT_COUNT: &str = "assigned.exit.count"; +pub(crate) const ASSIGNED_MIXING_COUNT: &str = "assigned.mixing.count"; pub(crate) const GATEWAYS_BONDED_COUNT: &str = "gateways.bonded.count"; -pub(crate) const GATEWAYS_BLACKLISTED_COUNT: &str = "gateways.blacklisted.count"; pub(crate) const MIXNODES_HISTORICAL_COUNT: &str = "mixnodes.historical.count"; pub(crate) const GATEWAYS_HISTORICAL_COUNT: &str = "gateways.historical.count"; -// `utoipa`` goes crazy if you use module-qualified prefix as field type so we +// `utoipa` goes crazy if you use module-qualified prefix as field type so we // have to import it use gateway::GatewaySummary; use mixnode::MixnodeSummary; #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub(crate) struct NetworkSummary { + pub(crate) total_nodes: i32, pub(crate) mixnodes: MixnodeSummary, pub(crate) gateways: GatewaySummary, } @@ -239,23 +240,15 @@ pub(crate) mod mixnode { #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub(crate) struct MixnodeSummary { - pub(crate) bonded: MixnodeSummaryBonded, - pub(crate) blacklisted: MixnodeSummaryBlacklisted, + pub(crate) bonded: MixingNodesSummary, pub(crate) historical: MixnodeSummaryHistorical, } #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] - pub(crate) struct MixnodeSummaryBonded { - pub(crate) count: i32, - pub(crate) active: i32, - pub(crate) inactive: i32, - pub(crate) reserve: i32, - pub(crate) last_updated_utc: String, - } - - #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] - pub(crate) struct MixnodeSummaryBlacklisted { + pub(crate) struct MixingNodesSummary { pub(crate) count: i32, + pub(crate) self_described: i32, + pub(crate) legacy: i32, pub(crate) last_updated_utc: String, } @@ -272,19 +265,15 @@ pub(crate) mod gateway { #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub(crate) struct GatewaySummary { pub(crate) bonded: GatewaySummaryBonded, - pub(crate) blacklisted: GatewaySummaryBlacklisted, - pub(crate) historical: GatewaySummaryHistorical, - } - #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] - pub(crate) struct GatewaySummaryExplorer { - pub(crate) count: i32, - pub(crate) last_updated_utc: String, + pub(crate) historical: GatewaySummaryHistorical, } #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub(crate) struct GatewaySummaryBonded { pub(crate) count: i32, + pub(crate) entry: i32, + pub(crate) exit: i32, pub(crate) last_updated_utc: String, } @@ -293,12 +282,6 @@ pub(crate) mod gateway { pub(crate) count: i32, pub(crate) last_updated_utc: String, } - - #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] - pub(crate) struct GatewaySummaryBlacklisted { - pub(crate) count: i32, - pub(crate) last_updated_utc: String, - } } #[allow(dead_code)] // not dead code, this is SQL data model @@ -377,8 +360,130 @@ impl TryFrom for http::models::SessionStats { } } +pub(crate) enum MixingNodeKind { + LegacyMixnode, + NymNode, +} + pub(crate) struct ScraperNodeInfo { pub node_id: i64, - pub host: String, + pub node_kind: MixingNodeKind, + pub hosts: Vec, pub http_api_port: i64, } + +impl ScraperNodeInfo { + pub(crate) fn contact_addresses(&self) -> Vec { + let mut urls = Vec::new(); + for host in &self.hosts { + urls.append(&mut vec![ + format!("http://{}:{}", host, DEFAULT_NYM_NODE_HTTP_PORT), + format!("http://{}:8000", host), + format!("https://{}", host), + format!("http://{}", host), + ]); + + if self.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 { + urls.insert(0, format!("http://{}:{}", host, self.http_api_port)); + } + } + + urls + } +} + +#[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, + pub x25519_sphinx_pubkey: String, + pub node_role: serde_json::Value, + pub supported_roles: serde_json::Value, + pub entry: Option, + pub performance: String, +} + +#[derive(Debug)] +pub(crate) struct NymNodeInsertRecord { + pub node_id: i64, + pub ed25519_identity_pubkey: String, + pub total_stake: i64, + pub ip_addresses: serde_json::Value, + pub mix_port: i64, + pub x25519_sphinx_pubkey: String, + pub node_role: serde_json::Value, + pub supported_roles: serde_json::Value, + pub performance: String, + pub entry: Option, + pub last_updated_utc: String, +} + +impl NymNodeInsertRecord { + pub fn new(skimmed_node: SkimmedNode, total_stake: i64) -> anyhow::Result { + let now = OffsetDateTime::now_utc().to_string(); + + let record = Self { + node_id: skimmed_node.node_id.into(), + ed25519_identity_pubkey: skimmed_node.ed25519_identity_pubkey.to_base58_string(), + total_stake, + ip_addresses: serde_json::to_value(&skimmed_node.ip_addresses)?, + mix_port: skimmed_node.mix_port as i64, + x25519_sphinx_pubkey: skimmed_node.x25519_sphinx_pubkey.to_base58_string(), + 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, + }, + last_updated_utc: now, + }; + + Ok(record) + } +} + +impl TryFrom for SkimmedNode { + type Error = anyhow::Error; + + fn try_from(other: NymNodeDto) -> Result { + let node_id = u32::try_from(other.node_id).context("Invalid node_id in DB")?; + let supported_roles = + serde_json::from_value(other.supported_roles).context("supported_roles")?; + let node_role = serde_json::from_value(other.node_role).context("node_role")?; + let ip_addresses = serde_json::from_value(other.ip_addresses).context("ip_addresses")?; + let entry = match other.entry { + Some(raw) => Some(serde_json::from_value(raw).context("entry")?), + None => None, + }; + + let skimmed_node = SkimmedNode { + node_id, + ed25519_identity_pubkey: ed25519::PublicKey::from_base58_string( + other.ed25519_identity_pubkey, + ) + .context("ed25519_identity_pubkey")?, + ip_addresses, + mix_port: other.mix_port.try_into()?, + x25519_sphinx_pubkey: x25519::PublicKey::from_base58_string(other.x25519_sphinx_pubkey) + .context("x25519_sphinx_pubkey")?, + role: node_role, + supported_roles, + entry, + performance: Percent::from_str(&other.performance).context("can't parse Percent")?, + }; + + Ok(skimmed_node) + } +} + +#[derive(Debug, Serialize, Deserialize, sqlx::Decode)] +pub struct NodeStats { + pub packets_received: i64, + pub packets_sent: i64, + pub packets_dropped: i64, +} 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 bcf9c2d6ca..9dc3ddbb11 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,12 +1,11 @@ use crate::{ db::{ - models::{BondedStatusDto, GatewayDto, GatewayRecord}, + models::{GatewayDto, GatewayRecord}, DbPool, }, http::models::Gateway, }; use futures_util::TryStreamExt; -use nym_validator_client::models::NymNodeDescription; use sqlx::{pool::PoolConnection, Sqlite}; use tracing::error; @@ -37,20 +36,18 @@ pub(crate) async fn insert_gateways( for record in gateways { sqlx::query!( "INSERT INTO gateways - (gateway_identity_key, bonded, blacklisted, + (gateway_identity_key, bonded, self_described, explorer_pretty_bond, last_updated_utc, performance) - VALUES (?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(gateway_identity_key) DO UPDATE SET bonded=excluded.bonded, - blacklisted=excluded.blacklisted, self_described=excluded.self_described, explorer_pretty_bond=excluded.explorer_pretty_bond, last_updated_utc=excluded.last_updated_utc, performance = excluded.performance;", record.identity_key, record.bonded, - record.blacklisted, record.self_described, record.explorer_pretty_bond, record.last_updated_utc, @@ -63,82 +60,6 @@ pub(crate) async fn insert_gateways( Ok(()) } -pub(crate) async fn write_blacklisted_gateways_to_db<'a, I>( - pool: &DbPool, - gateways: I, -) -> anyhow::Result<()> -where - I: Iterator, -{ - let mut conn = pool.acquire().await?; - for gateway_identity_key in gateways { - sqlx::query!( - "UPDATE gateways - SET blacklisted = true - WHERE gateway_identity_key = ?;", - gateway_identity_key, - ) - .execute(&mut *conn) - .await?; - } - - Ok(()) -} - -/// Ensure all gateways that are set as bonded, are still bonded -pub(crate) async fn ensure_gateways_still_bonded( - pool: &DbPool, - gateways: &[&NymNodeDescription], -) -> anyhow::Result { - let bonded_gateways_rows = get_all_bonded_gateways_row_ids_by_status(pool, true).await?; - let unbonded_gateways_rows = bonded_gateways_rows.iter().filter(|v| { - !gateways - .iter() - .any(|bonded| *bonded.ed25519_identity_key().to_base58_string() == v.identity_key) - }); - - let recently_unbonded_gateways = unbonded_gateways_rows.to_owned().count(); - let last_updated_utc = chrono::offset::Utc::now().timestamp(); - let mut transaction = pool.begin().await?; - for row in unbonded_gateways_rows { - sqlx::query!( - "UPDATE gateways - SET bonded = ?, last_updated_utc = ? - WHERE id = ?;", - false, - last_updated_utc, - row.id, - ) - .execute(&mut *transaction) - .await?; - } - transaction.commit().await?; - - Ok(recently_unbonded_gateways) -} - -async fn get_all_bonded_gateways_row_ids_by_status( - pool: &DbPool, - status: bool, -) -> anyhow::Result> { - let mut conn = pool.acquire().await?; - let items = sqlx::query_as!( - BondedStatusDto, - r#"SELECT - id as "id!", - gateway_identity_key as "identity_key!", - bonded as "bonded: bool" - FROM gateways - WHERE bonded = ?"#, - status, - ) - .fetch(&mut *conn) - .try_collect::>() - .await?; - - Ok(items) -} - pub(crate) async fn get_all_gateways(pool: &DbPool) -> anyhow::Result> { let mut conn = pool.acquire().await?; let items = sqlx::query_as!( @@ -146,9 +67,8 @@ pub(crate) async fn get_all_gateways(pool: &DbPool) -> anyhow::Result, ) -> anyhow::Result<()> { @@ -18,7 +18,7 @@ pub(crate) async fn insert_summaries( async fn insert_summary( pool: &DbPool, - summaries: &[(&str, &usize)], + summaries: &[(&str, usize)], last_updated: DateTime, ) -> anyhow::Result<()> { let timestamp = last_updated.timestamp(); 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 58af9bd429..09f185d5cd 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,10 +1,9 @@ use futures_util::TryStreamExt; -use nym_validator_client::models::MixNodeBondAnnotated; use tracing::error; use crate::{ db::{ - models::{BondedStatusDto, MixnodeDto, MixnodeRecord}, + models::{MixnodeDto, MixnodeRecord}, DbPool, }, http::models::{DailyStats, Mixnode}, @@ -21,13 +20,13 @@ pub(crate) async fn insert_mixnodes( sqlx::query!( "INSERT INTO mixnodes (mix_id, identity_key, bonded, total_stake, - host, http_api_port, blacklisted, full_details, + host, http_api_port, full_details, self_described, last_updated_utc, is_dp_delegatee) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(mix_id) DO UPDATE SET bonded=excluded.bonded, total_stake=excluded.total_stake, host=excluded.host, - http_api_port=excluded.http_api_port,blacklisted=excluded.blacklisted, + http_api_port=excluded.http_api_port, full_details=excluded.full_details,self_described=excluded.self_described, last_updated_utc=excluded.last_updated_utc, is_dp_delegatee = excluded.is_dp_delegatee;", @@ -37,7 +36,6 @@ pub(crate) async fn insert_mixnodes( record.total_stake, record.host, record.http_port, - record.blacklisted, record.full_details, record.self_described, record.last_updated_utc, @@ -57,7 +55,6 @@ pub(crate) async fn get_all_mixnodes(pool: &DbPool) -> anyhow::Result anyhow::Result anyhow::Result> { +/// `offset` = slides our fixed-day period further into the past by N days +pub(crate) async fn get_daily_stats(pool: &DbPool, offset: i64) -> anyhow::Result> { let mut conn = pool.acquire().await?; let items = sqlx::query_as!( DailyStats, r#" SELECT date_utc as "date_utc!", - packets_received as "total_packets_received!: i64", - packets_sent as "total_packets_sent!: i64", - packets_dropped as "total_packets_dropped!: i64", - total_stake as "total_stake!: i64" + SUM(total_stake) as "total_stake!: i64", + SUM(packets_received) as "total_packets_received!: i64", + SUM(packets_sent) as "total_packets_sent!: i64", + SUM(packets_dropped) as "total_packets_dropped!: i64" FROM ( SELECT date_utc, - SUM(packets_received) as packets_received, - SUM(packets_sent) as packets_sent, - SUM(packets_dropped) as packets_dropped, - SUM(total_stake) as total_stake - FROM mixnode_daily_stats - GROUP BY date_utc - ORDER BY date_utc DESC - LIMIT 30 + n.total_stake, + n.packets_received, + n.packets_sent, + n.packets_dropped + FROM nym_node_daily_mixing_stats n + UNION ALL + SELECT + m.date_utc, + m.total_stake, + m.packets_received, + m.packets_sent, + m.packets_dropped + FROM mixnode_daily_stats m + LEFT JOIN nym_node_daily_mixing_stats ON m.mix_id = nym_node_daily_mixing_stats.node_id + WHERE nym_node_daily_mixing_stats.node_id IS NULL ) GROUP BY date_utc - ORDER BY date_utc - "# + ORDER BY date_utc DESC + LIMIT 30 + OFFSET ? + "#, + offset ) .fetch(&mut *conn) .try_collect::>() @@ -121,57 +127,3 @@ pub(crate) async fn get_daily_stats(pool: &DbPool) -> anyhow::Result anyhow::Result { - let bonded_mixnodes_rows = get_all_bonded_mixnodes_row_ids_by_status(pool, true).await?; - let unbonded_mixnodes_rows = bonded_mixnodes_rows.iter().filter(|v| { - !mixnodes - .iter() - .any(|bonded| *bonded.mixnode_details.bond_information.identity() == v.identity_key) - }); - - let recently_unbonded_mixnodes = unbonded_mixnodes_rows.to_owned().count(); - let last_updated_utc = chrono::offset::Utc::now().timestamp(); - let mut transaction = pool.begin().await?; - for row in unbonded_mixnodes_rows { - sqlx::query!( - "UPDATE mixnodes - SET bonded = ?, last_updated_utc = ? - WHERE id = ?;", - false, - last_updated_utc, - row.id, - ) - .execute(&mut *transaction) - .await?; - } - transaction.commit().await?; - - Ok(recently_unbonded_mixnodes) -} - -async fn get_all_bonded_mixnodes_row_ids_by_status( - pool: &DbPool, - status: bool, -) -> anyhow::Result> { - let mut conn = pool.acquire().await?; - let items = sqlx::query_as!( - BondedStatusDto, - r#"SELECT - id as "id!", - identity_key as "identity_key!", - bonded as "bonded: bool" - FROM mixnodes - WHERE bonded = ?"#, - status, - ) - .fetch(&mut *conn) - .try_collect::>() - .await?; - - 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 1a8c4f579b..86b9475e25 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 @@ -2,19 +2,19 @@ mod gateways; mod gateways_stats; mod misc; mod mixnodes; +mod nym_nodes; +mod packet_stats; pub(crate) mod scraper; mod summary; pub(crate) mod testruns; -pub(crate) use gateways::{ - ensure_gateways_still_bonded, get_all_gateways, insert_gateways, select_gateway_identity, - write_blacklisted_gateways_to_db, -}; -pub(crate) use misc::insert_summaries; -pub(crate) use mixnodes::{ - ensure_mixnodes_still_bonded, get_all_mixnodes, get_daily_stats, insert_mixnodes, -}; -pub(crate) use scraper::fetch_active_nodes; -pub(crate) use summary::{get_summary, get_summary_history}; - +pub(crate) use gateways::{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 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 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 new file mode 100644 index 0000000000..c7dbf82ca9 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs @@ -0,0 +1,107 @@ +use std::collections::HashMap; + +use futures_util::TryStreamExt; +use nym_validator_client::{client::NymNodeDetails, nym_api::SkimmedNode}; +use tracing::instrument; + +use crate::{ + db::{ + models::{NymNodeDto, NymNodeInsertRecord}, + DbPool, + }, + monitor::decimal_to_i64, +}; + +pub(crate) async fn get_nym_nodes(pool: &DbPool) -> anyhow::Result> { + let mut conn = pool.acquire().await?; + + let items = 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 + FROM + nym_nodes + "#, + ) + .fetch(&mut *conn) + .try_collect::>() + .await?; + + 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); + } + } + } + + Ok(skimmed_nodes) +} + +#[instrument(level = "debug", skip_all)] +pub(crate) async fn insert_nym_nodes( + pool: &DbPool, + nym_nodes: Vec, + bonded_node_info: &HashMap, +) -> anyhow::Result<()> { + let mut conn = pool.acquire().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); + + let record = NymNodeInsertRecord::new(nym_node, total_stake)?; + // https://www.sqlite.org/lang_upsert.html + sqlx::query!( + "INSERT INTO nym_nodes + (node_id, ed25519_identity_pubkey, + total_stake, + ip_addresses, mix_port, + x25519_sphinx_pubkey, node_role, + supported_roles, entry, + performance, last_updated_utc + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(node_id) DO UPDATE SET + ed25519_identity_pubkey=excluded.ed25519_identity_pubkey, + ip_addresses=excluded.ip_addresses, + mix_port=excluded.mix_port, + x25519_sphinx_pubkey=excluded.x25519_sphinx_pubkey, + node_role=excluded.node_role, + supported_roles=excluded.supported_roles, + entry=excluded.entry, + performance=excluded.performance, + last_updated_utc=excluded.last_updated_utc + ;", + record.node_id, + record.ed25519_identity_pubkey, + record.total_stake, + record.ip_addresses, + record.mix_port, + record.x25519_sphinx_pubkey, + record.node_role, + record.supported_roles, + record.entry, + record.performance, + record.last_updated_utc, + ) + .execute(&mut *conn) + .await?; + } + + 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 new file mode 100644 index 0000000000..fd9195ec0c --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/packet_stats.rs @@ -0,0 +1,188 @@ +use crate::db::{ + models::{MixingNodeKind, NodeStats, ScraperNodeInfo}, + DbPool, +}; +use anyhow::Result; + +pub(crate) async fn insert_node_packet_stats( + pool: &DbPool, + node_id: i64, + node_kind: &MixingNodeKind, + stats: &NodeStats, + timestamp_utc: i64, +) -> Result<()> { + let mut conn = pool.acquire().await?; + + match node_kind { + MixingNodeKind::LegacyMixnode => { + sqlx::query!( + r#" + INSERT INTO mixnode_packet_stats_raw ( + mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?) + "#, + node_id, + timestamp_utc, + stats.packets_received, + stats.packets_sent, + stats.packets_dropped, + ) + .execute(&mut *conn) + .await?; + } + MixingNodeKind::NymNode => { + sqlx::query!( + r#" + INSERT INTO nym_nodes_packet_stats_raw ( + node_id, timestamp_utc, packets_received, packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?) + "#, + node_id, + timestamp_utc, + stats.packets_received, + stats.packets_sent, + stats.packets_dropped, + ) + .execute(&mut *conn) + .await?; + } + } + + Ok(()) +} + +pub(crate) async fn get_raw_node_stats( + pool: &DbPool, + node: &ScraperNodeInfo, +) -> Result> { + let mut conn = pool.acquire().await?; + + 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 => { + sqlx::query_as!( + NodeStats, + r#" + SELECT + COALESCE(packets_received, 0) as packets_received, + COALESCE(packets_sent, 0) as packets_sent, + COALESCE(packets_dropped, 0) as packets_dropped + FROM mixnode_packet_stats_raw + WHERE mix_id = ? + ORDER BY timestamp_utc DESC + LIMIT 1 OFFSET 1 + "#, + node.node_id + ) + .fetch_optional(&mut *conn) + .await? + } + MixingNodeKind::NymNode => { + sqlx::query_as!( + NodeStats, + r#" + SELECT + COALESCE(packets_received, 0) as packets_received, + COALESCE(packets_sent, 0) as packets_sent, + COALESCE(packets_dropped, 0) as packets_dropped + FROM nym_nodes_packet_stats_raw + WHERE node_id = ? + ORDER BY timestamp_utc DESC + LIMIT 1 OFFSET 1 + "#, + node.node_id + ) + .fetch_optional(&mut *conn) + .await? + } + }; + + Ok(packets) +} + +pub(crate) async fn insert_daily_node_stats( + pool: &DbPool, + node: &ScraperNodeInfo, + date_utc: &str, + packets: NodeStats, +) -> Result<()> { + let mut conn = pool.acquire().await?; + + match node.node_kind { + MixingNodeKind::LegacyMixnode => { + let total_stake = sqlx::query_scalar!( + r#" + SELECT + total_stake + FROM mixnodes + WHERE mix_id = ? + "#, + node.node_id + ) + .fetch_one(&mut *conn) + .await?; + + sqlx::query!( + r#" + INSERT INTO mixnode_daily_stats ( + mix_id, date_utc, + total_stake, packets_received, + packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(mix_id, date_utc) DO UPDATE SET + total_stake = excluded.total_stake, + packets_received = mixnode_daily_stats.packets_received + excluded.packets_received, + packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent, + packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped + "#, + node.node_id, + date_utc, + total_stake, + packets.packets_received, + packets.packets_sent, + packets.packets_dropped, + ) + .execute(&mut *conn) + .await?; + } + MixingNodeKind::NymNode => { + let total_stake = sqlx::query_scalar!( + r#" + SELECT + total_stake + FROM nym_nodes + WHERE node_id = ? + "#, + node.node_id + ) + .fetch_one(&mut *conn) + .await?; + + sqlx::query!( + r#" + INSERT INTO nym_node_daily_mixing_stats ( + node_id, date_utc, + total_stake, packets_received, + packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(node_id, date_utc) DO UPDATE SET + total_stake = nym_node_daily_mixing_stats.total_stake + excluded.total_stake, + packets_received = nym_node_daily_mixing_stats.packets_received + excluded.packets_received, + 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, + date_utc, + total_stake, + packets.packets_received, + packets.packets_sent, + packets.packets_dropped, + ) + .execute(&mut *conn) + .await?; + } + } + + Ok(()) +} 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 848273cc9c..2da881964f 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,20 +1,149 @@ -use crate::db::{models::ScraperNodeInfo, DbPool}; +use crate::{ + db::{ + models::{MixingNodeKind, ScraperNodeInfo}, + queries, DbPool, + }, + mixnet_scraper::helpers::NodeDescriptionResponse, +}; use anyhow::Result; +use chrono::Utc; + +pub(crate) async fn get_mixing_nodes_for_scraping(pool: &DbPool) -> Result> { + let mut nodes_to_scrape = Vec::new(); + + queries::get_nym_nodes(pool) + .await? + .into_iter() + .for_each(|node| { + nodes_to_scrape.push(ScraperNodeInfo { + node_id: node.node_id.into(), + node_kind: MixingNodeKind::NymNode, + hosts: node + .ip_addresses + .into_iter() + .map(|ip| ip.to_string()) + .collect::>(), + http_api_port: node.mix_port.into(), + }) + }); + + tracing::debug!("Fetched {} 🌟 nym nodes", nodes_to_scrape.len()); -pub(crate) async fn fetch_active_nodes(pool: &DbPool) -> Result> { let mut conn = pool.acquire().await?; - let nodes = sqlx::query_as!( - ScraperNodeInfo, + let mixnodes = sqlx::query!( r#" - SELECT mix_id as node_id, host, http_api_port - FROM mixnodes + SELECT mix_id as node_id, host, http_api_port + FROM mixnodes WHERE bonded = true "# ) .fetch_all(&mut *conn) .await?; + drop(conn); - Ok(nodes) + tracing::debug!("Fetched {} 🦖 mixnodes", nodes_to_scrape.len()); + + let mut duplicates = 0; + let mut legacy_not_in_nym_node_list = 0; + let total_legacy_mixnodes = mixnodes.len(); + for mixnode in mixnodes { + if nodes_to_scrape + .iter() + .all(|node| node.node_id != mixnode.node_id) + { + legacy_not_in_nym_node_list += 1; + } else { + duplicates += 1; + } + + // technically, mixnodes shouldn't be in nym_nodes table, but it's + // possible due to polyfilling on Nym API + if nodes_to_scrape + .iter() + .all(|node| node.node_id != mixnode.node_id) + { + nodes_to_scrape.push(ScraperNodeInfo { + node_id: mixnode.node_id, + node_kind: MixingNodeKind::LegacyMixnode, + hosts: vec![mixnode.host], + http_api_port: mixnode.http_api_port, + }) + } + } + tracing::debug!( + "{}/{} legacy mixnodes already included in nym_node list", + duplicates, + total_legacy_mixnodes + ); + tracing::debug!( + "{}/{} legacy mixnodes NOT included in nym_node list", + legacy_not_in_nym_node_list, + total_legacy_mixnodes + ); + tracing::debug!("In total: {} 🌟+🦖 mixing nodes", nodes_to_scrape.len()); + + Ok(nodes_to_scrape) } // TODO: add stuff for gateways + +pub(crate) async fn insert_scraped_node_description( + pool: &DbPool, + node_kind: &MixingNodeKind, + node_id: i64, + description: &NodeDescriptionResponse, +) -> Result<()> { + let timestamp = Utc::now().timestamp(); + let mut conn = pool.acquire().await?; + + match node_kind { + MixingNodeKind::LegacyMixnode => { + sqlx::query!( + r#" + INSERT INTO mixnode_description ( + mix_id, moniker, website, security_contact, details, last_updated_utc + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (mix_id) DO UPDATE SET + moniker = excluded.moniker, + website = excluded.website, + security_contact = excluded.security_contact, + details = excluded.details, + last_updated_utc = excluded.last_updated_utc + "#, + node_id, + description.moniker, + description.website, + description.security_contact, + description.details, + timestamp, + ) + .execute(&mut *conn) + .await?; + } + MixingNodeKind::NymNode => { + sqlx::query!( + r#" + INSERT INTO nym_node_descriptions ( + node_id, moniker, website, security_contact, details, last_updated_utc + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (node_id) DO UPDATE SET + moniker = excluded.moniker, + website = excluded.website, + security_contact = excluded.security_contact, + details = excluded.details, + last_updated_utc = excluded.last_updated_utc + "#, + node_id, + 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/db/queries/summary.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs index 4b2ecd22a5..ec23ad1500 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs @@ -6,15 +6,12 @@ use tracing::error; use crate::{ db::{ models::{ - gateway::{ - GatewaySummary, GatewaySummaryBlacklisted, GatewaySummaryBonded, - GatewaySummaryHistorical, - }, - mixnode::{ - MixnodeSummary, MixnodeSummaryBlacklisted, MixnodeSummaryBonded, - MixnodeSummaryHistorical, - }, - NetworkSummary, SummaryDto, SummaryHistoryDto, + gateway::{GatewaySummary, GatewaySummaryBonded, GatewaySummaryHistorical}, + mixnode::{MixingNodesSummary, MixnodeSummary, MixnodeSummaryHistorical}, + NetworkSummary, SummaryDto, SummaryHistoryDto, 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, }, DbPool, }, @@ -76,16 +73,6 @@ pub(crate) async fn get_summary(pool: &DbPool) -> HttpResult { } async fn from_summary_dto(items: Vec) -> HttpResult { - const MIXNODES_BONDED_COUNT: &str = "mixnodes.bonded.count"; - const MIXNODES_BONDED_ACTIVE: &str = "mixnodes.bonded.active"; - const MIXNODES_BONDED_INACTIVE: &str = "mixnodes.bonded.inactive"; - const MIXNODES_BONDED_RESERVE: &str = "mixnodes.bonded.reserve"; - const MIXNODES_BLACKLISTED_COUNT: &str = "mixnodes.blacklisted.count"; - const GATEWAYS_BONDED_COUNT: &str = "gateways.bonded.count"; - const GATEWAYS_BLACKLISTED_COUNT: &str = "gateways.blacklisted.count"; - const MIXNODES_HISTORICAL_COUNT: &str = "mixnodes.historical.count"; - const GATEWAYS_HISTORICAL_COUNT: &str = "gateways.historical.count"; - // convert database rows into a map by key let mut map = HashMap::new(); for item in items { @@ -94,15 +81,15 @@ async fn from_summary_dto(items: Vec) -> HttpResult // check we have all the keys we are expecting, and build up a map of errors for missing one let keys = [ + NYMNODE_COUNT, + ASSIGNED_MIXING_COUNT, + MIXNODES_LEGACY_COUNT, + NYMNODES_DESCRIBED_COUNT, GATEWAYS_BONDED_COUNT, - GATEWAYS_HISTORICAL_COUNT, - GATEWAYS_BLACKLISTED_COUNT, - MIXNODES_BLACKLISTED_COUNT, - MIXNODES_BONDED_ACTIVE, - MIXNODES_BONDED_COUNT, - MIXNODES_BONDED_INACTIVE, - MIXNODES_BONDED_RESERVE, + ASSIGNED_ENTRY_COUNT, + ASSIGNED_EXIT_COUNT, MIXNODES_HISTORICAL_COUNT, + GATEWAYS_HISTORICAL_COUNT, ]; let mut errors: Vec<&str> = vec![]; @@ -119,22 +106,17 @@ async fn from_summary_dto(items: Vec) -> HttpResult } // strip the options and use default values (anything missing is trapped above) - let mixnodes_bonded_count: SummaryDto = - map.get(MIXNODES_BONDED_COUNT).cloned().unwrap_or_default(); - let mixnodes_bonded_active: SummaryDto = - map.get(MIXNODES_BONDED_ACTIVE).cloned().unwrap_or_default(); - let mixnodes_bonded_inactive: SummaryDto = map - .get(MIXNODES_BONDED_INACTIVE) - .cloned() - .unwrap_or_default(); - let mixnodes_bonded_reserve: SummaryDto = map - .get(MIXNODES_BONDED_RESERVE) - .cloned() - .unwrap_or_default(); - let mixnodes_blacklisted_count: SummaryDto = map - .get(MIXNODES_BLACKLISTED_COUNT) + let total_nodes: SummaryDto = map.get(NYMNODE_COUNT).cloned().unwrap_or_default(); + let assigned_mixing_count: SummaryDto = + map.get(ASSIGNED_MIXING_COUNT).cloned().unwrap_or_default(); + let assigned_entry: SummaryDto = map.get(ASSIGNED_ENTRY_COUNT).cloned().unwrap_or_default(); + let assigned_exit: SummaryDto = map.get(ASSIGNED_EXIT_COUNT).cloned().unwrap_or_default(); + let self_described: SummaryDto = map + .get(NYMNODES_DESCRIBED_COUNT) .cloned() .unwrap_or_default(); + let legacy_mixnodes_count: SummaryDto = + map.get(MIXNODES_LEGACY_COUNT).cloned().unwrap_or_default(); let gateways_bonded_count: SummaryDto = map.get(GATEWAYS_BONDED_COUNT).cloned().unwrap_or_default(); let mixnodes_historical_count: SummaryDto = map @@ -145,23 +127,15 @@ async fn from_summary_dto(items: Vec) -> HttpResult .get(GATEWAYS_HISTORICAL_COUNT) .cloned() .unwrap_or_default(); - let gateways_blacklisted_count: SummaryDto = map - .get(GATEWAYS_BLACKLISTED_COUNT) - .cloned() - .unwrap_or_default(); Ok(NetworkSummary { + total_nodes: to_count_i32(&total_nodes), mixnodes: MixnodeSummary { - bonded: MixnodeSummaryBonded { - count: to_count_i32(&mixnodes_bonded_count), - active: to_count_i32(&mixnodes_bonded_active), - reserve: to_count_i32(&mixnodes_bonded_reserve), - inactive: to_count_i32(&mixnodes_bonded_inactive), - last_updated_utc: to_timestamp(&mixnodes_bonded_count), - }, - blacklisted: MixnodeSummaryBlacklisted { - count: to_count_i32(&mixnodes_blacklisted_count), - last_updated_utc: to_timestamp(&mixnodes_blacklisted_count), + bonded: MixingNodesSummary { + count: to_count_i32(&assigned_mixing_count), + self_described: to_count_i32(&self_described), + legacy: to_count_i32(&legacy_mixnodes_count), + last_updated_utc: to_timestamp(&assigned_mixing_count), }, historical: MixnodeSummaryHistorical { count: to_count_i32(&mixnodes_historical_count), @@ -171,12 +145,10 @@ async fn from_summary_dto(items: Vec) -> HttpResult gateways: GatewaySummary { bonded: GatewaySummaryBonded { count: to_count_i32(&gateways_bonded_count), + entry: to_count_i32(&assigned_entry), + exit: to_count_i32(&assigned_exit), last_updated_utc: to_timestamp(&gateways_bonded_count), }, - blacklisted: GatewaySummaryBlacklisted { - count: to_count_i32(&gateways_blacklisted_count), - last_updated_utc: to_timestamp(&gateways_blacklisted_count), - }, historical: GatewaySummaryHistorical { count: to_count_i32(&gateways_historical_count), last_updated_utc: to_timestamp(&gateways_historical_count), diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/mixnodes.rs b/nym-node-status-api/nym-node-status-api/src/http/api/mixnodes.rs index ae9842e1df..096c9fa27a 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/api/mixnodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/api/mixnodes.rs @@ -77,15 +77,33 @@ async fn get_mixnodes( } } +#[derive(Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +struct MixStatsQueryParams { + offset: Option, +} + #[utoipa::path( tag = "Mixnodes", get, path = "/v2/mixnodes/stats", + params( + MixStatsQueryParams + ), responses( (status = 200, body = Vec) ) )] -async fn get_stats(State(state): State) -> HttpResult>> { - let stats = state.cache().get_mixnode_stats(state.db_pool()).await; - Ok(Json(stats)) +#[instrument(level = "debug", skip(state))] +async fn get_stats( + Query(MixStatsQueryParams { offset }): Query, + State(state): State, +) -> HttpResult>> { + let offset = offset.unwrap_or(0); + let last_30_days = state + .cache() + .get_mixnode_stats(state.db_pool(), offset) + .await; + + Ok(Json(last_30_days)) } 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 276424d2b0..a3975b3a02 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 @@ -8,7 +8,6 @@ pub(crate) use nym_node_status_client::models::TestrunAssignment; pub struct Gateway { pub gateway_identity_key: String, pub bonded: bool, - pub blacklisted: bool, pub performance: u8, pub self_described: Option, pub explorer_pretty_bond: Option, @@ -38,7 +37,6 @@ pub struct GatewaySkinny { pub struct Mixnode { pub mix_id: u32, pub bonded: bool, - pub blacklisted: bool, pub is_dp_delegatee: bool, pub total_stake: i64, pub full_details: Option, diff --git a/nym-node-status-api/nym-node-status-api/src/http/server.rs b/nym-node-status-api/nym-node-status-api/src/http/server.rs index c8be0d63a1..b0613ca1c1 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/server.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/server.rs @@ -17,13 +17,20 @@ pub(crate) async fn start_http_api( nym_http_cache_ttl: u64, agent_key_list: Vec, agent_max_count: i64, + hm_url: String, ) -> anyhow::Result { let router_builder = RouterBuilder::with_default_routes(); - let state = AppState::new(db_pool, nym_http_cache_ttl, agent_key_list, agent_max_count); + let state = AppState::new( + db_pool, + nym_http_cache_ttl, + agent_key_list, + agent_max_count, + hm_url, + ) + .await; let router = router_builder.with_state(state); - // TODO dz do we need this to be configurable? let bind_addr = format!("0.0.0.0:{}", http_port); tracing::info!("Binding server to {bind_addr}"); let server = router.build_server(bind_addr).await?; 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 0c79c0e16f..a8a41fc73f 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 @@ -20,15 +20,16 @@ pub(crate) struct AppState { } impl AppState { - pub(crate) fn new( + pub(crate) async fn new( db_pool: DbPool, cache_ttl: u64, agent_key_list: Vec, agent_max_count: i64, + hm_url: String, ) -> Self { Self { db_pool, - cache: HttpCache::new(cache_ttl), + cache: HttpCache::new(cache_ttl, hm_url).await, agent_key_list, agent_max_count, } @@ -51,6 +52,90 @@ impl AppState { } } +#[derive(Debug, Clone)] +struct HistoricMixingStats { + historic_stats: Vec, +} + +impl HistoricMixingStats { + /// Collect historic stats only on initialization. From this point onwards, + /// service will collect its own stats + async fn init(hm_url: String) -> Self { + tracing::info!("Fetching historic mixnode stats from {}", hm_url); + + let target_url = format!("{}/v2/mixnodes/stats", hm_url); + if let Ok(response) = reqwest::get(&target_url) + .await + .and_then(|res| res.error_for_status()) + .inspect_err(|err| tracing::error!("Failed to fetch cache from HM: {}", err)) + { + if let Ok(mut daily_stats) = response.json::>().await { + // sorting required for seamless comparison later (descending, newest first) + daily_stats.sort_by(|left, right| right.date_utc.cmp(&left.date_utc)); + + tracing::info!( + "Successfully fetched {} historic entries from {}", + daily_stats.len(), + hm_url + ); + return Self { + historic_stats: daily_stats, + }; + } + }; + + tracing::warn!("Failed to get historic daily stats from {}", hm_url); + Self { + historic_stats: Vec::new(), + } + } + + /// polyfill with historical data obtained from Harbour Master + fn merge_with_historic_stats(&self, mut new_stats: Vec) -> Vec { + // newest first + new_stats.sort_by(|left, right| right.date_utc.cmp(&left.date_utc)); + + // historic stats are only used for dates when we don't have new data + let oldest_date_in_new_stats = new_stats + .last() + .map(|day| day.date_utc.to_owned()) + .unwrap_or(String::from("1900-01-01")); + + // given 2 arrays + // index historic_stats new_stats + // 0 30-01 31-01 + // 1 29-01 30-01 + // 2 28-01 + // ... + // N 01-01 + // cutoff point would be at historic_stats[1] + // (first date smaller than oldest we've already got) + if let Some(cutoff) = self + .historic_stats + .iter() + .position(|elem| elem.date_utc < oldest_date_in_new_stats) + { + // missing data = (all historic data) - (however many days we already have) + let missing_data = self.historic_stats.iter().skip(cutoff).cloned(); + + // extend new data with missing days + tracing::debug!( + "Polyfilled with {} historic records from {:?} to {:?}", + missing_data.len(), + self.historic_stats.last(), + self.historic_stats.get(cutoff) + ); + new_stats.extend(missing_data); + + // oldest first + new_stats.into_iter().rev().collect::>() + } else { + // if all historic data is older than what we've got, don't use it + new_stats + } + } +} + static GATEWAYS_LIST_KEY: &str = "gateways"; static MIXNODES_LIST_KEY: &str = "mixnodes"; static MIXSTATS_LIST_KEY: &str = "mixstats"; @@ -64,10 +149,11 @@ pub(crate) struct HttpCache { mixstats: Cache>>>, history: Cache>>>, session_stats: Cache>>>, + mixnode_historic_daily_stats: HistoricMixingStats, } impl HttpCache { - pub fn new(ttl_seconds: u64) -> Self { + pub async fn new(ttl_seconds: u64, hm_url: String) -> Self { HttpCache { gateways: Cache::builder() .max_capacity(2) @@ -89,6 +175,7 @@ impl HttpCache { .max_capacity(2) .time_to_live(Duration::from_secs(ttl_seconds)) .build(), + mixnode_historic_daily_stats: HistoricMixingStats::init(hm_url).await, } } @@ -114,12 +201,13 @@ impl HttpCache { pub async fn get_gateway_list(&self, db: &DbPool) -> Vec { match self.gateways.get(GATEWAYS_LIST_KEY).await { Some(guard) => { + tracing::trace!("Fetching from cache..."); let read_lock = guard.read().await; read_lock.clone() } None => { // the key is missing so populate it - tracing::warn!("No gateways in cache, refreshing cache from DB..."); + tracing::trace!("No gateways in cache, refreshing cache from DB..."); let gateways = crate::db::queries::get_all_gateways(db) .await @@ -157,11 +245,12 @@ impl HttpCache { pub async fn get_mixnodes_list(&self, db: &DbPool) -> Vec { match self.mixnodes.get(MIXNODES_LIST_KEY).await { Some(guard) => { + tracing::trace!("Fetching from cache..."); let read_lock = guard.read().await; read_lock.clone() } None => { - tracing::warn!("No mixnodes in cache, refreshing cache from DB..."); + tracing::trace!("No mixnodes in cache, refreshing cache from DB..."); let mixnodes = crate::db::queries::get_all_mixnodes(db) .await @@ -196,16 +285,22 @@ impl HttpCache { .await } - pub async fn get_mixnode_stats(&self, db: &DbPool) -> Vec { + pub async fn get_mixnode_stats(&self, db: &DbPool, offset: i64) -> Vec { match self.mixstats.get(MIXSTATS_LIST_KEY).await { Some(guard) => { let read_lock = guard.read().await; read_lock.to_vec() } None => { - let mixnode_stats = crate::db::queries::get_daily_stats(db) + let new_node_stats = crate::db::queries::get_daily_stats(db, offset) .await .unwrap_or_default(); + // for every day that's missing, fill it with cached historic data + let mut mixnode_stats = self + .mixnode_historic_daily_stats + .merge_with_historic_stats(new_node_stats); + mixnode_stats.truncate(30); + self.upsert_mixnode_stats(mixnode_stats.clone()).await; mixnode_stats } diff --git a/nym-node-status-api/nym-node-status-api/src/logging.rs b/nym-node-status-api/nym-node-status-api/src/logging.rs index 6bdad3bae5..8d94549885 100644 --- a/nym-node-status-api/nym-node-status-api/src/logging.rs +++ b/nym-node-status-api/nym-node-status-api/src/logging.rs @@ -33,6 +33,7 @@ pub(crate) fn setup_tracing_logger() -> anyhow::Result<()> { "tendermint_rpc", "tower_http", "axum", + "html5ever", ]; for crate_name in warn_crates { filter = filter.add_directive(directive_checked(format!("{}=warn", crate_name))?); @@ -40,28 +41,6 @@ pub(crate) fn setup_tracing_logger() -> anyhow::Result<()> { let log_level_hint = filter.max_level_hint(); - // debug or higher granularity (e.g. trace) - let debug_or_higher = std::cmp::max( - log_level_hint.unwrap_or(LevelFilter::DEBUG), - LevelFilter::DEBUG, - ); - filter = filter.add_directive(directive_checked(format!( - "nym_bin_common={}", - debug_or_higher - ))?); - filter = filter.add_directive(directive_checked(format!( - "nym_explorer_client={}", - debug_or_higher - ))?); - filter = filter.add_directive(directive_checked(format!( - "nym_network_defaults={}", - debug_or_higher - ))?); - filter = filter.add_directive(directive_checked(format!( - "nym_validator_client={}", - debug_or_higher - ))?); - log_builder.with_env_filter(filter).init(); tracing::info!("Log level: {:?}", log_level_hint); 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 afa93f6424..0cde5a440f 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 @@ -35,7 +35,6 @@ async fn main() -> anyhow::Result<()> { tokio::spawn(async move { scraper.start().await; }); - tracing::info!("Started node scraper task"); // Start the monitor let args_clone = args.clone(); @@ -67,6 +66,7 @@ async fn main() -> anyhow::Result<()> { args.nym_http_cache_ttl, agent_key_list.to_owned(), args.max_agent_count, + args.hm_url, ) .await .expect("Failed to start server"); 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 273dff2176..6190978268 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,20 +1,18 @@ -use crate::db::models::ScraperNodeInfo; +use crate::db::{ + models::{NodeStats, ScraperNodeInfo}, + queries::{ + get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats, + insert_scraped_node_description, + }, +}; use ammonia::Builder; use anyhow::Result; use chrono::{DateTime, Datelike, Utc}; -use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT; use reqwest; use serde::{Deserialize, Serialize}; use sqlx::SqlitePool; use std::time::Duration; -#[derive(Debug, Serialize, Deserialize)] -pub struct NodeStats { - pub packets_received: i64, - pub packets_sent: i64, - pub packets_dropped: i64, -} - #[derive(Debug, Serialize, Deserialize)] pub struct NodeDescriptionResponse { pub moniker: Option, @@ -106,16 +104,7 @@ pub fn sanitize_description(description: NodeDescriptionResponse) -> NodeDescrip pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> { let client = build_client()?; - let mut urls = vec![ - format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host), - format!("http://{}:8000", node.host), - format!("https://{}", node.host), - format!("http://{}", node.host), - ]; - - if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 { - urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port)); - } + let urls = node.contact_addresses(); let mut description = None; let mut error = None; @@ -140,31 +129,8 @@ pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeI })?; let sanitized_description = sanitize_description(description); - - let mut conn = pool.acquire().await?; - let timestamp = Utc::now().timestamp(); - - sqlx::query!( - r#" - INSERT INTO mixnode_description ( - mix_id, moniker, website, security_contact, details, last_updated_utc - ) VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT (mix_id) DO UPDATE SET - moniker = excluded.moniker, - website = excluded.website, - security_contact = excluded.security_contact, - details = excluded.details, - last_updated_utc = excluded.last_updated_utc - "#, - node.node_id, - sanitized_description.moniker, - sanitized_description.website, - sanitized_description.security_contact, - sanitized_description.details, - timestamp, - ) - .execute(&mut *conn) - .await?; + insert_scraped_node_description(pool, &node.node_kind, node.node_id, &sanitized_description) + .await?; Ok(()) } @@ -174,16 +140,7 @@ pub async fn scrape_and_store_packet_stats( node: &ScraperNodeInfo, ) -> Result<()> { let client = build_client()?; - let mut urls = vec![ - format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host), - format!("http://{}:8000", node.host), - format!("https://{}", node.host), - format!("http://{}", node.host), - ]; - - if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 { - urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port)); - } + let urls = node.contact_addresses(); let mut stats = None; let mut error = None; @@ -209,38 +166,20 @@ pub async fn scrape_and_store_packet_stats( let timestamp = Utc::now(); let timestamp_utc = timestamp.timestamp(); - let mut conn = pool.acquire().await?; - - // Store raw stats - sqlx::query!( - r#" - INSERT INTO mixnode_packet_stats_raw ( - mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped - ) VALUES (?, ?, ?, ?, ?) - "#, - node.node_id, - timestamp_utc, - stats.packets_received, - stats.packets_sent, - stats.packets_dropped, - ) - .execute(&mut *conn) - .await?; + insert_node_packet_stats(pool, node.node_id, &node.node_kind, &stats, timestamp_utc).await?; // Update daily stats - update_daily_stats(pool, node.node_id, timestamp, &stats).await?; + update_daily_stats(pool, node, timestamp, &stats).await?; Ok(()) } pub async fn update_daily_stats( pool: &SqlitePool, - node_id: i64, + node: &ScraperNodeInfo, timestamp: DateTime, current_stats: &NodeStats, ) -> Result<()> { - let mut conn = pool.acquire().await?; - let date_utc = format!( "{:04}-{:02}-{:02}", timestamp.year(), @@ -248,67 +187,29 @@ pub async fn update_daily_stats( timestamp.day() ); - let total_stake = sqlx::query_scalar!( - r#" - SELECT - total_stake - FROM mixnodes - WHERE mix_id = ? - "#, - node_id - ) - .fetch_one(&mut *conn) - .await?; - // Get previous stats - let previous_stats = sqlx::query!( - r#" - SELECT packets_received, packets_sent, packets_dropped - FROM mixnode_packet_stats_raw - WHERE mix_id = ? - ORDER BY timestamp_utc DESC - LIMIT 1 OFFSET 1 - "#, - node_id - ) - .fetch_optional(&mut *conn) - .await?; + let previous_stats = get_raw_node_stats(pool, node).await?; let (diff_received, diff_sent, diff_dropped) = if let Some(prev) = previous_stats { ( - calculate_packet_difference( - current_stats.packets_received, - prev.packets_received.unwrap_or(0), - ), - calculate_packet_difference(current_stats.packets_sent, prev.packets_sent.unwrap_or(0)), - calculate_packet_difference( - current_stats.packets_dropped, - prev.packets_dropped.unwrap_or(0), - ), + calculate_packet_difference(current_stats.packets_received, prev.packets_received), + calculate_packet_difference(current_stats.packets_sent, prev.packets_sent), + calculate_packet_difference(current_stats.packets_dropped, prev.packets_dropped), ) } else { (0, 0, 0) // No previous stats available }; - sqlx::query!( - r#" - INSERT INTO mixnode_daily_stats ( - mix_id, date_utc, total_stake, packets_received, packets_sent, packets_dropped - ) VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(mix_id, date_utc) DO UPDATE SET - total_stake = excluded.total_stake, - packets_received = mixnode_daily_stats.packets_received + excluded.packets_received, - packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent, - packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped - "#, - node_id, - date_utc, - total_stake, - diff_received, - diff_sent, - diff_dropped, + insert_daily_node_stats( + pool, + node, + &date_utc, + NodeStats { + packets_received: diff_received, + packets_sent: diff_sent, + packets_dropped: diff_dropped, + }, ) - .execute(&mut *conn) .await?; Ok(()) 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 f9fcd21b22..73169c2804 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 @@ -5,13 +5,13 @@ pub mod helpers; use anyhow::Result; use helpers::{scrape_and_store_description, scrape_and_store_packet_stats}; use sqlx::SqlitePool; -use tracing::{debug, error, warn}; +use tracing::{debug, error, instrument, warn}; use crate::db::models::ScraperNodeInfo; -use crate::db::queries::fetch_active_nodes; +use crate::db::queries::get_mixing_nodes_for_scraping; -const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); // 4 hours -const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); // 1 hour +const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); +const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); const QUEUE_CHECK_INTERVAL: Duration = Duration::from_millis(250); const MAX_CONCURRENT_TASKS: usize = 5; @@ -45,8 +45,9 @@ impl Scraper { tokio::spawn(async move { loop { if let Err(e) = Self::run_description_scraper(&pool, queue.clone()).await { - error!("Description scraper failed: {}", e); + error!(name: "description_scraper", "Description scraper failed: {}", e); } + debug!(name: "description_scraper", "Sleeping for {}s", DESCRIPTION_SCRAPE_INTERVAL.as_secs()); tokio::time::sleep(DESCRIPTION_SCRAPE_INTERVAL).await; } }); @@ -56,21 +57,24 @@ impl Scraper { let pool = self.pool.clone(); let queue = self.packet_queue.clone(); tracing::info!("Starting packet scraper"); + tokio::spawn(async move { loop { if let Err(e) = Self::run_packet_scraper(&pool, queue.clone()).await { - error!("Packet scraper failed: {}", e); + error!(name: "packet_scraper", "Packet scraper failed: {}", e); } + debug!(name: "packet_scraper", "Sleeping for {}s", PACKET_SCRAPE_INTERVAL.as_secs()); tokio::time::sleep(PACKET_SCRAPE_INTERVAL).await; } }); } + #[instrument(level = "info", name = "description_scraper", skip_all)] async fn run_description_scraper( pool: &SqlitePool, queue: Arc>>, ) -> Result<()> { - let nodes = fetch_active_nodes(pool).await?; + let nodes = get_mixing_nodes_for_scraping(pool).await?; if let Ok(mut queue_lock) = queue.lock() { queue_lock.extend(nodes); } else { @@ -82,12 +86,13 @@ impl Scraper { Ok(()) } + #[instrument(level = "info", name = "packet_scraper", skip_all)] async fn run_packet_scraper( pool: &SqlitePool, queue: Arc>>, ) -> Result<()> { - let nodes = fetch_active_nodes(pool).await?; - tracing::info!("Found {} active nodes", nodes.len()); + let nodes = get_mixing_nodes_for_scraping(pool).await?; + tracing::info!("Querying {} mixing nodes", nodes.len()); if let Ok(mut queue_lock) = queue.lock() { queue_lock.extend(nodes); } else { @@ -125,7 +130,7 @@ impl Scraper { let pool = pool.clone(); tokio::spawn(async move { - match Self::scrape_and_store_description(&pool, &node).await { + match scrape_and_store_description(&pool, &node).await { Ok(_) => debug!( "✅ Description task #{} for node {} complete", task_id, node.node_id @@ -170,7 +175,7 @@ impl Scraper { let pool = pool.clone(); tokio::spawn(async move { - match Self::scrape_and_store_packet_stats(&pool, &node).await { + match scrape_and_store_packet_stats(&pool, &node).await { Ok(_) => debug!( "✅ Packet stats task #{} for node {} complete", task_id, node.node_id @@ -188,15 +193,4 @@ impl Scraper { } Ok(()) } - - async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> { - scrape_and_store_description(pool, node).await - } - - async fn scrape_and_store_packet_stats( - pool: &SqlitePool, - node: &ScraperNodeInfo, - ) -> Result<()> { - scrape_and_store_packet_stats(pool, node).await - } } 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 5c8bc86deb..afb77ed8d7 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,10 +1,9 @@ #![allow(deprecated)] use crate::db::models::{ - gateway, mixnode, GatewayRecord, MixnodeRecord, NetworkSummary, GATEWAYS_BLACKLISTED_COUNT, - GATEWAYS_BONDED_COUNT, GATEWAYS_HISTORICAL_COUNT, MIXNODES_BLACKLISTED_COUNT, - MIXNODES_BONDED_ACTIVE, MIXNODES_BONDED_COUNT, MIXNODES_BONDED_INACTIVE, - MIXNODES_BONDED_RESERVE, MIXNODES_HISTORICAL_COUNT, + gateway, mixnode, GatewayRecord, 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}; @@ -16,7 +15,7 @@ use nym_validator_client::client::{NodeId, NymApiClientExt}; use nym_validator_client::models::{ LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription, }; -use nym_validator_client::nym_nodes::SkimmedNode; +use nym_validator_client::nym_nodes::{NodeRole, SkimmedNode}; use nym_validator_client::nyxd::contract_traits::PagedMixnetQueryClient; use nym_validator_client::nyxd::{AccountId, NyxdClient}; use nym_validator_client::NymApiClient; @@ -102,104 +101,101 @@ impl Monitor { let api_client = NymApiClient::new_with_timeout(default_api_url, self.nym_api_client_timeout); - let all_nodes = api_client + let described_nodes = api_client .get_all_described_nodes() .await .log_error("get_all_described_nodes")?; - tracing::debug!("Fetched {} total nodes", all_nodes.len()); + tracing::info!("🟣 described nodes: {}", described_nodes.len()); - let gateways = all_nodes + let gateways = described_nodes .iter() .filter(|node| node.description.declared_role.entry) .collect::>(); - tracing::debug!( - "{}/{} with declared entry gateway capability", - gateways.len(), - all_nodes.len() - ); - - let mixnodes = all_nodes - .iter() - .filter(|node| node.description.declared_role.mixnode) - .collect::>(); - tracing::debug!( - "{}/{} with declared mixnode capability", - mixnodes.len(), - all_nodes.len() - ); let bonded_node_info = api_client .get_all_bonded_nym_nodes() .await? .into_iter() - .map(|node| (node.bond_information.node_id, node.bond_information)) + .map(|node| (node.bond_information.node_id, node)) // for faster reads .collect::>(); + 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).await?; + let mut gateway_geodata = Vec::new(); for gateway in gateways.iter() { - if let Some(node_info) = bonded_node_info.get(&gateway.node_id) { + if let Some(node_details) = bonded_node_info.get(&gateway.node_id) { + let bond_info = &node_details.bond_information; let gw_geodata = NodeGeoData { - identity_key: node_info.node.identity_key.to_owned(), - owner: node_info.owner.to_owned(), - pledge_amount: node_info.original_pledge.to_owned(), + 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); } } - // contains performance data - let all_skimmed_nodes = api_client - .get_all_basic_nodes() - .await - .log_error("get_all_basic_nodes")?; - - let gateways_blacklisted = all_skimmed_nodes - .iter() - .filter_map(|node| { - if node.performance.round_to_integer() <= 50 && node.supported_roles.entry { - Some(node.ed25519_identity_pubkey.to_base58_string()) - } else { - None - } - }) - .collect::>(); - - // Cached mixnodes don't include blacklisted nodes - // We need that to calculate the total locked tokens later - // TODO dz deprecated API, remove - let legacy_mixnodes = api_client + let mixnodes_detailed = api_client .nym_api .get_mixnodes_detailed_unfiltered() .await .log_error("get_mixnodes_detailed_unfiltered")?; + + tracing::info!( + "🟣 mixnodes_detailed_unfiltered: {}", + mixnodes_detailed.len() + ); + + let mixnodes_detailed_set = mixnodes_detailed + .iter() + .map(|elem| elem.identity_key().to_owned()) + .collect::>(); + + let mixnodes_legacy = nym_nodes + .iter() + .filter(|node| { + mixnodes_detailed_set.contains(&node.ed25519_identity_pubkey.to_base58_string()) + }) + .collect::>(); + let mixnodes_described = api_client .nym_api .get_mixnodes_described() .await .log_error("get_mixnodes_described")?; - let mixnodes_active = api_client + + tracing::info!("🟣 mixnodes_described: {}", mixnodes_described.len()); + let mixing_assigned_nodes = api_client .nym_api .get_basic_active_mixing_assigned_nodes(false, None, None) .await - .log_error("get_active_mixnodes")? + .log_error("get_basic_active_mixing_assigned_nodes")? .nodes .data; + let delegation_program_members = get_delegation_program_details(&self.network_details, &self.nyxd_addr).await?; // keep stats for later - let count_bonded_mixnodes = mixnodes.len(); + let assigned_entry_count = nym_nodes + .iter() + .filter(|elem| matches!(elem.role, NodeRole::EntryGateway)) + .count(); + let assigned_exit_count = nym_nodes + .iter() + .filter(|elem| matches!(elem.role, NodeRole::ExitGateway)) + .count(); let count_bonded_gateways = gateways.len(); - let count_bonded_mixnodes_active = mixnodes_active.len(); + let assigned_mixing_count = mixing_assigned_nodes.len(); + let count_legacy_mixnodes = mixnodes_legacy.len(); - let gateway_records = self.prepare_gateway_data( - &gateways, - &gateways_blacklisted, - gateway_geodata, - all_skimmed_nodes, - )?; + let gateway_records = self.prepare_gateway_data(&gateways, gateway_geodata, &nym_nodes)?; let pool = self.db_pool.clone(); queries::insert_gateways(&pool, gateway_records) @@ -208,27 +204,8 @@ impl Monitor { tracing::debug!("Gateway info written to DB!"); })?; - let count_gateways_blacklisted = gateways - .iter() - .filter(|gw| { - let gw_identity = gw.ed25519_identity_key().to_base58_string(); - gateways_blacklisted.contains(&gw_identity) - }) - .count(); - - if count_gateways_blacklisted > 0 { - queries::write_blacklisted_gateways_to_db(&pool, gateways_blacklisted.iter()) - .await - .map(|_| { - tracing::debug!( - "Gateway blacklist info written to DB! {} blacklisted by Nym API", - count_gateways_blacklisted - ) - })?; - } - let mixnode_records = self.prepare_mixnode_data( - &legacy_mixnodes, + &mixnodes_detailed, mixnodes_described, delegation_program_members, )?; @@ -238,20 +215,6 @@ impl Monitor { tracing::debug!("Mixnode info written to DB!"); })?; - let count_mixnodes_blacklisted = legacy_mixnodes - .iter() - .filter(|elem| elem.blacklisted) - .count(); - - let recently_unbonded_gateways = - queries::ensure_gateways_still_bonded(&pool, &gateways).await?; - let recently_unbonded_mixnodes = - queries::ensure_mixnodes_still_bonded(&pool, &legacy_mixnodes).await?; - - let count_bonded_mixnodes_reserve = 0; // TODO: NymAPI doesn't report the reserve set size - let count_bonded_mixnodes_inactive = - count_bonded_mixnodes.saturating_sub(count_bonded_mixnodes_active); - let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(&pool).await?; // @@ -259,30 +222,28 @@ impl Monitor { // let nodes_summary = vec![ - (MIXNODES_BONDED_COUNT, &count_bonded_mixnodes), - (MIXNODES_BONDED_ACTIVE, &count_bonded_mixnodes_active), - (MIXNODES_BONDED_INACTIVE, &count_bonded_mixnodes_inactive), - (MIXNODES_BONDED_RESERVE, &count_bonded_mixnodes_reserve), - (MIXNODES_BLACKLISTED_COUNT, &count_mixnodes_blacklisted), - (GATEWAYS_BONDED_COUNT, &count_bonded_gateways), - (MIXNODES_HISTORICAL_COUNT, &all_historical_mixnodes), - (GATEWAYS_HISTORICAL_COUNT, &all_historical_gateways), - (GATEWAYS_BLACKLISTED_COUNT, &count_gateways_blacklisted), + (NYMNODE_COUNT, nym_nodes.len()), + (ASSIGNED_MIXING_COUNT, assigned_mixing_count), + (MIXNODES_LEGACY_COUNT, count_legacy_mixnodes), + (NYMNODES_DESCRIBED_COUNT, described_nodes.len()), + (GATEWAYS_BONDED_COUNT, count_bonded_gateways), + (ASSIGNED_ENTRY_COUNT, assigned_entry_count), + (ASSIGNED_EXIT_COUNT, assigned_exit_count), + // TODO dz doesn't make sense, could make sense with historical Nym + // Nodes if we really need this data + (MIXNODES_HISTORICAL_COUNT, all_historical_mixnodes), + (GATEWAYS_HISTORICAL_COUNT, all_historical_gateways), ]; let last_updated = chrono::offset::Utc::now(); let last_updated_utc = last_updated.timestamp().to_string(); let network_summary = NetworkSummary { + total_nodes: nym_nodes.len().cast_checked()?, mixnodes: mixnode::MixnodeSummary { - bonded: mixnode::MixnodeSummaryBonded { - count: count_bonded_mixnodes.cast_checked()?, - active: count_bonded_mixnodes_active.cast_checked()?, - inactive: count_bonded_mixnodes_inactive.cast_checked()?, - reserve: count_bonded_mixnodes_reserve.cast_checked()?, - last_updated_utc: last_updated_utc.to_owned(), - }, - blacklisted: mixnode::MixnodeSummaryBlacklisted { - count: count_mixnodes_blacklisted.cast_checked()?, + bonded: mixnode::MixingNodesSummary { + count: assigned_mixing_count.cast_checked()?, + self_described: described_nodes.len().cast_checked()?, + legacy: count_legacy_mixnodes.cast_checked()?, last_updated_utc: last_updated_utc.to_owned(), }, historical: mixnode::MixnodeSummaryHistorical { @@ -293,10 +254,8 @@ impl Monitor { gateways: gateway::GatewaySummary { bonded: gateway::GatewaySummaryBonded { count: count_bonded_gateways.cast_checked()?, - last_updated_utc: last_updated_utc.to_owned(), - }, - blacklisted: gateway::GatewaySummaryBlacklisted { - count: count_gateways_blacklisted.cast_checked()?, + entry: assigned_entry_count.cast_checked()?, + exit: assigned_exit_count.cast_checked()?, last_updated_utc: last_updated_utc.to_owned(), }, historical: gateway::GatewaySummaryHistorical { @@ -312,14 +271,6 @@ impl Monitor { for (key, value) in nodes_summary.iter() { log_lines.push(format!("{} = {}", key, value)); } - log_lines.push(format!( - "recently_unbonded_mixnodes = {}", - recently_unbonded_mixnodes - )); - log_lines.push(format!( - "recently_unbonded_gateways = {}", - recently_unbonded_gateways - )); tracing::info!("Directory summary: \n{}", log_lines.join("\n")); @@ -349,9 +300,8 @@ impl Monitor { fn prepare_gateway_data( &self, gateways: &[&NymNodeDescription], - gateways_blacklisted: &HashSet, gateway_geodata: Vec, - skimmed_gateways: Vec, + skimmed_gateways: &[SkimmedNode], ) -> anyhow::Result> { let mut gateway_records = Vec::new(); @@ -359,7 +309,6 @@ impl Monitor { let identity_key = gateway.ed25519_identity_key().to_base58_string(); let bonded = true; let last_updated_utc = chrono::offset::Utc::now().timestamp(); - let blacklisted = gateways_blacklisted.contains(&identity_key); let self_described = serde_json::to_string(&gateway.description)?; @@ -383,7 +332,6 @@ impl Monitor { gateway_records.push(GatewayRecord { identity_key: identity_key.to_owned(), bonded, - blacklisted, self_described, explorer_pretty_bond, last_updated_utc, @@ -407,7 +355,6 @@ impl Monitor { let identity_key = mixnode.identity_key(); let bonded = true; let total_stake = decimal_to_i64(mixnode.mixnode_details.total_stake()); - let blacklisted = mixnode.blacklisted; let node_info = mixnode.mix_node(); let host = node_info.host.clone(); let http_port = node_info.http_api_port; @@ -427,7 +374,6 @@ impl Monitor { total_stake, host, http_port, - blacklisted, full_details, self_described, last_updated_utc, @@ -519,7 +465,7 @@ async fn get_delegation_program_details( Ok(mix_ids) } -fn decimal_to_i64(decimal: Decimal) -> i64 { +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; 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 8d0664316f..76f68a6549 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 @@ -123,6 +123,7 @@ impl MetricsScrapingData { } } + #[instrument(level = "debug", name = "metrics_scraper", skip_all)] async fn try_scrape_metrics(&self) -> Option { match self.try_get_client().await { Ok(client) => { @@ -136,13 +137,13 @@ impl MetricsScrapingData { } } Err(e) => { - tracing::error!("[metrics scraper]: {e}"); + tracing::warn!("{e}"); None } } } Err(e) => { - tracing::error!("[metrics scraper]: {e}"); + tracing::warn!("{e}"); None } } 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 7ce0a851ef..db1f72e3a7 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 @@ -109,7 +109,6 @@ pub(crate) async fn try_queue_testrun( }) } -// TODO dz do we need these? pub fn now_utc() -> DateTime { SystemTime::now().into() }