Batch SQL writes for packet stats

nym-node-status-api/nym-node-status-agent/run.sh

@@ -3,7 +3,7 @@
 set -eu
 export ENVIRONMENT=${ENVIRONMENT:-"mainnet"}
 
-probe_git_ref="nym-vpn-core-v1.10.0"
+probe_git_ref="nym-vpn-core-v1.4.0"
 
 crate_root=$(dirname $(realpath "$0"))
 monorepo_root=$(realpath "${crate_root}/../..")

nym-node-status-api/nym-node-status-api/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-node-status-api"
-version = "3.1.1"
+version = "3.1.2"
 authors.workspace = true
 repository.workspace = true
 homepage.workspace = true

nym-node-status-api/nym-node-status-api/src/cli/mod.rs

@@ -73,6 +73,13 @@ pub(crate) struct Cli {
     #[arg(value_delimiter = ',')]
     pub(crate) agent_key_list: Vec<String>,
 
+    #[clap(
+        long,
+        default_value_t = 10,
+        env = "NYM_NODE_STATUS_API_PACKET_STATS_MAX_CONCURRENT_TASKS"
+    )]
+    pub(crate) packet_stats_max_concurrent_tasks: usize,
+
     /// https://github.com/ipinfo/rust
     #[clap(long, env = "IPINFO_API_TOKEN")]
     pub(crate) ipinfo_api_token: String,

nym-node-status-api/nym-node-status-api/src/db/models.rs

@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
 use std::str::FromStr;
 use strum_macros::{EnumString, FromRepr};
-use time::{Date, OffsetDateTime};
+use time::{Date, OffsetDateTime, UtcDateTime};
 use utoipa::ToSchema;
 
 macro_rules! serialize_opt_to_value {
@@ -362,7 +362,7 @@ impl TryFrom<GatewaySessionsRecord> for http::models::SessionStats {
     }
 }
 
