NS API - Gateway stats scraping (#5180)

* squashed commit before rebasing * removed blank lines
2024-12-02 12:15:30 +01:00
parent a9e62889c3
commit 4851614375
22 changed files with 650 additions and 41 deletions
@@ -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",
@@ -22,6 +22,7 @@ tracing = { workspace = true }

 nym-sphinx = { path = "../nymsphinx" }
 nym-credentials-interface = { path = "../credentials-interface" }
+nym-statistics-common = { path = "../statistics" }


 [build-dependencies]
@@ -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;
@@ -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<TicketType> 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,
@@ -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<TicketType> 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,
+        }
+    }
+}
@@ -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 }
@@ -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);
+
@@ -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<String>,
    pub explorer_pretty_bond: Option<String>,
 }
+
+#[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<String>,
+    pub vpn_sessions: Option<String>,
+    pub mixnet_sessions: Option<String>,
+    pub unknown_sessions: Option<String>,
+}
+
+impl TryFrom<GatewaySessionsRecord> for http::models::SessionStats {
+    type Error = anyhow::Error;
+
+    fn try_from(value: GatewaySessionsRecord) -> Result<Self, Self::Error> {
+        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,
+        })
+    }
+}
@@ -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<GatewaySessionsRecord>,
+) -> 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<Vec<SessionStats>> {
+    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::<Vec<GatewaySessionsRecord>>()
+    .await?;
+
+    let items: Vec<SessionStats> = items
+        .into_iter()
+        .map(|item| item.try_into())
+        .collect::<anyhow::Result<Vec<_>>>()
+        .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(())
+}
@@ -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};
@@ -0,0 +1,10 @@
+use axum::Router;
+
+use crate::http::state::AppState;
+
+pub(crate) mod sessions;
+
+pub(crate) fn routes() -> Router<AppState> {
+    Router::new().nest("/sessions", sessions::routes())
+    //eventually add other metrics type
+}
@@ -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<AppState> {
+    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<usize>,
+    page: Option<usize>,
+    node_id: Option<String>,
+    day: Option<String>,
+}
+
+#[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<SessionQueryParams>,
+    State(state): State<AppState>,
+) -> HttpResult<Json<PagedResult<SessionStats>>> {
+    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::<u32>() {
+            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,
+    )))
+}
@@ -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",
@@ -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;

@@ -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<GatewaySkinny>,
    PagedMixnode = PagedResult<Mixnode>,
    PagedService = PagedResult<Service>,
+    PagedSessionStats = PagedResult<SessionStats>
 )]
 pub struct PagedResult<T> {
    pub page: usize,
@@ -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<serde_json::Value>,
+    pub vpn_sessions: Option<serde_json::Value>,
+    pub mixnet_sessions: Option<serde_json::Value>,
+    pub unknown_sessions: Option<serde_json::Value>,
+}
@@ -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<String, Arc<RwLock<Vec<Mixnode>>>>,
    mixstats: Cache<String, Arc<RwLock<Vec<DailyStats>>>>,
    history: Cache<String, Arc<RwLock<Vec<SummaryHistory>>>>,
+    session_stats: Cache<String, Arc<RwLock<Vec<SessionStats>>>>,
 }

 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<SessionStats> {
+        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<SessionStats>,
+    ) -> Entry<String, Arc<RwLock<Vec<SessionStats>>>> {
+        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
+    }
 }
@@ -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,
@@ -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 },
+}
@@ -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<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,
+        }
+    }
+
+    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::error!("[metrics scraper]: {e}");
+                        None
+                    }
+                }
+            }
+            Err(e) => {
+                tracing::error!("[metrics scraper]: {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")
+                    .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,
+    }
+}
@@ -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<ErrorResponse>;

@@ -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<SessionStats, NymNodeApiClientError> {
+        self.get_json_from(routes::api::v1::metrics::sessions_absolute())
+            .await
+    }
 }

 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -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<String>,

    pub sessions: Vec<Session>,