From 61361d84cfd2506cb2a9ae30e53d6b43656763ae Mon Sep 17 00:00:00 2001 From: dynco-nym <173912580+dynco-nym@users.noreply.github.com> Date: Thu, 26 Jun 2025 14:32:41 +0200 Subject: [PATCH] Move stuff around --- .../src/db/queries/gateways.rs | 2 +- .../src/db/queries/mixnodes.rs | 2 +- .../src/db/queries/nym_nodes.rs | 2 +- .../src/db/queries/scraper.rs | 2 +- .../nym-node-status-api/src/main.rs | 11 +- .../error.rs | 0 .../src/metrics_scraper/mod.rs | 284 +++++++++++++++++ .../mod.rs => node_scraper/description.rs} | 94 +----- .../helpers.rs | 2 +- .../src/node_scraper/mod.rs | 288 +----------------- .../src/node_scraper/packet_stats.rs | 116 +++++++ 11 files changed, 424 insertions(+), 379 deletions(-) rename nym-node-status-api/nym-node-status-api/src/{node_scraper => metrics_scraper}/error.rs (100%) create mode 100644 nym-node-status-api/nym-node-status-api/src/metrics_scraper/mod.rs rename nym-node-status-api/nym-node-status-api/src/{mixnet_scraper/mod.rs => node_scraper/description.rs} (53%) rename nym-node-status-api/nym-node-status-api/src/{mixnet_scraper => node_scraper}/helpers.rs (99%) create mode 100644 nym-node-status-api/nym-node-status-api/src/node_scraper/packet_stats.rs diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs index adb4241835..961da755b2 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/gateways.rs @@ -6,7 +6,7 @@ use crate::{ DbPool, }, http::models::Gateway, - mixnet_scraper::helpers::NodeDescriptionResponse, + node_scraper::helpers::NodeDescriptionResponse, }; use futures_util::TryStreamExt; use sqlx::{pool::PoolConnection, Sqlite}; diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs index 98709e099e..ec8cf5c36c 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs @@ -10,7 +10,7 @@ use crate::{ DbPool, }, http::models::{DailyStats, Mixnode}, - mixnet_scraper::helpers::NodeDescriptionResponse, + node_scraper::helpers::NodeDescriptionResponse, }; pub(crate) async fn update_mixnodes( diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs index 5595b10fcc..ec32557d8c 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs @@ -13,7 +13,7 @@ use crate::{ models::{NymNodeDto, NymNodeInsertRecord}, DbPool, }, - mixnet_scraper::helpers::NodeDescriptionResponse, + node_scraper::helpers::NodeDescriptionResponse, }; pub(crate) async fn get_all_nym_nodes(pool: &DbPool) -> anyhow::Result> { diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs index 02113affc4..9dbb5d4484 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs @@ -7,7 +7,7 @@ use crate::{ }, DbPool, }, - mixnet_scraper::helpers::NodeDescriptionResponse, + node_scraper::helpers::NodeDescriptionResponse, utils::now_utc, }; use anyhow::Result; 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 7db3015a7d..63ddba3d54 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 @@ -9,7 +9,7 @@ mod cli; mod db; mod http; mod logging; -mod mixnet_scraper; +mod metrics_scraper; mod monitor; mod node_scraper; mod testruns; @@ -35,7 +35,11 @@ async fn main() -> anyhow::Result<()> { let db_pool = storage.pool_owned(); // Start the node scraper - let scraper = mixnet_scraper::Scraper::new(storage.pool_owned()); + let scraper = node_scraper::DescriptionScraper::new(storage.pool_owned()); + tokio::spawn(async move { + scraper.start().await; + }); + let scraper = node_scraper::PacketScraper::new(storage.pool_owned()); tokio::spawn(async move { scraper.start().await; }); @@ -74,7 +78,8 @@ async fn main() -> anyhow::Result<()> { 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; + metrics_scraper::spawn_in_background(db_pool_scraper, args_clone.nym_api_client_timeout) + .await; tracing::info!("Started metrics scraper task"); }); 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/metrics_scraper/error.rs similarity index 100% rename from nym-node-status-api/nym-node-status-api/src/node_scraper/error.rs rename to nym-node-status-api/nym-node-status-api/src/metrics_scraper/error.rs diff --git a/nym-node-status-api/nym-node-status-api/src/metrics_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/metrics_scraper/mod.rs new file mode 100644 index 0000000000..aeab6cb4b9 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/metrics_scraper/mod.rs @@ -0,0 +1,284 @@ +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_validator_client::{ + client::{NodeId, NymNodeDetails}, + models::{DescribedNodeType, NymNodeDescription}, + NymApiClient, +}; +use time::OffsetDateTime; + +use nym_statistics_common::types::SessionType; +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); +const STALE_DURATION: Duration = Duration::from_secs(86400 * 365); //one year + +#[instrument(level = "info", name = "metrics_scraper", skip_all)] +pub(crate) async fn spawn_in_background(db_pool: DbPool, nym_api_client_timeout: Duration) { + let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env(); + + loop { + 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 nym_api = nym_http_api_client::ClientBuilder::new_with_url(default_api_url) + .no_hickory_dns() + .with_timeout(nym_api_client_timeout) + .build::<&str>()?; + + let api_client = NymApiClient::from(nym_api); + + //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, + } + } + + #[instrument(level = "info", name = "metrics_scraper", skip_all)] + 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::warn!("{e}"); + None + } + } + } + Err(e) => { + tracing::warn!("{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") + .no_hickory_dns() + .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-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/description.rs similarity index 53% rename from nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs rename to nym-node-status-api/nym-node-status-api/src/node_scraper/description.rs index 686d043987..7a482f7ca3 100644 --- a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/description.rs @@ -1,41 +1,36 @@ +use super::helpers::scrape_and_store_description; +use anyhow::Result; +use sqlx::SqlitePool; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -pub mod helpers; -use anyhow::Result; -use helpers::{scrape_and_store_description, scrape_and_store_packet_stats}; -use sqlx::SqlitePool; use tracing::{debug, error, instrument, warn}; use crate::db::models::ScraperNodeInfo; use crate::db::queries::get_nodes_for_scraping; const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); -const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); const QUEUE_CHECK_INTERVAL: Duration = Duration::from_millis(250); const MAX_CONCURRENT_TASKS: usize = 5; static TASK_COUNTER: AtomicUsize = AtomicUsize::new(0); static TASK_ID_COUNTER: AtomicUsize = AtomicUsize::new(0); -pub struct Scraper { +pub struct DescriptionScraper { pool: SqlitePool, description_queue: Arc>>, - packet_queue: Arc>>, } -impl Scraper { +impl DescriptionScraper { pub fn new(pool: SqlitePool) -> Self { Self { pool, description_queue: Arc::new(Mutex::new(Vec::new())), - packet_queue: Arc::new(Mutex::new(Vec::new())), } } pub async fn start(&self) { self.spawn_description_scraper().await; - self.spawn_packet_scraper().await; } async fn spawn_description_scraper(&self) { @@ -53,22 +48,6 @@ impl Scraper { }); } - async fn spawn_packet_scraper(&self) { - let pool = self.pool.clone(); - let queue = self.packet_queue.clone(); - tracing::info!("Starting packet scraper"); - - tokio::spawn(async move { - loop { - if let Err(e) = Self::run_packet_scraper(&pool, queue.clone()).await { - error!(name: "packet_scraper", "Packet scraper failed: {}", e); - } - debug!(name: "packet_scraper", "Sleeping for {}s", PACKET_SCRAPE_INTERVAL.as_secs()); - tokio::time::sleep(PACKET_SCRAPE_INTERVAL).await; - } - }); - } - #[instrument(level = "info", name = "description_scraper", skip_all)] async fn run_description_scraper( pool: &SqlitePool, @@ -86,24 +65,6 @@ impl Scraper { Ok(()) } - #[instrument(level = "info", name = "packet_scraper", skip_all)] - async fn run_packet_scraper( - pool: &SqlitePool, - queue: Arc>>, - ) -> Result<()> { - let nodes = get_nodes_for_scraping(pool).await?; - tracing::info!("Querying {} mixing nodes", nodes.len()); - if let Ok(mut queue_lock) = queue.lock() { - queue_lock.extend(nodes); - } else { - warn!("Failed to acquire packet queue lock"); - return Ok(()); - } - - Self::process_packet_queue(pool, queue).await; - Ok(()) - } - async fn process_description_queue(pool: &SqlitePool, queue: Arc>>) { loop { let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); @@ -147,50 +108,7 @@ impl Scraper { tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; } } - } - async fn process_packet_queue(pool: &SqlitePool, queue: Arc>>) { - loop { - let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); - - if running_tasks < MAX_CONCURRENT_TASKS { - let node = { - if let Ok(mut queue_lock) = queue.lock() { - if queue_lock.is_empty() { - TASK_ID_COUNTER.store(0, Ordering::Relaxed); - break; - } - queue_lock.remove(0) - } else { - warn!("Failed to acquire packet queue lock"); - break; - } - }; - - TASK_COUNTER.fetch_add(1, Ordering::Relaxed); - let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed); - let pool = pool.clone(); - - tokio::spawn(async move { - match scrape_and_store_packet_stats(&pool, &node).await { - Ok(_) => debug!( - "📊 ✅ Packet stats task #{} for node {} complete", - task_id, - node.node_id() - ), - Err(e) => debug!( - "📊 ❌ Packet stats task #{} for {} {} failed: {}", - task_id, - node.node_kind, - node.node_id(), - e - ), - } - TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); - }); - } else { - tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; - } - } + // TODO After all tasks complete, write results to the DB } } diff --git a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/helpers.rs similarity index 99% rename from nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs rename to nym-node-status-api/nym-node-status-api/src/node_scraper/helpers.rs index 986f1472c5..738cef7dc0 100644 --- a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/helpers.rs @@ -191,7 +191,7 @@ pub async fn scrape_and_store_packet_stats( let timestamp_utc = timestamp.unix_timestamp(); insert_node_packet_stats(pool, &node.node_kind, &stats, timestamp_utc).await?; - // Update daily stats + // TODO dz does this need to run every time? update_daily_stats(pool, node, timestamp, &stats).await?; Ok(()) diff --git a/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs index aeab6cb4b9..851db3c1e4 100644 --- a/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/mod.rs @@ -1,284 +1,6 @@ -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_validator_client::{ - client::{NodeId, NymNodeDetails}, - models::{DescribedNodeType, NymNodeDescription}, - NymApiClient, -}; -use time::OffsetDateTime; +pub(crate) mod description; +pub(crate) mod helpers; +pub(crate) mod packet_stats; -use nym_statistics_common::types::SessionType; -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); -const STALE_DURATION: Duration = Duration::from_secs(86400 * 365); //one year - -#[instrument(level = "info", name = "metrics_scraper", skip_all)] -pub(crate) async fn spawn_in_background(db_pool: DbPool, nym_api_client_timeout: Duration) { - let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env(); - - loop { - 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 nym_api = nym_http_api_client::ClientBuilder::new_with_url(default_api_url) - .no_hickory_dns() - .with_timeout(nym_api_client_timeout) - .build::<&str>()?; - - let api_client = NymApiClient::from(nym_api); - - //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, - } - } - - #[instrument(level = "info", name = "metrics_scraper", skip_all)] - 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::warn!("{e}"); - None - } - } - } - Err(e) => { - tracing::warn!("{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") - .no_hickory_dns() - .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, - } -} +pub(crate) use description::DescriptionScraper; +pub(crate) use packet_stats::PacketScraper; diff --git a/nym-node-status-api/nym-node-status-api/src/node_scraper/packet_stats.rs b/nym-node-status-api/nym-node-status-api/src/node_scraper/packet_stats.rs new file mode 100644 index 0000000000..561830bfda --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/node_scraper/packet_stats.rs @@ -0,0 +1,116 @@ +use super::helpers::scrape_and_store_packet_stats; +use anyhow::Result; +use sqlx::SqlitePool; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tracing::{debug, error, instrument, warn}; + +use crate::db::models::ScraperNodeInfo; +use crate::db::queries::get_nodes_for_scraping; + +const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); +const QUEUE_CHECK_INTERVAL: Duration = Duration::from_millis(250); +const MAX_CONCURRENT_TASKS: usize = 5; + +static TASK_COUNTER: AtomicUsize = AtomicUsize::new(0); +static TASK_ID_COUNTER: AtomicUsize = AtomicUsize::new(0); + +pub struct PacketScraper { + pool: SqlitePool, + packet_queue: Arc>>, +} + +impl PacketScraper { + pub fn new(pool: SqlitePool) -> Self { + Self { + pool, + packet_queue: Arc::new(Mutex::new(Vec::new())), + } + } + + pub async fn start(&self) { + self.spawn_packet_scraper().await; + } + + async fn spawn_packet_scraper(&self) { + let pool = self.pool.clone(); + let queue = self.packet_queue.clone(); + tracing::info!("Starting packet scraper"); + + tokio::spawn(async move { + loop { + if let Err(e) = Self::run_packet_scraper(&pool, queue.clone()).await { + error!(name: "packet_scraper", "Packet scraper failed: {}", e); + } + debug!(name: "packet_scraper", "Sleeping for {}s", PACKET_SCRAPE_INTERVAL.as_secs()); + tokio::time::sleep(PACKET_SCRAPE_INTERVAL).await; + } + }); + } + + #[instrument(level = "info", name = "packet_scraper", skip_all)] + async fn run_packet_scraper( + pool: &SqlitePool, + queue: Arc>>, + ) -> Result<()> { + let nodes = get_nodes_for_scraping(pool).await?; + tracing::info!("Querying {} mixing nodes", nodes.len()); + if let Ok(mut queue_lock) = queue.lock() { + queue_lock.extend(nodes); + } else { + warn!("Failed to acquire packet queue lock"); + return Ok(()); + } + + Self::process_packet_queue(pool, queue).await; + Ok(()) + } + + async fn process_packet_queue(pool: &SqlitePool, queue: Arc>>) { + loop { + let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); + + if running_tasks < MAX_CONCURRENT_TASKS { + let node = { + if let Ok(mut queue_lock) = queue.lock() { + if queue_lock.is_empty() { + TASK_ID_COUNTER.store(0, Ordering::Relaxed); + break; + } + queue_lock.remove(0) + } else { + warn!("Failed to acquire packet queue lock"); + break; + } + }; + + TASK_COUNTER.fetch_add(1, Ordering::Relaxed); + let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + let pool = pool.clone(); + + tokio::spawn(async move { + match scrape_and_store_packet_stats(&pool, &node).await { + Ok(_) => debug!( + "📊 ✅ Packet stats task #{} for node {} complete", + task_id, + node.node_id() + ), + Err(e) => debug!( + "📊 ❌ Packet stats task #{} for {} {} failed: {}", + task_id, + node.node_kind, + node.node_id(), + e + ), + } + TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); + }); + } else { + tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; + } + } + + // TODO After all tasks complete, write results to the DB + } +}