Move stuff around

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};

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(

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<Vec<NymNodeDto>> {

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;

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");
     });
 

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<NodeId, MetricsScrapingData> = 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<u16>,
+}
+
+impl MetricsScrapingData {
+    pub fn new(
+        host: impl Into<String>,
+        node_id: NodeId,
+        id_key: String,
+        port: Option<u16>,
+    ) -> 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<SessionStats> {
+        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<nym_node_requests::api::Client, NodeScraperError> {
+        // 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<NymNodeDetails> 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<NymNodeDescription> 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::<Vec<_>>();
+
+    let mixnet_durations = stat
+        .sessions
+        .iter()
+        .filter(|s| SessionType::from_string(&s.typ) == SessionType::Mixnet)
+        .map(|s| s.duration_ms)
+        .collect::<Vec<_>>();
+
+    let unkown_durations = stat
+        .sessions
+        .iter()
+        .filter(|s| SessionType::from_string(&s.typ) == SessionType::Unknown)
+        .map(|s| s.duration_ms)
+        .collect::<Vec<_>>();
+
+    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,
+    }
+}

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<Mutex<Vec<ScraperNodeInfo>>>,
-    packet_queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
 }
 
-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<Mutex<Vec<ScraperNodeInfo>>>,
-    ) -> 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<Mutex<Vec<ScraperNodeInfo>>>) {
         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<Mutex<Vec<ScraperNodeInfo>>>) {
-        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
     }
 }

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(())

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<NodeId, MetricsScrapingData> = 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<u16>,
-}
-
-impl MetricsScrapingData {
-    pub fn new(
-        host: impl Into<String>,
-        node_id: NodeId,
-        id_key: String,
-        port: Option<u16>,
-    ) -> 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<SessionStats> {
-        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<nym_node_requests::api::Client, NodeScraperError> {
-        // 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<NymNodeDetails> 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<NymNodeDescription> 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::<Vec<_>>();
-
-    let mixnet_durations = stat
-        .sessions
-        .iter()
-        .filter(|s| SessionType::from_string(&s.typ) == SessionType::Mixnet)
-        .map(|s| s.duration_ms)
-        .collect::<Vec<_>>();
-
-    let unkown_durations = stat
-        .sessions
-        .iter()
-        .filter(|s| SessionType::from_string(&s.typ) == SessionType::Unknown)
-        .map(|s| s.duration_ms)
-        .collect::<Vec<_>>();
-
-    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;

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<Mutex<Vec<ScraperNodeInfo>>>,
+}
+
+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<Mutex<Vec<ScraperNodeInfo>>>,
+    ) -> 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<Mutex<Vec<ScraperNodeInfo>>>) {
+        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
+    }
+}