From 2f051fd943664392ebab0cd0b0eb740901398e59 Mon Sep 17 00:00:00 2001 From: Dinko Zdravac <173912580+dynco-nym@users.noreply.github.com> Date: Fri, 13 Sep 2024 17:14:07 +0200 Subject: [PATCH] Node Status API background task (#4854) * Setup new package * Setup DB * Fetch & store mixnodes/GWs - refactor db package structure - finally solve DATABASE_URL: absolute path works best * Additional query functionality - missing only daily summary, which requires type refactoring * Replace type alias tuples with structs * Insert summary * Add github job to build package * Build script for sqlx * Remove data dir - useless now that sqlx DB sits in OUT_DIR * PR feedback --- Cargo.lock | 23 ++ Cargo.toml | 2 + nym-node-status-api/.gitignore | 1 + nym-node-status-api/Cargo.toml | 37 ++ nym-node-status-api/build.rs | 33 ++ nym-node-status-api/launch_node_status_api.sh | 64 +++ nym-node-status-api/migrations/000_init.sql | 59 +++ nym-node-status-api/src/cli/mod.rs | 17 + nym-node-status-api/src/db/mod.rs | 40 ++ nym-node-status-api/src/db/models.rs | 128 ++++++ .../src/db/queries/gateways.rs | 116 ++++++ nym-node-status-api/src/db/queries/misc.rs | 86 ++++ .../src/db/queries/mixnodes.rs | 101 +++++ nym-node-status-api/src/db/queries/mod.rs | 9 + nym-node-status-api/src/logging.rs | 41 ++ nym-node-status-api/src/main.rs | 38 ++ nym-node-status-api/src/monitor/mod.rs | 374 ++++++++++++++++++ 17 files changed, 1169 insertions(+) create mode 100644 nym-node-status-api/.gitignore create mode 100644 nym-node-status-api/Cargo.toml create mode 100644 nym-node-status-api/build.rs create mode 100755 nym-node-status-api/launch_node_status_api.sh create mode 100644 nym-node-status-api/migrations/000_init.sql create mode 100644 nym-node-status-api/src/cli/mod.rs create mode 100644 nym-node-status-api/src/db/mod.rs create mode 100644 nym-node-status-api/src/db/models.rs create mode 100644 nym-node-status-api/src/db/queries/gateways.rs create mode 100644 nym-node-status-api/src/db/queries/misc.rs create mode 100644 nym-node-status-api/src/db/queries/mixnodes.rs create mode 100644 nym-node-status-api/src/db/queries/mod.rs create mode 100644 nym-node-status-api/src/logging.rs create mode 100644 nym-node-status-api/src/main.rs create mode 100644 nym-node-status-api/src/monitor/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 0a402bea90..a1b8a47c19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5878,6 +5878,29 @@ dependencies = [ "utoipa", ] +[[package]] +name = "nym-node-status-api" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum 0.7.5", + "chrono", + "clap 4.5.17", + "cosmwasm-std", + "futures-util", + "nym-bin-common", + "nym-explorer-client", + "nym-network-defaults", + "nym-validator-client", + "serde", + "serde_json", + "sqlx", + "thiserror", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "nym-node-tester-utils" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 6303eccca0..8f2bc57f13 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -120,6 +120,7 @@ members = [ "nym-node", "nym-node/nym-node-http-api", "nym-node/nym-node-requests", + "nym-node-status-api", "nym-outfox", "nym-validator-rewarder", "tools/echo-server", @@ -238,6 +239,7 @@ eyre = "0.6.9" fastrand = "2.1.1" flate2 = "1.0.34" futures = "0.3.28" +futures-util = "0.3" generic-array = "0.14.7" getrandom = "0.2.10" getset = "0.1.3" diff --git a/nym-node-status-api/.gitignore b/nym-node-status-api/.gitignore new file mode 100644 index 0000000000..8fce603003 --- /dev/null +++ b/nym-node-status-api/.gitignore @@ -0,0 +1 @@ +data/ diff --git a/nym-node-status-api/Cargo.toml b/nym-node-status-api/Cargo.toml new file mode 100644 index 0000000000..f670afc8d8 --- /dev/null +++ b/nym-node-status-api/Cargo.toml @@ -0,0 +1,37 @@ +# Copyright 2024 - Nym Technologies SA +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "nym-node-status-api" +version = "0.1.0" +authors.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] +anyhow = { workspace = true } +axum = { workspace = true } +chrono = { workspace = true } +clap = { workspace = true, features = ["cargo", "derive"] } +cosmwasm-std = { workspace = true } +futures-util = { workspace = true } +nym-bin-common = { path = "../common/bin-common" } +nym-explorer-client = { path = "../explorer-api/explorer-client" } +nym-network-defaults = { path = "../common/network-defaults" } +nym-validator-client = { path = "../common/client-libs/validator-client" } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-subscriber = { workspace = true, features = ["env-filter"] } + +[build-dependencies] +anyhow = { workspace = true } +tokio = { workspace = true, features = ["macros" ] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] } diff --git a/nym-node-status-api/build.rs b/nym-node-status-api/build.rs new file mode 100644 index 0000000000..74c6767c6f --- /dev/null +++ b/nym-node-status-api/build.rs @@ -0,0 +1,33 @@ +use anyhow::{anyhow, Result}; +use sqlx::{Connection, SqliteConnection}; + +const SQLITE_DB_FILENAME: &str = "nym-node-status-api.sqlite"; + +#[tokio::main] +async fn main() -> Result<()> { + let out_dir = read_env_var("OUT_DIR")?; + let database_path = format!("sqlite://{}/{}?mode=rwc", out_dir, SQLITE_DB_FILENAME); + + let mut conn = SqliteConnection::connect(&database_path).await?; + sqlx::migrate!("./migrations").run(&mut conn).await?; + + #[cfg(target_family = "unix")] + println!("cargo::rustc-env=DATABASE_URL=sqlite://{}", &database_path); + + #[cfg(target_family = "windows")] + // for some strange reason we need to add a leading `/` to the windows path even though it's + // not a valid windows path... but hey, it works... + println!("cargo::rustc-env=DATABASE_URL=sqlite:///{}", &database_path); + + rerun_if_changed(); + Ok(()) +} + +fn read_env_var(var: &str) -> Result { + std::env::var(var).map_err(|_| anyhow!("You need to set {} env var", var)) +} + +fn rerun_if_changed() { + println!("cargo::rerun-if-changed=migrations"); + println!("cargo::rerun-if-changed=src/db/queries"); +} diff --git a/nym-node-status-api/launch_node_status_api.sh b/nym-node-status-api/launch_node_status_api.sh new file mode 100755 index 0000000000..8af370232f --- /dev/null +++ b/nym-node-status-api/launch_node_status_api.sh @@ -0,0 +1,64 @@ +#!/bin/bash + +set -e + +function usage() { + echo "Usage: $0 [-ci]" + echo " -c Clear DB and re-initialize it before launching the binary." + echo " -i Only initialize and prepare database, env vars then exit without" + echo " launching" + exit 0 +} + +function init_db() { + rm -rf data/* + # https://github.com/launchbadge/sqlx/blob/main/sqlx-cli/README.md + cargo sqlx database drop -y + + cargo sqlx database create + cargo sqlx migrate run + cargo sqlx prepare + + echo "Fresh database ready!" +} + +# export DATABASE_URL as absolute path due to this +# https://github.com/launchbadge/sqlx/issues/3099 +db_filename="nym-node-status-api.sqlite" +script_abs_path=$(realpath "$0") +package_dir=$(dirname "$script_abs_path") +db_abs_path="$package_dir/data/$db_filename" +dotenv_file="$package_dir/.env" +echo "DATABASE_URL=sqlite://$db_abs_path" > "$dotenv_file" + +export RUST_LOG=${RUST_LOG:-debug} + +# export DATABASE_URL from .env file +set -a && source "$dotenv_file" && set +a + +clear_db=false +init_only=false + +while getopts "ci" opt; do + case ${opt} in + c) + clear_db=true + ;; + i) + init_only=true + ;; + \?) + usage + ;; + esac +done + +if [ "$clear_db" = true ] || [ "$init_only" = true ]; then + init_db +fi + +if [ "$init_only" = true ]; then + exit 0 +fi + +cargo run --package nym-node-status-api diff --git a/nym-node-status-api/migrations/000_init.sql b/nym-node-status-api/migrations/000_init.sql new file mode 100644 index 0000000000..f950b49229 --- /dev/null +++ b/nym-node-status-api/migrations/000_init.sql @@ -0,0 +1,59 @@ +CREATE TABLE gateways +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + gateway_identity_key VARCHAR NOT NULL UNIQUE, + self_described VARCHAR, + explorer_pretty_bond VARCHAR, + last_probe_result VARCHAR, + last_probe_log VARCHAR, + config_score INTEGER NOT NULL DEFAULT (0), + config_score_successes REAL NOT NULL DEFAULT (0), + config_score_samples REAL NOT NULL DEFAULT (0), + routing_score INTEGER NOT NULL DEFAULT (0), + routing_score_successes REAL NOT NULL DEFAULT (0), + routing_score_samples REAL NOT NULL DEFAULT (0), + test_run_samples REAL NOT NULL DEFAULT (0), + last_testrun_utc INTEGER, + last_updated_utc INTEGER NOT NULL, + bonded INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0, + blacklisted INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0, + performance INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX idx_gateway_description_gateway_identity_key ON gateways (gateway_identity_key); + + +CREATE TABLE mixnodes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + identity_key VARCHAR NOT NULL UNIQUE, + mix_id INTEGER NOT NULL UNIQUE, + bonded INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0, + total_stake INTEGER NOT NULL, + host VARCHAR NOT NULL, + http_api_port INTEGER NOT NULL, + blacklisted INTEGER CHECK (blacklisted in (0, 1)) NOT NULL DEFAULT 0, + full_details VARCHAR, + self_described VARCHAR, + last_updated_utc INTEGER NOT NULL + , is_dp_delegatee INTEGER CHECK (is_dp_delegatee IN (0, 1)) NOT NULL DEFAULT 0); +CREATE INDEX idx_mixnodes_mix_id ON mixnodes (mix_id); +CREATE INDEX idx_mixnodes_identity_key ON mixnodes (identity_key); + + +CREATE TABLE summary +( + key VARCHAR PRIMARY KEY, + value_json VARCHAR, + last_updated_utc INTEGER NOT NULL +); + + +CREATE TABLE summary_history +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + date VARCHAR UNIQUE NOT NULL, + timestamp_utc INTEGER NOT NULL, + value_json VARCHAR +); +CREATE INDEX idx_summary_history_timestamp_utc ON summary_history (timestamp_utc); +CREATE INDEX idx_summary_history_date ON summary_history (date); diff --git a/nym-node-status-api/src/cli/mod.rs b/nym-node-status-api/src/cli/mod.rs new file mode 100644 index 0000000000..7f46578863 --- /dev/null +++ b/nym-node-status-api/src/cli/mod.rs @@ -0,0 +1,17 @@ +use clap::Parser; +use nym_bin_common::bin_info; +use std::sync::OnceLock; + +// Helper for passing LONG_VERSION to clap +fn pretty_build_info_static() -> &'static str { + static PRETTY_BUILD_INFORMATION: OnceLock = OnceLock::new(); + PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print()) +} + +#[derive(Parser, Debug)] +#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)] +pub(crate) struct Cli { + /// Path pointing to an env file that configures the Nym API. + #[clap(short, long)] + pub(crate) config_env_file: Option, +} diff --git a/nym-node-status-api/src/db/mod.rs b/nym-node-status-api/src/db/mod.rs new file mode 100644 index 0000000000..d9850abf7d --- /dev/null +++ b/nym-node-status-api/src/db/mod.rs @@ -0,0 +1,40 @@ +use std::str::FromStr; + +use crate::read_env_var; +use anyhow::{anyhow, Result}; +use sqlx::{migrate::Migrator, sqlite::SqliteConnectOptions, ConnectOptions, SqlitePool}; +pub(crate) const DATABASE_URL_ENV_VAR: &str = "DATABASE_URL"; +static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); + +pub(crate) mod models; +pub(crate) mod queries; + +pub(crate) type DbPool = SqlitePool; + +pub(crate) struct Storage { + pool: DbPool, +} + +impl Storage { + pub async fn init() -> Result { + let connection_url = read_env_var(DATABASE_URL_ENV_VAR)?; + let connect_options = { + let connect_options = SqliteConnectOptions::from_str(&connection_url)?; + let mut connect_options = connect_options.create_if_missing(true); + let connect_options = connect_options.disable_statement_logging(); + (*connect_options).clone() + }; + + let pool = sqlx::SqlitePool::connect_with(connect_options) + .await + .map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?; + + MIGRATOR.run(&pool).await?; + + Ok(Storage { pool }) + } + + pub async fn pool(&self) -> &DbPool { + &self.pool + } +} diff --git a/nym-node-status-api/src/db/models.rs b/nym-node-status-api/src/db/models.rs new file mode 100644 index 0000000000..83e04d99ee --- /dev/null +++ b/nym-node-status-api/src/db/models.rs @@ -0,0 +1,128 @@ +use serde::{Deserialize, Serialize}; + +pub(crate) struct GatewayRecord { + pub(crate) identity_key: String, + pub(crate) bonded: bool, + pub(crate) blacklisted: bool, + pub(crate) self_described: Option, + pub(crate) explorer_pretty_bond: Option, + pub(crate) last_updated_utc: i64, + pub(crate) performance: u8, +} + +pub(crate) struct MixnodeRecord { + pub(crate) mix_id: u32, + pub(crate) identity_key: String, + pub(crate) bonded: bool, + 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, + pub(crate) is_dp_delegatee: bool, +} + +#[allow(unused)] +#[derive(Debug, Clone)] +pub(crate) struct BondedStatusDto { + pub(crate) id: i64, + pub(crate) identity_key: String, + pub(crate) bonded: bool, +} + +#[allow(unused)] +#[derive(Debug, Clone, Default)] +pub(crate) struct SummaryDto { + pub(crate) key: String, + pub(crate) value_json: String, + pub(crate) last_updated_utc: i64, +} + +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 GATEWAYS_BONDED_COUNT: &str = "gateways.bonded.count"; +pub(crate) const GATEWAYS_EXPLORER_COUNT: &str = "gateways.explorer.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"; + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub(crate) struct NetworkSummary { + pub(crate) mixnodes: mixnode::MixnodeSummary, + pub(crate) gateways: gateway::GatewaySummary, +} + +pub(crate) mod mixnode { + use super::*; + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct MixnodeSummary { + pub(crate) bonded: MixnodeSummaryBonded, + pub(crate) blacklisted: MixnodeSummaryBlacklisted, + pub(crate) historical: MixnodeSummaryHistorical, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + 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)] + pub(crate) struct MixnodeSummaryBlacklisted { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct MixnodeSummaryHistorical { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } +} + +pub(crate) mod gateway { + use super::*; + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct GatewaySummary { + pub(crate) bonded: GatewaySummaryBonded, + pub(crate) blacklisted: GatewaySummaryBlacklisted, + pub(crate) historical: GatewaySummaryHistorical, + pub(crate) explorer: GatewaySummaryExplorer, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct GatewaySummaryExplorer { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct GatewaySummaryBonded { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct GatewaySummaryHistorical { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } + + #[derive(Debug, Clone, Deserialize, Serialize)] + pub(crate) struct GatewaySummaryBlacklisted { + pub(crate) count: i32, + pub(crate) last_updated_utc: String, + } +} diff --git a/nym-node-status-api/src/db/queries/gateways.rs b/nym-node-status-api/src/db/queries/gateways.rs new file mode 100644 index 0000000000..69f77a97fd --- /dev/null +++ b/nym-node-status-api/src/db/queries/gateways.rs @@ -0,0 +1,116 @@ +use crate::db::{ + models::{BondedStatusDto, GatewayRecord}, + DbPool, +}; +use futures_util::TryStreamExt; +use nym_validator_client::models::DescribedGateway; + +pub(crate) async fn insert_gateways( + pool: &DbPool, + gateways: Vec, +) -> anyhow::Result<()> { + let mut db = pool.acquire().await?; + for record in gateways { + sqlx::query!( + "INSERT INTO gateways + (gateway_identity_key, bonded, blacklisted, + self_described, explorer_pretty_bond, + last_updated_utc, performance) + 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, + record.performance + ) + .execute(&mut *db) + .await?; + } + + 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: &[DescribedGateway], +) -> 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.bond.identity() == 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) +} diff --git a/nym-node-status-api/src/db/queries/misc.rs b/nym-node-status-api/src/db/queries/misc.rs new file mode 100644 index 0000000000..c2c7b6d9eb --- /dev/null +++ b/nym-node-status-api/src/db/queries/misc.rs @@ -0,0 +1,86 @@ +use crate::db::{models::NetworkSummary, DbPool}; +use chrono::{DateTime, Utc}; + +/// take `last_updated` instead of calculating it so that `summary` matches +/// `daily_summary` +pub(crate) async fn insert_summaries( + pool: &DbPool, + summaries: &[(&str, &usize)], + summary: &NetworkSummary, + last_updated: DateTime, +) -> anyhow::Result<()> { + insert_summary(pool, summaries, last_updated).await?; + + insert_summary_history(pool, summary, last_updated).await?; + + Ok(()) +} + +async fn insert_summary( + pool: &DbPool, + summaries: &[(&str, &usize)], + last_updated: DateTime, +) -> anyhow::Result<()> { + let timestamp = last_updated.timestamp(); + let mut tx = pool.begin().await?; + + for (kind, value) in summaries { + let value = value.to_string(); + sqlx::query!( + "INSERT INTO summary + (key, value_json, last_updated_utc) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + value_json=excluded.value_json, + last_updated_utc=excluded.last_updated_utc;", + kind, + value, + timestamp + ) + .execute(&mut tx) + .await + .map_err(|err| { + tracing::error!("Failed to insert data for {kind}: {err}, aborting transaction",); + err + })?; + } + + Ok(()) +} + +/// For ``, `summary_history` is updated with fresh data on every +/// iteration. +/// +/// After UTC midnight, summary is inserted for `` and last entry for +/// `` stays there forever. +/// +/// This is not aggregate data, it's a set of latest data points +async fn insert_summary_history( + pool: &DbPool, + summary: &NetworkSummary, + last_updated: DateTime, +) -> anyhow::Result<()> { + let mut conn = pool.acquire().await?; + + let value_json = serde_json::to_string(&summary)?; + let timestamp = last_updated.timestamp(); + let now_rfc3339 = last_updated.to_rfc3339(); + // YYYY-MM-DD, without time + let date = &now_rfc3339[..10]; + + sqlx::query!( + "INSERT INTO summary_history + (date, timestamp_utc, value_json) + VALUES (?, ?, ?) + ON CONFLICT(date) DO UPDATE SET + timestamp_utc=excluded.timestamp_utc, + value_json=excluded.value_json;", + date, + timestamp, + value_json + ) + .execute(&mut *conn) + .await?; + + Ok(()) +} diff --git a/nym-node-status-api/src/db/queries/mixnodes.rs b/nym-node-status-api/src/db/queries/mixnodes.rs new file mode 100644 index 0000000000..aed48ed4d7 --- /dev/null +++ b/nym-node-status-api/src/db/queries/mixnodes.rs @@ -0,0 +1,101 @@ +use futures_util::TryStreamExt; +use nym_validator_client::models::MixNodeBondAnnotated; + +use crate::db::{ + models::{BondedStatusDto, MixnodeRecord}, + DbPool, +}; + +pub(crate) async fn insert_mixnodes( + pool: &DbPool, + mixnodes: Vec, +) -> anyhow::Result<()> { + let mut conn = pool.acquire().await?; + + for record in mixnodes.iter() { + // https://www.sqlite.org/lang_upsert.html + sqlx::query!( + "INSERT INTO mixnodes + (mix_id, identity_key, bonded, total_stake, + host, http_api_port, blacklisted, full_details, + self_described, last_updated_utc, is_dp_delegatee) + 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, + full_details=excluded.full_details,self_described=excluded.self_described, + last_updated_utc=excluded.last_updated_utc, + is_dp_delegatee = excluded.is_dp_delegatee;", + record.mix_id, + record.identity_key, + record.bonded, + record.total_stake, + record.host, + record.http_port, + record.blacklisted, + record.full_details, + record.self_described, + record.last_updated_utc, + record.is_dp_delegatee + ) + .execute(&mut *conn) + .await?; + } + + Ok(()) +} + +/// Ensure all mixnodes that are set as bonded, are still bonded +pub(crate) async fn ensure_mixnodes_still_bonded( + pool: &DbPool, + mixnodes: &[MixNodeBondAnnotated], +) -> 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/src/db/queries/mod.rs b/nym-node-status-api/src/db/queries/mod.rs new file mode 100644 index 0000000000..38f1faab61 --- /dev/null +++ b/nym-node-status-api/src/db/queries/mod.rs @@ -0,0 +1,9 @@ +mod gateways; +mod misc; +mod mixnodes; + +pub(crate) use gateways::{ + ensure_gateways_still_bonded, insert_gateways, write_blacklisted_gateways_to_db, +}; +pub(crate) use misc::insert_summaries; +pub(crate) use mixnodes::{ensure_mixnodes_still_bonded, insert_mixnodes}; diff --git a/nym-node-status-api/src/logging.rs b/nym-node-status-api/src/logging.rs new file mode 100644 index 0000000000..d61cd78ee1 --- /dev/null +++ b/nym-node-status-api/src/logging.rs @@ -0,0 +1,41 @@ +use tracing::level_filters::LevelFilter; +use tracing_subscriber::{filter::Directive, EnvFilter}; + +pub(crate) fn setup_tracing_logger() { + fn directive_checked(directive: String) -> Directive { + directive.parse().expect("Failed to parse log directive") + } + + let log_builder = tracing_subscriber::fmt() + // Use a more compact, abbreviated log format + .compact() + // Display source code file paths + .with_file(true) + // Display source code line numbers + .with_line_number(true) + // Don't display the event's target (module path) + .with_target(false); + + let mut filter = EnvFilter::builder() + // if RUST_LOG isn't set, set default level + .with_default_directive(LevelFilter::INFO.into()) + .from_env_lossy(); + // these crates are more granularly filtered + let filter_crates = [ + "nym_bin_common", + "nym_explorer_client", + "nym_network_defaults", + "nym_validator_client", + "reqwest", + "rustls", + "hyper", + "sqlx", + "h2", + "tendermint_rpc", + ]; + for crate_name in filter_crates { + filter = filter.add_directive(directive_checked(format!("{}=warn", crate_name))); + } + + log_builder.with_env_filter(filter).init(); +} diff --git a/nym-node-status-api/src/main.rs b/nym-node-status-api/src/main.rs new file mode 100644 index 0000000000..6296406701 --- /dev/null +++ b/nym-node-status-api/src/main.rs @@ -0,0 +1,38 @@ +use anyhow::anyhow; +use clap::Parser; +use nym_network_defaults::setup_env; + +mod cli; +mod db; +mod logging; +mod monitor; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + logging::setup_tracing_logger(); + + let args = cli::Cli::parse(); + // if dotenv file is present, load its values + // otherwise, default to mainnet + setup_env(args.config_env_file.as_ref()); + tracing::debug!("{:?}", std::env::var("NETWORK_NAME")); + tracing::debug!("{:?}", std::env::var("EXPLORER_API")); + tracing::debug!("{:?}", std::env::var("NYM_API")); + + let storage = db::Storage::init().await?; + monitor::spawn_in_background(storage) + .await + .expect("Monitor task failed"); + tracing::info!("Started server"); + + Ok(()) +} + +pub(crate) fn read_env_var(env_var: &str) -> anyhow::Result { + std::env::var(env_var) + .map(|value| { + tracing::trace!("{}={}", env_var, value); + value + }) + .map_err(|_| anyhow!("You need to set {}", env_var)) +} diff --git a/nym-node-status-api/src/monitor/mod.rs b/nym-node-status-api/src/monitor/mod.rs new file mode 100644 index 0000000000..810320bdd0 --- /dev/null +++ b/nym-node-status-api/src/monitor/mod.rs @@ -0,0 +1,374 @@ +use crate::db::models::{ + gateway, mixnode, GatewayRecord, MixnodeRecord, NetworkSummary, GATEWAYS_BLACKLISTED_COUNT, + GATEWAYS_BONDED_COUNT, GATEWAYS_EXPLORER_COUNT, GATEWAYS_HISTORICAL_COUNT, + MIXNODES_BLACKLISTED_COUNT, MIXNODES_BONDED_ACTIVE, MIXNODES_BONDED_COUNT, + MIXNODES_BONDED_INACTIVE, MIXNODES_BONDED_RESERVE, MIXNODES_HISTORICAL_COUNT, +}; +use crate::db::{queries, DbPool, Storage}; +use anyhow::anyhow; +use cosmwasm_std::Decimal; +use nym_explorer_client::{ExplorerClient, PrettyDetailedGatewayBond}; +use nym_network_defaults::NymNetworkDetails; +use nym_validator_client::client::NymApiClientExt; +use nym_validator_client::models::{DescribedGateway, DescribedMixNode, MixNodeBondAnnotated}; +use nym_validator_client::nym_nodes::SkimmedNode; +use nym_validator_client::nyxd::contract_traits::PagedMixnetQueryClient; +use nym_validator_client::nyxd::{AccountId, NyxdClient}; +use nym_validator_client::NymApiClient; +use std::collections::HashSet; +use std::str::FromStr; +use tokio::task::JoinHandle; +use tokio::time::Duration; + +const REFRESH_DELAY: Duration = Duration::from_secs(60 * 5); +const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); +static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw"; + +// TODO dz: query many NYM APIs: +// multiple instances running directory cache, ask sachin +pub(crate) fn spawn_in_background(storage: Storage) -> JoinHandle<()> { + tokio::spawn(async move { + let db_pool = storage.pool().await; + let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env(); + + loop { + tracing::info!("Refreshing node info..."); + + if let Err(e) = run(db_pool, &network_defaults).await { + tracing::error!( + "Monitor run failed: {e}, retrying in {}s...", + FAILURE_RETRY_DELAY.as_secs() + ); + tokio::time::sleep(FAILURE_RETRY_DELAY).await; + } else { + tracing::info!( + "Info successfully collected, sleeping for {}s...", + REFRESH_DELAY.as_secs() + ); + tokio::time::sleep(REFRESH_DELAY).await; + } + } + }) +} + +async fn run(pool: &DbPool, network_details: &NymNetworkDetails) -> anyhow::Result<()> { + let default_api_url = network_details + .endpoints + .first() + .expect("rust sdk mainnet default incorrectly configured") + .api_url() + .clone() + .expect("rust sdk mainnet default missing api_url"); + let default_explorer_url = network_details.explorer_api.clone().map(|url| { + url.parse() + .expect("rust sdk mainnet default explorer url not parseable") + }); + + let default_explorer_url = + default_explorer_url.expect("explorer url missing in network config"); + let explorer_client = ExplorerClient::new(default_explorer_url)?; + let explorer_gateways = explorer_client.get_gateways().await?; + + let api_client = NymApiClient::new(default_api_url); + let gateways = api_client.get_cached_described_gateways().await?; + tracing::debug!("Fetched {} gateways", gateways.len()); + let skimmed_gateways = api_client.get_basic_gateways(None).await?; + + let mixnodes = api_client.get_cached_mixnodes().await?; + tracing::debug!("Fetched {} mixnodes", mixnodes.len()); + + // TODO dz can we calculate blacklisted GWs from their performance? + // where do we get their performance? + let gateways_blacklisted = api_client + .nym_api + .get_gateways_blacklisted() + .await + .map(|vec| vec.into_iter().collect::>())?; + + // Cached mixnodes don't include blacklisted nodes + // We need that to calculate the total locked tokens later + let mixnodes = api_client + .nym_api + .get_mixnodes_detailed_unfiltered() + .await?; + let mixnodes_described = api_client.nym_api.get_mixnodes_described().await?; + let mixnodes_active = api_client.nym_api.get_active_mixnodes().await?; + let delegation_program_members = get_delegation_program_details(network_details).await?; + + // keep stats for later + let count_bonded_mixnodes = mixnodes.len(); + let count_bonded_gateways = gateways.len(); + let count_explorer_gateways = explorer_gateways.len(); + let count_bonded_mixnodes_active = mixnodes_active.len(); + + let gateway_records = prepare_gateway_data( + &gateways, + &gateways_blacklisted, + explorer_gateways, + skimmed_gateways, + )?; + queries::insert_gateways(pool, gateway_records) + .await + .map(|_| { + tracing::debug!("Gateway info written to DB!"); + })?; + + // instead of counting blacklisted GWs returned from API cache, count from the active set + let count_gateways_blacklisted = gateways + .iter() + .filter(|gw| { + let gw_identity = gw.bond.identity(); + 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 = + prepare_mixnode_data(&mixnodes, mixnodes_described, delegation_program_members)?; + queries::insert_mixnodes(pool, mixnode_records) + .await + .map(|_| { + tracing::debug!("Mixnode info written to DB!"); + })?; + + let count_mixnodes_blacklisted = 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, &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 - count_bonded_mixnodes_active; + + let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(pool).await?; + + // + // write summary keys and values to table + // + + 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), + (GATEWAYS_EXPLORER_COUNT, &count_explorer_gateways), + (MIXNODES_HISTORICAL_COUNT, &all_historical_mixnodes), + (GATEWAYS_HISTORICAL_COUNT, &all_historical_gateways), + (GATEWAYS_BLACKLISTED_COUNT, &count_gateways_blacklisted), + ]; + + // TODO dz do we need signed int in type definition? maybe because of API? + let last_updated = chrono::offset::Utc::now(); + let last_updated_utc = last_updated.timestamp().to_string(); + let network_summary = NetworkSummary { + mixnodes: mixnode::MixnodeSummary { + bonded: mixnode::MixnodeSummaryBonded { + count: count_bonded_mixnodes as i32, + active: count_bonded_mixnodes_active as i32, + inactive: count_bonded_mixnodes_inactive as i32, + reserve: count_bonded_mixnodes_reserve as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + blacklisted: mixnode::MixnodeSummaryBlacklisted { + count: count_mixnodes_blacklisted as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + historical: mixnode::MixnodeSummaryHistorical { + count: all_historical_mixnodes as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + }, + gateways: gateway::GatewaySummary { + bonded: gateway::GatewaySummaryBonded { + count: count_bonded_gateways as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + blacklisted: gateway::GatewaySummaryBlacklisted { + count: count_gateways_blacklisted as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + historical: gateway::GatewaySummaryHistorical { + count: all_historical_gateways as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + explorer: gateway::GatewaySummaryExplorer { + count: count_explorer_gateways as i32, + last_updated_utc: last_updated_utc.to_owned(), + }, + }, + }; + + queries::insert_summaries(pool, &nodes_summary, &network_summary, last_updated).await?; + + let mut log_lines: Vec = vec![]; + 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")); + + Ok(()) +} + +fn prepare_gateway_data( + gateways: &[DescribedGateway], + gateways_blacklisted: &HashSet, + explorer_gateways: Vec, + skimmed_gateways: Vec, +) -> anyhow::Result> { + let mut gateway_records = Vec::new(); + + for gateway in gateways { + let identity_key = gateway.bond.identity(); + let bonded = true; + let last_updated_utc = chrono::offset::Utc::now().timestamp(); + let blacklisted = gateways_blacklisted.contains(identity_key); + + let self_described = gateway + .self_described + .as_ref() + .and_then(|v| serde_json::to_string(&v).ok()); + + let explorer_pretty_bond = explorer_gateways + .iter() + .find(|g| g.gateway.identity_key.eq(identity_key)); + let explorer_pretty_bond = explorer_pretty_bond.and_then(|g| serde_json::to_string(g).ok()); + + let performance = skimmed_gateways + .iter() + .find(|g| g.ed25519_identity_pubkey.eq(identity_key)) + .map(|g| g.performance) + .unwrap_or_default() + .round_to_integer(); + + gateway_records.push(GatewayRecord { + identity_key: identity_key.to_owned(), + bonded, + blacklisted, + self_described, + explorer_pretty_bond, + last_updated_utc, + performance, + }); + } + + Ok(gateway_records) +} + +fn prepare_mixnode_data( + mixnodes: &[MixNodeBondAnnotated], + mixnodes_described: Vec, + delegation_program_members: Vec, +) -> anyhow::Result> { + let mut mixnode_records = Vec::new(); + + for mixnode in mixnodes { + let mix_id = mixnode.mix_id(); + 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; + // Contains all the information including what's above + let full_details = serde_json::to_string(&mixnode)?; + + let mixnode_described = mixnodes_described.iter().find(|m| m.bond.mix_id == mix_id); + let self_described = mixnode_described.and_then(|v| serde_json::to_string(v).ok()); + let is_dp_delegatee = delegation_program_members.contains(&mix_id); + + let last_updated_utc = chrono::offset::Utc::now().timestamp(); + + mixnode_records.push(MixnodeRecord { + mix_id, + identity_key: identity_key.to_owned(), + bonded, + total_stake, + host, + http_port, + blacklisted, + full_details, + self_described, + last_updated_utc, + is_dp_delegatee, + }); + } + + Ok(mixnode_records) +} + +async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> { + let mut conn = pool.acquire().await?; + + let all_historical_gateways = sqlx::query_scalar!(r#"SELECT count(id) FROM gateways"#) + .fetch_one(&mut *conn) + .await? as usize; + + let all_historical_mixnodes = sqlx::query_scalar!(r#"SELECT count(id) FROM mixnodes"#) + .fetch_one(&mut *conn) + .await? as usize; + + Ok((all_historical_gateways, all_historical_mixnodes)) +} + +async fn get_delegation_program_details( + network_details: &NymNetworkDetails, +) -> anyhow::Result> { + let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(network_details)?; + + // TODO dz should this be configurable? + let client = NyxdClient::connect(config, "https://rpc.nymtech.net") + .map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?; + + let account_id = AccountId::from_str(DELEGATION_PROGRAM_WALLET) + .map_err(|e| anyhow!("Invalid bech32 address: {}", e))?; + + let delegations = client.get_all_delegator_delegations(&account_id).await?; + + let mix_ids: Vec = delegations + .iter() + .map(|delegation| delegation.mix_id) + .collect(); + + Ok(mix_ids) +} + +fn decimal_to_i64(decimal: Decimal) -> i64 { + // Convert the underlying Uint128 to a u128 + let atomics = decimal.atomics().u128(); + let precision = 1_000_000_000_000_000_000u128; + + // Get the fractional part + let fractional = atomics % precision; + + // Get the integer part + let integer = atomics / precision; + + // Combine them into a float + let float_value = integer as f64 + (fractional as f64 / 1_000_000_000_000_000_000_f64); + + // Limit to 6 decimal places + let rounded_value = (float_value * 1_000_000.0).round() / 1_000_000.0; + + rounded_value as i64 +}