-#[derive(strum_macros::Display)]
+#[derive(strum_macros::Display, Clone)]
 pub(crate) enum ScrapeNodeKind {
     LegacyMixnode { mix_id: i64 },
     MixingNymNode { node_id: i64 },
@@ -520,3 +520,10 @@ pub struct NodeStats {
     pub packets_sent: i64,
     pub packets_dropped: i64,
 }
+
+pub struct InsertStatsRecord {
+    pub node_kind: ScrapeNodeKind,
+    pub timestamp_utc: UtcDateTime,
+    pub unix_timestamp: i64,
+    pub stats: NodeStats,
+}

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/mod.rs

@@ -19,7 +19,7 @@ pub(crate) use nym_nodes::{
     get_described_node_bond_info, get_node_self_description, update_nym_nodes,
 };
 pub(crate) use packet_stats::{
-    get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats,
+    batch_store_packet_stats, get_raw_node_stats, insert_daily_node_stats_uncommitted,
 };
 pub(crate) use scraper::{get_nodes_for_scraping, insert_scraped_node_description};
 pub(crate) use summary::{get_summary, get_summary_history};

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/packet_stats.rs

@@ -1,17 +1,70 @@
-use crate::db::{
-    models::{NodeStats, ScrapeNodeKind, ScraperNodeInfo},
-    DbPool,
+use crate::{
+    db::{
+        models::{InsertStatsRecord, NodeStats, ScrapeNodeKind},
+        DbPool,
+    },
+    node_scraper::helpers::update_daily_stats_uncommitted,
+    utils::now_utc,
 };
 use anyhow::Result;
+use sqlx::Transaction;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::{info, instrument};
 
-pub(crate) async fn insert_node_packet_stats(
+#[instrument(level = "info", skip_all)]
+pub(crate) async fn batch_store_packet_stats(
     pool: &DbPool,
+    results: Arc<Mutex<Vec<InsertStatsRecord>>>,
+) -> anyhow::Result<()> {
+    let results_iter = results.lock().await;
+    info!(
+        "📊 ⏳ Storing {} packet stats into the DB",
+        results_iter.len()
+    );
+    let started_at = now_utc();
+
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|err| anyhow::anyhow!("Failed to begin transaction: {err}"))?;
+
+    for stats_record in &(*results_iter) {
+        insert_node_packet_stats_uncommitted(
+            &mut tx,
+            &stats_record.node_kind,
+            &stats_record.stats,
+            stats_record.unix_timestamp,
+        )
+        .await?;
+
+        update_daily_stats_uncommitted(
+            &mut tx,
+            &stats_record.node_kind,
+            stats_record.timestamp_utc,
+            &stats_record.stats,
+        )
+        .await?;
+    }
+
+    tx.commit()
+        .await
+        .inspect(|_| {
+            let elapsed = now_utc() - started_at;
+            info!(
+                "📊 ☑️ Packet stats successfully committed to DB (took {}s)",
+                elapsed.as_seconds_f32()
+            );
+        })
+        .map_err(|err| anyhow::anyhow!("Failed to commit: {err}"))
+}
+
+async fn insert_node_packet_stats_uncommitted(
+    tx: &mut Transaction<'static, sqlx::Sqlite>,
     node_kind: &ScrapeNodeKind,
     stats: &NodeStats,
     timestamp_utc: i64,
 ) -> Result<()> {
-    let mut conn = pool.acquire().await?;
-
     match node_kind {
         ScrapeNodeKind::LegacyMixnode { mix_id } => {
             sqlx::query!(
@@ -26,7 +79,7 @@ pub(crate) async fn insert_node_packet_stats(
                 stats.packets_sent,
                 stats.packets_dropped,
             )
-            .execute(&mut *conn)
+            .execute(tx.as_mut())
             .await?;
         }
         ScrapeNodeKind::MixingNymNode { node_id }
@@ -43,7 +96,7 @@ pub(crate) async fn insert_node_packet_stats(
                 stats.packets_sent,
                 stats.packets_dropped,
             )
-            .execute(&mut *conn)
+            .execute(tx.as_mut())
             .await?;
         }
     }
@@ -52,12 +105,10 @@ pub(crate) async fn insert_node_packet_stats(
 }
 
 pub(crate) async fn get_raw_node_stats(
-    pool: &DbPool,
-    node: &ScraperNodeInfo,
+    tx: &mut Transaction<'static, sqlx::Sqlite>,
+    node_kind: &ScrapeNodeKind,
 ) -> Result<Option<NodeStats>> {
-    let mut conn = pool.acquire().await?;
-
-    let packets = match node.node_kind {
+    let packets = match node_kind {
         // if no packets are found, it's fine to assume 0 because that's also
         // SQL default value if none provided
         ScrapeNodeKind::LegacyMixnode { mix_id } => {
@@ -75,7 +126,7 @@ pub(crate) async fn get_raw_node_stats(
                 "#,
                 mix_id
             )
-            .fetch_optional(&mut *conn)
+            .fetch_optional(tx.as_mut())
             .await?
         }
         ScrapeNodeKind::MixingNymNode { node_id }
@@ -94,23 +145,21 @@ pub(crate) async fn get_raw_node_stats(
                 "#,
                 node_id
             )
-            .fetch_optional(&mut *conn)
+            .fetch_optional(tx.as_mut())
             .await?
         }
     };
 
     Ok(packets)
 }
 
-pub(crate) async fn insert_daily_node_stats(
-    pool: &DbPool,
-    node: &ScraperNodeInfo,
+pub(crate) async fn insert_daily_node_stats_uncommitted(
+    tx: &mut Transaction<'static, sqlx::Sqlite>,
+    node_kind: &ScrapeNodeKind,
     date_utc: &str,
     packets: NodeStats,
 ) -> Result<()> {
-    let mut conn = pool.acquire().await?;
-
-    match node.node_kind {
+    match node_kind {
         ScrapeNodeKind::LegacyMixnode { mix_id } => {
             let total_stake = sqlx::query_scalar!(
                 r#"
@@ -121,7 +170,7 @@ pub(crate) async fn insert_daily_node_stats(
                    "#,
                 mix_id
             )
-            .fetch_one(&mut *conn)
+            .fetch_one(tx.as_mut())
             .await?;
 
             sqlx::query!(
@@ -144,7 +193,7 @@ pub(crate) async fn insert_daily_node_stats(
                 packets.packets_sent,
                 packets.packets_dropped,
             )
-            .execute(&mut *conn)
+            .execute(tx.as_mut())
             .await?;
         }
         ScrapeNodeKind::MixingNymNode { node_id }
@@ -158,7 +207,7 @@ pub(crate) async fn insert_daily_node_stats(
                 "#,
                 node_id
             )
-            .fetch_one(&mut *conn)
+            .fetch_one(tx.as_mut())
             .await?;
 
             sqlx::query!(
@@ -181,7 +230,7 @@ pub(crate) async fn insert_daily_node_stats(
                 packets.packets_sent,
                 packets.packets_dropped,
             )
-            .execute(&mut *conn)
+            .execute(tx.as_mut())
             .await?;
         }
     }

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/http/api/testruns.rs

@@ -160,11 +160,11 @@ async fn submit_testrun(
         .map(unix_timestamp_to_utc_rfc3339)
         .unwrap_or_else(|| String::from("never"));
     tracing::info!(
-        "✅ Testrun row_id {} for gateway {} complete (last assigned {}, created at {})",
+        gateway_id = gw_identity,
+        last_assigned = last_assigned,
+        created_at = created_at,
+        "✅ Testrun row_id {} for gateway complete",
         assigned_testrun.id,
-        gw_identity,
-        last_assigned,
-        created_at
     );
 
     Ok(StatusCode::CREATED)

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,14 @@ 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(),
+        args.packet_stats_max_concurrent_tasks,
+    );
     tokio::spawn(async move {
         scraper.start().await;
     });
@@ -74,7 +81,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");
     });