GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add storage to stats collector

Browse Source
This commit is contained in:
Simon Wicky
2025-01-28 11:32:32 +01:00
parent 73a036a3b9
commit 932c2cbe43
20 changed files with 285 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -15
View File
@@ -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",
]
common/client-core/src/client/statistics_control.rs
+1
View File
@@ -88,6 +88,7 @@ impl StatisticsControl {
log::error!("Failed to report client stats: {:?}", err);
} else {
self.stats.reset();
log::debug!("Stats report successfully sent");
}
}
common/statistics/src/clients/connection.rs
+7 -7
View File
@@ -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
common/statistics/src/clients/gateway_conn_statistics.rs
+1 -1
View File
@@ -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,
common/statistics/src/clients/nym_api_statistics.rs
+1 -1
View File
@@ -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,
common/statistics/src/clients/packet_statistics.rs
+1 -1
View File
@@ -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,
common/statistics/src/report.rs
+13 -13
View File
@@ -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<ClientStatsReport> for Vec<u8> {
@@ -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<String>,
pub(crate) os_arch: String,
pub os_type: String,
pub os_version: Option<String>,
pub os_arch: String,
}
impl OsInformation {
service-providers/statistics-collector/Cargo.toml
+16 -15
View File
@@ -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",
] }
service-providers/statistics-collector/build.rs
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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);
}
service-providers/statistics-collector/migrations/20250109100000_init_tables.sql
+23
View File
@@ -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)
);
service-providers/statistics-collector/src/config/persistence.rs
+5 -1
View File
@@ -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),
}
}
}
service-providers/statistics-collector/src/config/template.rs
+3
View File
@@ -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 #####
service-providers/statistics-collector/src/error.rs
+3
View File
@@ -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<T> = std::result::Result<T, StatsCollectorError>;
service-providers/statistics-collector/src/lib.rs
+1 -3
View File
@@ -1,11 +1,9 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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;
service-providers/statistics-collector/src/mixnet_client.rs
+1
View File
@@ -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());
service-providers/statistics-collector/src/mixnet_listener.rs
+14 -3
View File
@@ -1,7 +1,7 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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(())
}
service-providers/statistics-collector/src/stats_collector.rs
+5 -2
View File
@@ -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.");
service-providers/statistics-collector/src/storage/client_stats_report.rs
+69
View File
@@ -0,0 +1,69 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_statistics_common::report::ClientStatsReport;
pub(crate) type Result<T> = std::result::Result<T, sqlx::Error>;
#[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(())
}
}
service-providers/statistics-collector/src/storage/error.rs
+13
View File
@@ -0,0 +1,13 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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),
}
service-providers/statistics-collector/src/storage/mod.rs
+79
View File
@@ -0,0 +1,79 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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<P: AsRef<Path> + Send>(
database_path: P,
) -> Result<Self, ClientStatsReportStorageError> {
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?)
}
}