diff --git a/Cargo.lock b/Cargo.lock index 90ab79f6c8..a4b7a8e16e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6822,41 +6822,27 @@ name = "nym-statistics-collector" version = "0.1.0" dependencies = [ "anyhow", - "bincode", "bs58", - "bytes", "clap", - "defguard_wireguard_rs", - "fastrand 2.1.1", "futures", - "ipnetwork 0.20.0", "log", "nym-bin-common", "nym-client-core", "nym-config", - "nym-credential-verification", - "nym-credentials-interface", "nym-crypto", - "nym-gateway-requests", - "nym-gateway-storage", "nym-id", "nym-network-defaults", "nym-sdk", - "nym-service-provider-requests-common", "nym-service-providers-common", "nym-sphinx", "nym-statistics-common", "nym-task", "nym-types", - "nym-wireguard", - "nym-wireguard-types", - "rand", "serde", "serde_json", + "sqlx", "thiserror 1.0.69", "tokio", - "tokio-stream", - "tokio-util", "url", ] diff --git a/common/client-core/src/client/statistics_control.rs b/common/client-core/src/client/statistics_control.rs index c4e6997cb9..fe19147844 100644 --- a/common/client-core/src/client/statistics_control.rs +++ b/common/client-core/src/client/statistics_control.rs @@ -88,6 +88,7 @@ impl StatisticsControl { log::error!("Failed to report client stats: {:?}", err); } else { self.stats.reset(); + log::debug!("Stats report successfully sent"); } } diff --git a/common/statistics/src/clients/connection.rs b/common/statistics/src/clients/connection.rs index b4e4d65518..b073e17a24 100644 --- a/common/statistics/src/clients/connection.rs +++ b/common/statistics/src/clients/connection.rs @@ -7,16 +7,16 @@ use nym_credentials_interface::TicketType; use serde::{Deserialize, Serialize}; #[derive(Default, Debug, Clone, Serialize, Deserialize)] -pub(crate) struct ConnectionStats { +pub struct ConnectionStats { //tickets - mixnet_entry_spent: u32, - vpn_entry_spent: u32, - mixnet_exit_spent: u32, - vpn_exit_spent: u32, + pub mixnet_entry_spent: u32, + pub vpn_entry_spent: u32, + pub mixnet_exit_spent: u32, + pub vpn_exit_spent: u32, //country_connection - wg_exit_country_code: String, - mix_exit_country_code: String, + pub wg_exit_country_code: String, + pub mix_exit_country_code: String, } /// Event space for Nym API statistics tracking diff --git a/common/statistics/src/clients/gateway_conn_statistics.rs b/common/statistics/src/clients/gateway_conn_statistics.rs index 961c7a0f7f..f949873575 100644 --- a/common/statistics/src/clients/gateway_conn_statistics.rs +++ b/common/statistics/src/clients/gateway_conn_statistics.rs @@ -12,7 +12,7 @@ use nym_metrics::{inc, inc_by}; use serde::{Deserialize, Serialize}; #[derive(Default, Debug, Clone, Serialize, Deserialize)] -pub(crate) struct GatewayStats { +pub struct GatewayStats { // Sent real_packets_sent: u64, real_packets_sent_size: usize, diff --git a/common/statistics/src/clients/nym_api_statistics.rs b/common/statistics/src/clients/nym_api_statistics.rs index 9c3ee609c0..25b2fd9176 100644 --- a/common/statistics/src/clients/nym_api_statistics.rs +++ b/common/statistics/src/clients/nym_api_statistics.rs @@ -12,7 +12,7 @@ use nym_metrics::{inc, inc_by}; use serde::{Deserialize, Serialize}; #[derive(Default, Debug, Clone, Serialize, Deserialize)] -pub(crate) struct NymApiStats { +pub struct NymApiStats { // Sent real_packets_sent: u64, real_packets_sent_size: usize, diff --git a/common/statistics/src/clients/packet_statistics.rs b/common/statistics/src/clients/packet_statistics.rs index 866215c751..fe2a07c624 100644 --- a/common/statistics/src/clients/packet_statistics.rs +++ b/common/statistics/src/clients/packet_statistics.rs @@ -61,7 +61,7 @@ impl Instant { } #[derive(Default, Debug, Clone, Serialize, Deserialize)] -pub(crate) struct PacketStatistics { +pub struct PacketStatistics { // Sent real_packets_sent: u64, real_packets_sent_size: usize, diff --git a/common/statistics/src/report.rs b/common/statistics/src/report.rs index f01cf2be06..231149cf7a 100644 --- a/common/statistics/src/report.rs +++ b/common/statistics/src/report.rs @@ -18,16 +18,16 @@ const VERSION: &str = "v1"; /// Report object containing both data to be reported and client / device context. We take extra care not to overcapture context information. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ClientStatsReport { - pub(crate) kind: String, - pub(crate) api_version: String, - pub(crate) last_update_time: OffsetDateTime, - pub(crate) client_id: String, - pub(crate) client_type: String, - pub(crate) os_information: OsInformation, - pub(crate) packet_stats: PacketStatistics, - pub(crate) gateway_conn_stats: GatewayStats, - pub(crate) nym_api_stats: NymApiStats, - pub(crate) connection_stats: ConnectionStats, + pub kind: String, + pub api_version: String, + pub last_update_time: OffsetDateTime, + pub client_id: String, + pub client_type: String, + pub os_information: OsInformation, + pub packet_stats: PacketStatistics, + pub gateway_conn_stats: GatewayStats, + pub nym_api_stats: NymApiStats, + pub connection_stats: ConnectionStats, } impl From for Vec { @@ -65,9 +65,9 @@ impl Default for ClientStatsReport { #[derive(Clone, Debug, Serialize, Deserialize)] pub struct OsInformation { - pub(crate) os_type: String, - pub(crate) os_version: Option, - pub(crate) os_arch: String, + pub os_type: String, + pub os_version: Option, + pub os_arch: String, } impl OsInformation { diff --git a/service-providers/statistics-collector/Cargo.toml b/service-providers/statistics-collector/Cargo.toml index b59d509170..5003edd7e9 100644 --- a/service-providers/statistics-collector/Cargo.toml +++ b/service-providers/statistics-collector/Cargo.toml @@ -10,43 +10,44 @@ license.workspace = true [dependencies] anyhow = { workspace = true } -bincode = { workspace = true } bs58 = { workspace = true } -bytes = { workspace = true } clap = { workspace = true, features = ["cargo", "derive"] } -defguard_wireguard_rs = { workspace = true } -fastrand = { workspace = true } futures = { workspace = true } -ipnetwork = { workspace = true } log = { workspace = true } -rand = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sqlx = { workspace = true, features = [ + "runtime-tokio-rustls", + "sqlite", + "time", +] } thiserror = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread", "net"] } -tokio-stream = { workspace = true } -tokio-util = { workspace = true, features = ["codec"] } url = { workspace = true } nym-bin-common = { path = "../../common/bin-common", features = [ "clap", "output_format", ] } + nym-client-core = { path = "../../common/client-core", features = ["cli"] } nym-config = { path = "../../common/config" } -nym-credentials-interface = { path = "../../common/credentials-interface" } -nym-credential-verification = { path = "../../common/credential-verification" } nym-crypto = { path = "../../common/crypto" } -nym-gateway-requests = { path = "../../common/gateway-requests" } -nym-gateway-storage = { path = "../../common/gateway-storage" } nym-id = { path = "../../common/nym-id" } nym-network-defaults = { path = "../../common/network-defaults" } nym-sdk = { path = "../../sdk/rust/nym-sdk" } nym-service-providers-common = { path = "../common" } -nym-service-provider-requests-common = { path = "../../common/service-provider-requests-common" } nym-statistics-common = { path = "../../common/statistics" } nym-sphinx = { path = "../../common/nymsphinx" } nym-task = { path = "../../common/task" } nym-types = { path = "../../common/types" } -nym-wireguard = { path = "../../common/wireguard" } -nym-wireguard-types = { path = "../../common/wireguard-types" } + + +[build-dependencies] +tokio = { workspace = true, features = ["macros"] } +sqlx = { workspace = true, features = [ + "runtime-tokio-rustls", + "sqlite", + "macros", + "migrate", +] } diff --git a/service-providers/statistics-collector/build.rs b/service-providers/statistics-collector/build.rs new file mode 100644 index 0000000000..a67d07cba8 --- /dev/null +++ b/service-providers/statistics-collector/build.rs @@ -0,0 +1,28 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +#[tokio::main] +async fn main() { + use sqlx::{Connection, SqliteConnection}; + use std::env; + + let out_dir = env::var("OUT_DIR").unwrap(); + let database_path = format!("{out_dir}/stats-collector-example.sqlite"); + + let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc")) + .await + .expect("Failed to create SQLx database connection"); + + sqlx::migrate!("./migrations") + .run(&mut conn) + .await + .expect("Failed to perform SQLx migrations"); + + #[cfg(target_family = "unix")] + println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path); + + #[cfg(target_family = "windows")] + // for some strange reason we need to add a leading `/` to the windows path even though it's + // not a valid windows path... but hey, it works... + println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path); +} diff --git a/service-providers/statistics-collector/migrations/20250109100000_init_tables.sql b/service-providers/statistics-collector/migrations/20250109100000_init_tables.sql new file mode 100644 index 0000000000..e4d8782a21 --- /dev/null +++ b/service-providers/statistics-collector/migrations/20250109100000_init_tables.sql @@ -0,0 +1,23 @@ +CREATE TABLE report ( + day DATE NOT NULL, + client_id TEXT NOT NULL, + client_type TEXT, + os_type TEXT, + os_version TEXT, + architecture TEXT, + PRIMARY KEY (client_id, day) +); + +CREATE TABLE connection_stats ( + received_at TIMESTAMP WITH TIME ZONE NOT NULL, + client_id TEXT NOT NULL, + mixnet_entry_spent INTEGER, + vpn_entry_spent INTEGER, + mixnet_exit_spent INTEGER, + vpn_exit_spent INTEGER, + wg_exit_country_code TEXT, + mix_exit_country_code TEXT, + PRIMARY KEY (client_id, received_at) +); + + diff --git a/service-providers/statistics-collector/src/config/persistence.rs b/service-providers/statistics-collector/src/config/persistence.rs index 6b77810518..18a166a392 100644 --- a/service-providers/statistics-collector/src/config/persistence.rs +++ b/service-providers/statistics-collector/src/config/persistence.rs @@ -3,12 +3,15 @@ use nym_client_core::config::disk_persistence::CommonClientPaths; use serde::{Deserialize, Serialize}; -use std::path::Path; +use std::path::{Path, PathBuf}; + +pub const DEFAULT_REPORT_DATABASE_PATH: &str = "report.db"; #[derive(Debug, Deserialize, PartialEq, Eq, Serialize, Clone)] pub struct StatsCollectorPaths { #[serde(flatten)] pub common_paths: CommonClientPaths, + pub client_reports_database: PathBuf, } impl StatsCollectorPaths { @@ -17,6 +20,7 @@ impl StatsCollectorPaths { Self { common_paths: CommonClientPaths::new_base(base_dir), + client_reports_database: base_dir.join(DEFAULT_REPORT_DATABASE_PATH), } } } diff --git a/service-providers/statistics-collector/src/config/template.rs b/service-providers/statistics-collector/src/config/template.rs index e5c2355387..55af321518 100644 --- a/service-providers/statistics-collector/src/config/template.rs +++ b/service-providers/statistics-collector/src/config/template.rs @@ -65,6 +65,9 @@ reply_surb_database = '{{ storage_paths.reply_surb_database }}' # i.e. details such as their public keys, owner addresses or the network information. gateway_registrations = '{{ storage_paths.gateway_registrations }}' +# Path to the client reports database +client_reports_database = '{{ storage_paths.client_reports_database }}' + ##### logging configuration options ##### diff --git a/service-providers/statistics-collector/src/error.rs b/service-providers/statistics-collector/src/error.rs index 2deb511e2d..15027b2e54 100644 --- a/service-providers/statistics-collector/src/error.rs +++ b/service-providers/statistics-collector/src/error.rs @@ -30,6 +30,9 @@ pub enum StatsCollectorError { #[error(transparent)] NymIdError(#[from] NymIdError), + + #[error("Storage error : {0}")] + ReportStorageError(#[from] crate::storage::error::ClientStatsReportStorageError), } pub type Result = std::result::Result; diff --git a/service-providers/statistics-collector/src/lib.rs b/service-providers/statistics-collector/src/lib.rs index 609fbc449b..15a6bdc0dd 100644 --- a/service-providers/statistics-collector/src/lib.rs +++ b/service-providers/statistics-collector/src/lib.rs @@ -1,11 +1,9 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -//pub use authenticator::{Authenticator, OnStartData}; -//pub use config::Config; - pub mod config; pub mod error; pub mod mixnet_client; pub mod mixnet_listener; pub mod stats_collector; +pub mod storage; diff --git a/service-providers/statistics-collector/src/mixnet_client.rs b/service-providers/statistics-collector/src/mixnet_client.rs index 7e2965efd4..c1c93572d4 100644 --- a/service-providers/statistics-collector/src/mixnet_client.rs +++ b/service-providers/statistics-collector/src/mixnet_client.rs @@ -26,6 +26,7 @@ pub async fn create_mixnet_client( .disable_main_poisson_packet_distribution = true; debug_config.cover_traffic.disable_loop_cover_traffic_stream = true; debug_config.stats_reporting.enabled = false; + debug_config.topology.ignore_egress_epoch_role = true; //necessary for a fixed address let storage_paths = nym_sdk::mixnet::StoragePaths::from(paths.clone()); diff --git a/service-providers/statistics-collector/src/mixnet_listener.rs b/service-providers/statistics-collector/src/mixnet_listener.rs index 4f2d664a90..66446f6ab0 100644 --- a/service-providers/statistics-collector/src/mixnet_listener.rs +++ b/service-providers/statistics-collector/src/mixnet_listener.rs @@ -1,7 +1,7 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::error::*; +use crate::{error::*, storage::ClientStatsStorage}; use futures::StreamExt; use nym_sphinx::receiver::ReconstructedMessage; use nym_statistics_common::report::ClientStatsReport; @@ -11,14 +11,22 @@ pub(crate) struct MixnetListener { // The mixnet client that we use to send and receive packets from the mixnet pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient, + // Report storage + pub(crate) client_report_storage: ClientStatsStorage, + // The task handle for the main loop pub(crate) task_handle: TaskHandle, } impl MixnetListener { - pub fn new(mixnet_client: nym_sdk::mixnet::MixnetClient, task_handle: TaskHandle) -> Self { + pub fn new( + mixnet_client: nym_sdk::mixnet::MixnetClient, + client_report_storage: ClientStatsStorage, + task_handle: TaskHandle, + ) -> Self { MixnetListener { mixnet_client, + client_report_storage, task_handle, } } @@ -27,12 +35,15 @@ impl MixnetListener { &mut self, reconstructed: ReconstructedMessage, ) -> Result<()> { + println!("Received message"); log::debug!( "Received message with sender_tag: {:?}", reconstructed.sender_tag ); let report = deserialize_stats_report(&reconstructed)?; - println!("report : {report:?}"); + println!("Received report"); + self.client_report_storage.store_report(report).await?; + println!("Stored report"); Ok(()) } diff --git a/service-providers/statistics-collector/src/stats_collector.rs b/service-providers/statistics-collector/src/stats_collector.rs index 8a723d5e90..8393bb3d0a 100644 --- a/service-providers/statistics-collector/src/stats_collector.rs +++ b/service-providers/statistics-collector/src/stats_collector.rs @@ -7,7 +7,7 @@ use nym_client_core::{HardcodedTopologyProvider, TopologyProvider}; use nym_sdk::{mixnet::Recipient, GatewayTransceiver}; use nym_task::{TaskClient, TaskHandle}; -use crate::{config::Config, error::StatsCollectorError}; +use crate::{config::Config, error::StatsCollectorError, storage::ClientStatsStorage}; pub struct OnStartData { // to add more fields as required @@ -102,8 +102,11 @@ impl StatisticsCollector { let self_address = *mixnet_client.nym_address(); + let report_storage = + ClientStatsStorage::init(self.config.storage_paths.client_reports_database).await?; + let mixnet_listener = - crate::mixnet_listener::MixnetListener::new(mixnet_client, task_handle); + crate::mixnet_listener::MixnetListener::new(mixnet_client, report_storage, task_handle); log::info!("The address of this client is: {self_address}"); log::info!("All systems go. Press CTRL-C to stop the server."); diff --git a/service-providers/statistics-collector/src/storage/client_stats_report.rs b/service-providers/statistics-collector/src/storage/client_stats_report.rs new file mode 100644 index 0000000000..48c1c7800a --- /dev/null +++ b/service-providers/statistics-collector/src/storage/client_stats_report.rs @@ -0,0 +1,69 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_statistics_common::report::ClientStatsReport; + +pub(crate) type Result = std::result::Result; + +#[derive(Clone)] +pub(crate) struct ClientStatsReportManager { + connection_pool: sqlx::SqlitePool, +} + +impl ClientStatsReportManager { + /// Creates new instance of the `ClientStatsReportManager` with the provided sqlite connection pool. + /// + /// # Arguments + /// + /// * `connection_pool`: database connection pool to use. + pub(crate) fn new(connection_pool: sqlx::SqlitePool) -> Self { + ClientStatsReportManager { connection_pool } + } + + pub async fn store_report(&mut self, report: ClientStatsReport) -> Result<()> { + self.store_base(&report).await?; + self.store_connection_stats(&report).await?; + Ok(()) + } + + async fn store_base(&self, report: &ClientStatsReport) -> Result<()> { + let report_day = report.last_update_time.date(); + sqlx::query!( + "INSERT OR IGNORE INTO report (day, client_id, client_type, os_type, os_version, architecture) VALUES (?, ?, ?, ?, ?, ?)", + report_day, + report.client_id, + report.client_type, + report.os_information.os_type, + report.os_information.os_version, + report.os_information.os_arch + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + async fn store_connection_stats(&self, report: &ClientStatsReport) -> Result<()> { + sqlx::query!( + "INSERT OR IGNORE INTO connection_stats ( + received_at, + client_id, + mixnet_entry_spent, + vpn_entry_spent, + mixnet_exit_spent, + vpn_exit_spent, + wg_exit_country_code, + mix_exit_country_code) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + report.last_update_time, + report.client_id, + report.connection_stats.mixnet_entry_spent, + report.connection_stats.vpn_entry_spent, + report.connection_stats.mixnet_exit_spent, + report.connection_stats.vpn_exit_spent, + report.connection_stats.wg_exit_country_code, + report.connection_stats.mix_exit_country_code + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } +} diff --git a/service-providers/statistics-collector/src/storage/error.rs b/service-providers/statistics-collector/src/storage/error.rs new file mode 100644 index 0000000000..d5e0b32eb4 --- /dev/null +++ b/service-providers/statistics-collector/src/storage/error.rs @@ -0,0 +1,13 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum ClientStatsReportStorageError { + #[error("Database experienced an internal error: {0}")] + InternalDatabaseError(#[from] sqlx::Error), + + #[error("Failed to perform database migration: {0}")] + MigrationError(#[from] sqlx::migrate::MigrateError), +} diff --git a/service-providers/statistics-collector/src/storage/mod.rs b/service-providers/statistics-collector/src/storage/mod.rs new file mode 100644 index 0000000000..9cd8bcab37 --- /dev/null +++ b/service-providers/statistics-collector/src/storage/mod.rs @@ -0,0 +1,79 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use error::ClientStatsReportStorageError; +use log::{debug, error}; +use nym_statistics_common::report::ClientStatsReport; +use sqlx::{ + sqlite::{SqliteAutoVacuum, SqliteSynchronous}, + ConnectOptions, +}; +use std::path::Path; + +pub mod error; +//pub mod models; +mod client_stats_report; + +// note that clone here is fine as upon cloning the same underlying pool will be used +#[derive(Clone)] +pub struct ClientStatsStorage { + client_stats_report_manager: client_stats_report::ClientStatsReportManager, +} + +impl ClientStatsStorage { + /// Initialises `ClientStatsStorage` using the provided path. + /// + /// # Arguments + /// + /// * `database_path`: path to the database. + pub async fn init + Send>( + database_path: P, + ) -> Result { + debug!( + "Attempting to connect to database {:?}", + database_path.as_ref().as_os_str() + ); + + // TODO: we can inject here more stuff based on our gateway global config + // struct. Maybe different pool size or timeout intervals? + let opts = sqlx::sqlite::SqliteConnectOptions::new() + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .filename(database_path) + .create_if_missing(true) + .disable_statement_logging(); + + // TODO: do we want auto_vacuum ? + + let connection_pool = match sqlx::SqlitePool::connect_with(opts).await { + Ok(db) => db, + Err(err) => { + error!("Failed to connect to SQLx database: {err}"); + return Err(err.into()); + } + }; + + if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await { + error!("Failed to perform migration on the SQLx database: {err}"); + return Err(err.into()); + } + + // the cloning here are cheap as connection pool is stored behind an Arc + Ok(ClientStatsStorage { + client_stats_report_manager: client_stats_report::ClientStatsReportManager::new( + connection_pool, + ), + }) + } + + pub(crate) async fn store_report( + &mut self, + report: ClientStatsReport, + ) -> Result<(), ClientStatsReportStorageError> { + Ok(self + .client_stats_report_manager + .store_report(report) + .await?) + } +}