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

Add /v3/nym-nodes

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