diff --git a/Cargo.lock b/Cargo.lock index 936dadfbd6..cd9744933a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5536,6 +5536,7 @@ version = "0.1.0" dependencies = [ "nym-credentials-interface", "nym-sphinx", + "nym-statistics-common", "sqlx", "thiserror", "time", @@ -6077,6 +6078,7 @@ dependencies = [ "nym-network-defaults", "nym-node-requests", "nym-node-status-client", + "nym-statistics-common", "nym-task", "nym-validator-client", "regex", @@ -6088,6 +6090,7 @@ dependencies = [ "strum 0.26.3", "strum_macros 0.26.4", "thiserror", + "time", "tokio", "tokio-util", "tower-http", diff --git a/common/gateway-stats-storage/Cargo.toml b/common/gateway-stats-storage/Cargo.toml index d439b34a16..1e816e0233 100644 --- a/common/gateway-stats-storage/Cargo.toml +++ b/common/gateway-stats-storage/Cargo.toml @@ -22,6 +22,7 @@ tracing = { workspace = true } nym-sphinx = { path = "../nymsphinx" } nym-credentials-interface = { path = "../credentials-interface" } +nym-statistics-common = { path = "../statistics" } [build-dependencies] diff --git a/common/gateway-stats-storage/src/lib.rs b/common/gateway-stats-storage/src/lib.rs index 453258b606..3f94b40a87 100644 --- a/common/gateway-stats-storage/src/lib.rs +++ b/common/gateway-stats-storage/src/lib.rs @@ -2,8 +2,9 @@ // SPDX-License-Identifier: GPL-3.0-only use error::StatsStorageError; -use models::{ActiveSession, FinishedSession, SessionType, StoredFinishedSession}; +use models::{ActiveSession, FinishedSession, StoredFinishedSession}; use nym_sphinx::DestinationAddressBytes; +use nym_statistics_common::gateways::SessionType; use sessions::SessionManager; use sqlx::ConnectOptions; use std::path::Path; diff --git a/common/gateway-stats-storage/src/models.rs b/common/gateway-stats-storage/src/models.rs index 6f53cf429a..f35883d28e 100644 --- a/common/gateway-stats-storage/src/models.rs +++ b/common/gateway-stats-storage/src/models.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only use nym_credentials_interface::TicketType; +use nym_statistics_common::gateways::SessionType; use sqlx::prelude::FromRow; use time::{Duration, OffsetDateTime}; @@ -25,42 +26,6 @@ pub struct FinishedSession { pub typ: SessionType, } -#[derive(PartialEq)] -pub enum SessionType { - Vpn, - Mixnet, - Unknown, -} - -impl SessionType { - pub fn to_string(&self) -> &str { - match self { - Self::Vpn => "vpn", - Self::Mixnet => "mixnet", - Self::Unknown => "unknown", - } - } - - pub fn from_string(s: &str) -> Self { - match s { - "vpn" => Self::Vpn, - "mixnet" => Self::Mixnet, - _ => Self::Unknown, - } - } -} - -impl From for SessionType { - fn from(value: TicketType) -> Self { - match value { - TicketType::V1MixnetEntry => Self::Mixnet, - TicketType::V1MixnetExit => Self::Mixnet, - TicketType::V1WireguardEntry => Self::Vpn, - TicketType::V1WireguardExit => Self::Vpn, - } - } -} - #[derive(FromRow)] pub(crate) struct StoredActiveSession { start_time: OffsetDateTime, diff --git a/common/statistics/src/gateways.rs b/common/statistics/src/gateways.rs index 398fda559a..97d75334f1 100644 --- a/common/statistics/src/gateways.rs +++ b/common/statistics/src/gateways.rs @@ -87,3 +87,39 @@ pub enum SessionEvent { client: DestinationAddressBytes, }, } + +#[derive(PartialEq)] +pub enum SessionType { + Vpn, + Mixnet, + Unknown, +} + +impl SessionType { + pub fn to_string(&self) -> &str { + match self { + Self::Vpn => "vpn", + Self::Mixnet => "mixnet", + Self::Unknown => "unknown", + } + } + + pub fn from_string(s: &str) -> Self { + match s { + "vpn" => Self::Vpn, + "mixnet" => Self::Mixnet, + _ => Self::Unknown, + } + } +} + +impl From for SessionType { + fn from(value: TicketType) -> Self { + match value { + TicketType::V1MixnetEntry => Self::Mixnet, + TicketType::V1MixnetExit => Self::Mixnet, + TicketType::V1WireguardEntry => Self::Vpn, + TicketType::V1WireguardExit => Self::Vpn, + } + } +} 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 05d1586ae9..a915d31555 100644 --- a/nym-node-status-api/nym-node-status-api/Cargo.toml +++ b/nym-node-status-api/nym-node-status-api/Cargo.toml @@ -26,6 +26,7 @@ nym-node-status-client = { path = "../nym-node-status-client" } nym-crypto = { path = "../../common/crypto", features = ["asymmetric", "serde"] } nym-explorer-client = { path = "../../explorer-api/explorer-client" } nym-network-defaults = { path = "../../common/network-defaults" } +nym-statistics-common = { path = "../../common/statistics"} nym-validator-client = { path = "../../common/client-libs/validator-client" } nym-task = { path = "../../common/task" } nym-node-requests = { path = "../../nym-node/nym-node-requests", features = ["openapi"] } @@ -36,8 +37,9 @@ serde_json = { workspace = true } serde_json_path = { workspace = true } strum = { workspace = true } strum_macros = { workspace = true } -sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "time"] } thiserror = { workspace = true } +time = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread"] } tokio-util = { workspace = true } tracing = { workspace = true } diff --git a/nym-node-status-api/nym-node-status-api/migrations/002_session_stats.sql b/nym-node-status-api/nym-node-status-api/migrations/002_session_stats.sql new file mode 100644 index 0000000000..d29fe986a3 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/migrations/002_session_stats.sql @@ -0,0 +1,17 @@ + +CREATE TABLE gateway_session_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + gateway_identity_key VARCHAR NOT NULL, + node_id INTEGER NOT NULL, + day DATE NOT NULL, + unique_active_clients INTEGER NOT NULL, + session_started INTEGER NOT NULL, + users_hashes VARCHAR, + vpn_sessions VARCHAR, + mixnet_sessions VARCHAR, + unknown_sessions VARCHAR, + UNIQUE (node_id, day) -- This constraint automatically creates an index + ); +CREATE INDEX idx_gateway_session_stats_identity_key ON gateway_session_stats (gateway_identity_key); +CREATE INDEX idx_gateway_session_stats_day ON gateway_session_stats (day); + 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 7d9e93ed24..9de09271bf 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 @@ -4,7 +4,9 @@ use crate::{ }; use nym_node_requests::api::v1::node::models::NodeDescription; use serde::{Deserialize, Serialize}; +use sqlx::FromRow; use strum_macros::{EnumString, FromRepr}; +use time::Date; use utoipa::ToSchema; pub(crate) struct GatewayRecord { @@ -333,3 +335,44 @@ pub struct GatewayInfoDto { pub self_described: Option, pub explorer_pretty_bond: Option, } + +#[derive(Debug, Clone, FromRow)] +pub struct GatewaySessionsRecord { + pub gateway_identity_key: String, + pub node_id: i64, + pub day: Date, + pub unique_active_clients: i64, + pub session_started: i64, + pub users_hashes: Option, + pub vpn_sessions: Option, + pub mixnet_sessions: Option, + pub unknown_sessions: Option, +} + +impl TryFrom for http::models::SessionStats { + type Error = anyhow::Error; + + fn try_from(value: GatewaySessionsRecord) -> Result { + let users_hashes = value.users_hashes.clone().unwrap_or("null".to_string()); + let vpn_sessions = value.vpn_sessions.clone().unwrap_or("null".to_string()); + let mixnet_sessions = value.mixnet_sessions.clone().unwrap_or("null".to_string()); + let unknown_sessions = value.unknown_sessions.clone().unwrap_or("null".to_string()); + + let users_hashes = serde_json::from_str(&users_hashes).unwrap_or(None); + let vpn_sessions = serde_json::from_str(&vpn_sessions).unwrap_or(None); + let mixnet_sessions = serde_json::from_str(&mixnet_sessions).unwrap_or(None); + let unknown_sessions = serde_json::from_str(&unknown_sessions).unwrap_or(None); + + Ok(http::models::SessionStats { + gateway_identity_key: value.gateway_identity_key.clone(), + node_id: value.node_id as u32, + day: value.day, + unique_active_clients: value.unique_active_clients, + session_started: value.session_started, + users_hashes, + vpn_sessions, + mixnet_sessions, + unknown_sessions, + }) + } +} diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/gateways_stats.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways_stats.rs new file mode 100644 index 0000000000..8ba813ee1f --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways_stats.rs @@ -0,0 +1,75 @@ +use crate::{ + db::{models::GatewaySessionsRecord, DbPool}, + http::models::SessionStats, +}; +use futures_util::TryStreamExt; +use time::Date; +use tracing::error; + +pub(crate) async fn insert_session_records( + pool: &DbPool, + records: Vec, +) -> anyhow::Result<()> { + let mut tx = pool.begin().await?; + for record in records { + sqlx::query!( + "INSERT OR IGNORE INTO gateway_session_stats + (gateway_identity_key, node_id, day, + unique_active_clients, session_started, users_hashes, + vpn_sessions, mixnet_sessions, unknown_sessions) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + record.gateway_identity_key, + record.node_id, + record.day, + record.unique_active_clients, + record.session_started, + record.users_hashes, + record.vpn_sessions, + record.mixnet_sessions, + record.unknown_sessions, + ) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + + Ok(()) +} + +pub(crate) async fn get_sessions_stats(pool: &DbPool) -> anyhow::Result> { + let mut conn = pool.acquire().await?; + let items = sqlx::query_as( + "SELECT gateway_identity_key, + node_id, + day, + unique_active_clients, + session_started, + users_hashes, + vpn_sessions, + mixnet_sessions, + unknown_sessions + FROM gateway_session_stats", + ) + .fetch(&mut *conn) + .try_collect::>() + .await?; + + let items: Vec = items + .into_iter() + .map(|item| item.try_into()) + .collect::>>() + .map_err(|e| { + error!("Conversion from database failed: {e}. Invalidly stored data?"); + e + })?; + + Ok(items) +} + +pub(crate) async fn delete_old_records(pool: &DbPool, cut_off: Date) -> anyhow::Result<()> { + let mut conn = pool.acquire().await?; + sqlx::query!("DELETE FROM gateway_session_stats WHERE day <= ?", cut_off) + .execute(&mut *conn) + .await?; + Ok(()) +} 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 fe22ec27aa..8c87ab2fc0 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 @@ -1,4 +1,5 @@ mod gateways; +mod gateways_stats; mod misc; mod mixnodes; mod summary; @@ -13,3 +14,5 @@ pub(crate) use mixnodes::{ ensure_mixnodes_still_bonded, get_all_mixnodes, get_daily_stats, insert_mixnodes, }; pub(crate) use summary::{get_summary, get_summary_history}; + +pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_session_records}; diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/metrics/mod.rs b/nym-node-status-api/nym-node-status-api/src/http/api/metrics/mod.rs new file mode 100644 index 0000000000..8703f92830 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/http/api/metrics/mod.rs @@ -0,0 +1,10 @@ +use axum::Router; + +use crate::http::state::AppState; + +pub(crate) mod sessions; + +pub(crate) fn routes() -> Router { + Router::new().nest("/sessions", sessions::routes()) + //eventually add other metrics type +} diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/metrics/sessions.rs b/nym-node-status-api/nym-node-status-api/src/http/api/metrics/sessions.rs new file mode 100644 index 0000000000..e2e4cd9ced --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/http/api/metrics/sessions.rs @@ -0,0 +1,83 @@ +use axum::{ + extract::{Query, State}, + Json, Router, +}; +use time::Date; +use tracing::instrument; + +use crate::http::{ + error::{HttpError, HttpResult}, + models::SessionStats, + state::AppState, + PagedResult, Pagination, +}; + +pub(crate) fn routes() -> Router { + Router::new().route("/", axum::routing::get(get_all_sessions)) + // .route("/:node_id", axum::routing::get(get_node_sessions)) + // .route("/:day", axum::routing::get(get_daily_sessions)) +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, utoipa::IntoParams)] +#[into_params(parameter_in = Query)] +pub(crate) struct SessionQueryParams { + size: Option, + page: Option, + node_id: Option, + day: Option, +} + +#[utoipa::path( + tag = "Sessions", + get, + params( + SessionQueryParams + ), + path = "/v2/metrics/sessions", + responses( + (status = 200, body = PagedSessionStats) + ) +)] +#[instrument(level = tracing::Level::DEBUG, skip(state))] +async fn get_all_sessions( + Query(SessionQueryParams { + size, + page, + node_id, + day, + }): Query, + State(state): State, +) -> HttpResult>> { + let db = state.db_pool(); + let res = state.cache().get_sessions_stats(db).await; + + let day_filtered = if let Some(day) = day { + if let Ok(parsed_day) = + Date::parse(&day, &time::format_description::well_known::Iso8601::DATE) + { + res.into_iter().filter(|s| s.day == parsed_day).collect() + } else { + return Err(HttpError::invalid_input(day)); + } + } else { + res + }; + + let day_and_node_filtered = if let Some(node_id) = node_id { + if let Ok(parsed_node_id) = node_id.parse::() { + day_filtered + .into_iter() + .filter(|s| s.node_id == parsed_node_id) + .collect() + } else { + return Err(HttpError::invalid_input(node_id)); + } + } else { + day_filtered + }; + + Ok(Json(PagedResult::paginate( + Pagination { size, page }, + day_and_node_filtered, + ))) +} diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs b/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs index ed24fa80f5..6483a44519 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/api/mod.rs @@ -8,6 +8,7 @@ use utoipa_swagger_ui::SwaggerUi; use crate::http::{server::HttpServer, state::AppState}; pub(crate) mod gateways; +pub(crate) mod metrics; pub(crate) mod mixnodes; pub(crate) mod services; pub(crate) mod summary; @@ -34,7 +35,8 @@ impl RouterBuilder { .nest("/gateways", gateways::routes()) .nest("/mixnodes", mixnodes::routes()) .nest("/services", services::routes()) - .nest("/summary", summary::routes()), + .nest("/summary", summary::routes()) + .nest("/metrics", metrics::routes()), ) .nest( "/internal", diff --git a/nym-node-status-api/nym-node-status-api/src/http/api_docs.rs b/nym-node-status-api/nym-node-status-api/src/http/api_docs.rs index aa86a56ab0..fec4d25cd5 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/api_docs.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/api_docs.rs @@ -1,4 +1,4 @@ -use crate::http::{Gateway, GatewaySkinny, Mixnode, Service}; +use crate::http::{Gateway, GatewaySkinny, Mixnode, Service, SessionStats}; use utoipa::OpenApi; use utoipauto::utoipauto; diff --git a/nym-node-status-api/nym-node-status-api/src/http/mod.rs b/nym-node-status-api/nym-node-status-api/src/http/mod.rs index 1cc317337f..b1a7bf742d 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/mod.rs @@ -1,4 +1,4 @@ -use models::{Gateway, GatewaySkinny, Mixnode, Service}; +use models::{Gateway, GatewaySkinny, Mixnode, Service, SessionStats}; pub(crate) mod api; pub(crate) mod api_docs; @@ -20,6 +20,7 @@ pub(crate) mod state; PagedGatewaySkinny = PagedResult, PagedMixnode = PagedResult, PagedService = PagedResult, + PagedSessionStats = PagedResult )] pub struct PagedResult { pub page: usize, 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 aee2124b6c..19060d1ce8 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 @@ -74,3 +74,16 @@ pub(crate) struct SummaryHistory { pub value_json: serde_json::Value, pub timestamp_utc: String, } + +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +pub struct SessionStats { + pub gateway_identity_key: String, + pub node_id: u32, + pub day: time::Date, + pub unique_active_clients: i64, + pub session_started: i64, + pub users_hashes: Option, + pub vpn_sessions: Option, + pub mixnet_sessions: Option, + pub unknown_sessions: Option, +} 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 393da5c0b8..0c79c0e16f 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 @@ -9,6 +9,8 @@ use crate::{ http::models::{DailyStats, Gateway, Mixnode, SummaryHistory}, }; +use super::models::SessionStats; + #[derive(Debug, Clone)] pub(crate) struct AppState { db_pool: DbPool, @@ -53,6 +55,7 @@ static GATEWAYS_LIST_KEY: &str = "gateways"; static MIXNODES_LIST_KEY: &str = "mixnodes"; static MIXSTATS_LIST_KEY: &str = "mixstats"; static SUMMARY_HISTORY_LIST_KEY: &str = "summary-history"; +static SESSION_STATS_LIST_KEY: &str = "session-stats"; #[derive(Debug, Clone)] pub(crate) struct HttpCache { @@ -60,6 +63,7 @@ pub(crate) struct HttpCache { mixnodes: Cache>>>, mixstats: Cache>>>, history: Cache>>>, + session_stats: Cache>>>, } impl HttpCache { @@ -81,6 +85,10 @@ impl HttpCache { .max_capacity(2) .time_to_live(Duration::from_secs(ttl_seconds)) .build(), + session_stats: Cache::builder() + .max_capacity(2) + .time_to_live(Duration::from_secs(ttl_seconds)) + .build(), } } @@ -238,4 +246,39 @@ impl HttpCache { }) .await } + + pub async fn get_sessions_stats(&self, db: &DbPool) -> Vec { + match self.session_stats.get(SESSION_STATS_LIST_KEY).await { + Some(guard) => { + let read_lock = guard.read().await; + read_lock.to_vec() + } + None => { + let session_stats = crate::db::queries::get_sessions_stats(db) + .await + .unwrap_or_default(); + self.upsert_sessions_stats(session_stats.clone()).await; + session_stats + } + } + } + + pub async fn upsert_sessions_stats( + &self, + session_stats: Vec, + ) -> Entry>>> { + self.session_stats + .entry_by_ref(SESSION_STATS_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 = session_stats; + v.clone() + } else { + Arc::new(RwLock::new(session_stats)) + } + }) + .await + } } 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 6274a01fe5..4fd01d7960 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 @@ -7,6 +7,7 @@ mod db; mod http; mod logging; mod monitor; +mod node_scraper; mod testruns; #[tokio::main] @@ -44,6 +45,12 @@ async fn main() -> anyhow::Result<()> { testruns::spawn(storage.pool_owned(), args.testruns_refresh_interval).await; + let db_pool_scraper = storage.pool_owned(); + tokio::spawn(async move { + node_scraper::spawn_in_background(db_pool_scraper, args_clone.nym_api_client_timeout).await; + tracing::info!("Started metrics scraper task"); + }); + let shutdown_handles = http::server::start_http_api( storage.pool_owned(), args.http_port, diff --git a/nym-node-status-api/nym-node-status-api/src/node_scraper/error.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/error.rs new file mode 100644 index 0000000000..42f96647e3 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/error.rs @@ -0,0 +1,18 @@ +use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT; +use nym_node_requests::api::client::NymNodeApiClientError; +use nym_validator_client::client::NodeId; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum NodeScraperError { + #[error("node {node_id} has provided malformed host information ({host}: {source}")] + MalformedHost { + host: String, + node_id: NodeId, + #[source] + source: NymNodeApiClientError, + }, + + #[error("node {node_id} with host '{host}' doesn't seem to expose its declared http port nor any of the standard API ports, i.e.: 80, 443 or {}", DEFAULT_NYM_NODE_HTTP_PORT)] + NoHttpPortsAvailable { host: String, node_id: NodeId }, +} 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 new file mode 100644 index 0000000000..fb8cca4c29 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs @@ -0,0 +1,279 @@ +use crate::db::{models::GatewaySessionsRecord, queries, DbPool}; +use error::NodeScraperError; + +use nym_network_defaults::{NymNetworkDetails, DEFAULT_NYM_NODE_HTTP_PORT}; +use nym_node_requests::api::{client::NymNodeApiClientExt, v1::metrics::models::SessionStats}; +use nym_statistics_common::gateways::SessionType; +use nym_validator_client::{ + client::{NodeId, NymNodeDetails}, + models::{DescribedNodeType, NymNodeDescription}, + NymApiClient, +}; +use time::OffsetDateTime; + +use std::collections::HashMap; +use tokio::time::Duration; +use tracing::instrument; + +mod error; + +const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); +const REFRESH_INTERVAL: Duration = Duration::from_secs(60 * 60 * 6); //6h, data only update once a day +const STALE_DURATION: Duration = Duration::from_secs(86400 * 365); //one year + +#[instrument(level = "debug", name = "node_scraper", skip_all)] +pub(crate) async fn spawn_in_background(db_pool: DbPool, nym_api_client_timeout: Duration) { + let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env(); + + loop { + //No graceful shutdown? + tracing::info!("Refreshing node self-described metrics..."); + + if let Err(e) = run(&db_pool, &network_defaults, nym_api_client_timeout).await { + tracing::error!( + "Metrics collection failed: {e}, retrying in {}s...", + FAILURE_RETRY_DELAY.as_secs() + ); + + tokio::time::sleep(FAILURE_RETRY_DELAY).await; + } else { + tracing::info!( + "Metrics successfully collected, sleeping for {}s...", + REFRESH_INTERVAL.as_secs() + ); + tokio::time::sleep(REFRESH_INTERVAL).await; + } + } +} + +async fn run( + pool: &DbPool, + network_details: &NymNetworkDetails, + nym_api_client_timeout: Duration, +) -> 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 api_client = NymApiClient::new_with_timeout(default_api_url, nym_api_client_timeout); + + //SW TBC what nodes exactly need to be scraped, the skimmed node endpoint seems to return more nodes + let bonded_nodes = api_client.get_all_bonded_nym_nodes().await?; + let all_nodes = api_client.get_all_described_nodes().await?; //legacy node that did not upgrade the contract bond yet + tracing::debug!("Fetched {} total nodes", all_nodes.len()); + + let mut nodes_to_scrape: HashMap = bonded_nodes + .into_iter() + .map(|n| (n.node_id(), n.into())) + .collect(); + + all_nodes + .into_iter() + .filter(|n| n.contract_node_type != DescribedNodeType::LegacyMixnode) + .for_each(|n| { + nodes_to_scrape.entry(n.node_id).or_insert_with(|| n.into()); + }); + tracing::debug!("Will try to scrape {} nodes", nodes_to_scrape.len()); + + let mut session_records = Vec::new(); + for n in nodes_to_scrape.into_values() { + if let Some(stat) = n.try_scrape_metrics().await { + session_records.push(prepare_session_data(stat, &n)); + } + } + + queries::insert_session_records(pool, session_records) + .await + .map(|_| { + tracing::debug!("Session info written to DB!"); + })?; + let cut_off_date = (OffsetDateTime::now_utc() - STALE_DURATION).date(); + queries::delete_old_records(pool, cut_off_date) + .await + .map(|_| { + tracing::debug!("Cleared old data before {}", cut_off_date); + })?; + + Ok(()) +} + +#[derive(Debug)] +struct MetricsScrapingData { + host: String, + node_id: NodeId, + id_key: String, + port: Option, +} + +impl MetricsScrapingData { + pub fn new( + host: impl Into, + node_id: NodeId, + id_key: String, + port: Option, + ) -> Self { + MetricsScrapingData { + host: host.into(), + node_id, + id_key, + port, + } + } + + async fn try_scrape_metrics(&self) -> Option { + match self.try_get_client().await { + Ok(client) => { + match client.get_sessions_metrics().await { + Ok(session_stats) => { + if session_stats.update_time != OffsetDateTime::UNIX_EPOCH { + Some(session_stats) + } else { + //means no data + None + } + } + Err(e) => { + tracing::error!("[metrics scraper]: {e}"); + None + } + } + } + Err(e) => { + tracing::error!("[metrics scraper]: {e}"); + None + } + } + } + + async fn try_get_client(&self) -> Result { + // first try the standard port in case the operator didn't put the node behind the proxy, + // then default https (443) + // finally default http (80) + let mut addresses_to_try = vec![ + format!("http://{0}:{DEFAULT_NYM_NODE_HTTP_PORT}", self.host), // 'standard' nym-node + format!("https://{0}", self.host), // node behind https proxy (443) + format!("http://{0}", self.host), // node behind http proxy (80) + ]; + + // note: I removed 'standard' legacy mixnode port because it should now be automatically pulled via + // the 'custom_port' since it should have been present in the contract. + + if let Some(port) = self.port { + addresses_to_try.insert(0, format!("http://{0}:{port}", self.host)); + } + + for address in addresses_to_try { + // if provided host was malformed, no point in continuing + let client = match nym_node_requests::api::Client::builder(address).and_then(|b| { + b.with_timeout(Duration::from_secs(5)) + .with_user_agent("node-status-api-metrics-scraper") + .build() + }) { + Ok(client) => client, + Err(err) => { + return Err(NodeScraperError::MalformedHost { + host: self.host.to_string(), + node_id: self.node_id, + source: err, + }); + } + }; + + if let Ok(health) = client.get_health().await { + if health.status.is_up() { + return Ok(client); + } + } + } + + Err(NodeScraperError::NoHttpPortsAvailable { + host: self.host.to_string(), + node_id: self.node_id, + }) + } +} + +impl From for MetricsScrapingData { + fn from(value: NymNodeDetails) -> Self { + MetricsScrapingData::new( + value.bond_information.node.host.clone(), + value.node_id(), + value.bond_information.node.identity_key, + value.bond_information.node.custom_http_port, + ) + } +} + +impl From for MetricsScrapingData { + fn from(value: NymNodeDescription) -> Self { + MetricsScrapingData::new( + value.description.host_information.ip_address[0].to_string(), + value.node_id, + value.ed25519_identity_key().to_base58_string(), + None, + ) + } +} + +fn prepare_session_data( + stat: SessionStats, + node_data: &MetricsScrapingData, +) -> GatewaySessionsRecord { + let users_hashes = if !stat.unique_active_users_hashes.is_empty() { + Some(serde_json::to_string(&stat.unique_active_users_hashes).unwrap()) + } else { + None + }; + let vpn_durations = stat + .sessions + .iter() + .filter(|s| SessionType::from_string(&s.typ) == SessionType::Vpn) + .map(|s| s.duration_ms) + .collect::>(); + + let mixnet_durations = stat + .sessions + .iter() + .filter(|s| SessionType::from_string(&s.typ) == SessionType::Mixnet) + .map(|s| s.duration_ms) + .collect::>(); + + let unkown_durations = stat + .sessions + .iter() + .filter(|s| SessionType::from_string(&s.typ) == SessionType::Unknown) + .map(|s| s.duration_ms) + .collect::>(); + + let vpn_sessions = if !vpn_durations.is_empty() { + Some(serde_json::to_string(&vpn_durations).unwrap()) + } else { + None + }; + let mixnet_sessions = if !mixnet_durations.is_empty() { + Some(serde_json::to_string(&mixnet_durations).unwrap()) + } else { + None + }; + let unknown_sessions = if !unkown_durations.is_empty() { + Some(serde_json::to_string(&unkown_durations).unwrap()) + } else { + None + }; + + GatewaySessionsRecord { + gateway_identity_key: node_data.id_key.clone(), + node_id: node_data.node_id as i64, + day: stat.update_time.date(), + unique_active_clients: stat.unique_active_users as i64, + session_started: stat.sessions_started as i64, + users_hashes, + vpn_sessions, + mixnet_sessions, + unknown_sessions, + } +} diff --git a/nym-node/nym-node-requests/src/api/client.rs b/nym-node/nym-node-requests/src/api/client.rs index 11ec66a4b9..068b128ef2 100644 --- a/nym-node/nym-node-requests/src/api/client.rs +++ b/nym-node/nym-node-requests/src/api/client.rs @@ -19,6 +19,7 @@ use crate::api::v1::network_requester::models::NetworkRequester; pub use nym_http_api_client::Client; use super::v1::gateway::models::Wireguard; +use super::v1::metrics::models::SessionStats; pub type NymNodeApiClientError = HttpClientError; @@ -87,6 +88,11 @@ pub trait NymNodeApiClientExt: ApiClient { self.get_json_from(routes::api::v1::gateway::client_interfaces::wireguard_absolute()) .await } + + async fn get_sessions_metrics(&self) -> Result { + self.get_json_from(routes::api::v1::metrics::sessions_absolute()) + .await + } } #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] diff --git a/nym-node/nym-node-requests/src/api/v1/metrics/models.rs b/nym-node/nym-node-requests/src/api/v1/metrics/models.rs index a32506db2d..4620db05f9 100644 --- a/nym-node/nym-node-requests/src/api/v1/metrics/models.rs +++ b/nym-node/nym-node-requests/src/api/v1/metrics/models.rs @@ -50,6 +50,7 @@ pub struct SessionStats { pub unique_active_users: u32, + #[serde(default = "Vec::new")] // field was added later pub unique_active_users_hashes: Vec, pub sessions: Vec,