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

[Product Data] Introduce data persistence on gateways (#5022)

Browse Source 
* add stats storage to gateways

* config fix

* add stats storage model and logic

* adapt stats collection to new storage

* stats cleanup on start

* change to linux only code

* tweaks

* modified stats cleanup + change session started

* change wrong table name

* store crashed session as 0 duration

* adapt for sqlx 0.7

* remove unused dependencies

* revert changes from gateway config, as it is broken anyway

* copyright and misc stuff

---------

Co-authored-by: Simon Wicky <simon@linode2-2.net>
This commit is contained in:
Simon Wicky
2024-10-28 09:25:37 +01:00
committed by GitHub
parent e65bfaeb31
commit ab11508235
23 changed files with 2277 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+14
View File
@@ -5258,6 +5258,7 @@ dependencies = [
"nym-credentials-interface",
"nym-crypto",
"nym-gateway-requests",
"nym-gateway-stats-storage",
"nym-gateway-storage",
"nym-ip-packet-router",
"nym-mixnet-client",
@@ -5353,6 +5354,19 @@ dependencies = [
"zeroize",
]
[[package]]
name = "nym-gateway-stats-storage"
version = "0.1.0"
dependencies = [
"nym-credentials-interface",
"nym-sphinx",
"sqlx",
"thiserror",
"time",
"tokio",
"tracing",
]
[[package]]
name = "nym-gateway-storage"
version = "0.1.0"
Cargo.toml
+1
View File
@@ -54,6 +54,7 @@ members = [
"common/exit-policy",
"common/gateway-requests",
"common/gateway-storage",
"common/gateway-stats-storage",
"common/http-api-client",
"common/http-api-common",
"common/inclusion-probability",
common/gateway-stats-storage/Cargo.toml
+34
View File
@@ -0,0 +1,34 @@
[package]
name = "nym-gateway-stats-storage"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"sqlite",
"macros",
"migrate",
"time",
] }
time = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
nym-sphinx = { path = "../nymsphinx" }
nym-credentials-interface = { path = "../credentials-interface" }
[build-dependencies]
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"sqlite",
"macros",
"migrate",
] }
common/gateway-stats-storage/build.rs
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use sqlx::{Connection, SqliteConnection};
use std::env;
#[tokio::main]
async fn main() {
let out_dir = env::var("OUT_DIR").unwrap();
let database_path = format!("{}/gateway-stats-example.sqlite", out_dir);
let mut conn = SqliteConnection::connect(&format!("sqlite://{}?mode=rwc", database_path))
.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);
}
common/gateway-stats-storage/migrations/20241017120000_create_initial_tables.sql
+26
View File
@@ -0,0 +1,26 @@
/*
* Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: GPL-3.0-only
*/
CREATE TABLE sessions_active
(
client_address TEXT NOT NULL PRIMARY KEY UNIQUE,
start_time TIMESTAMP WITHOUT TIME ZONE NOT NULL,
typ TEXT NOT NULL
);
CREATE TABLE sessions_finished
(
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
day DATE NOT NULL,
duration_ms INTEGER NOT NULL,
typ TEXT NOT NULL
);
CREATE TABLE sessions_unique_users
(
day DATE NOT NULL,
client_address TEXT NOT NULL,
PRIMARY KEY (day, client_address)
);
common/gateway-stats-storage/src/error.rs
+13
View File
@@ -0,0 +1,13 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use thiserror::Error;
#[derive(Error, Debug)]
pub enum StatsStorageError {
#[error("Database experienced an internal error: {0}")]
InternalDatabaseError(#[from] sqlx::Error),
#[error("Failed to perform database migration: {0}")]
MigrationError(#[from] sqlx::migrate::MigrateError),
}
common/gateway-stats-storage/src/lib.rs
+195
View File
@@ -0,0 +1,195 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use error::StatsStorageError;
use models::{ActiveSession, FinishedSession, SessionType, StoredFinishedSession};
use nym_sphinx::DestinationAddressBytes;
use sessions::SessionManager;
use sqlx::ConnectOptions;
use std::path::Path;
use time::Date;
use tracing::{debug, error};
pub mod error;
pub mod models;
mod sessions;
// note that clone here is fine as upon cloning the same underlying pool will be used
#[derive(Clone)]
pub struct PersistentStatsStorage {
session_manager: SessionManager,
}
impl PersistentStatsStorage {
/// Initialises `PersistentStatsStorage` using the provided path.
///
/// # Arguments
///
/// * `database_path`: path to the database.
pub async fn init<P: AsRef<Path> + Send>(database_path: P) -> Result<Self, StatsStorageError> {
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()
.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(PersistentStatsStorage {
session_manager: sessions::SessionManager::new(connection_pool),
})
}
//Sessions fn
pub async fn insert_finished_session(
&self,
date: Date,
session: FinishedSession,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.insert_finished_session(
date,
session.duration.whole_milliseconds() as i64,
session.typ.to_string().into(),
)
.await?)
}
pub async fn get_finished_sessions(
&self,
date: Date,
) -> Result<Vec<StoredFinishedSession>, StatsStorageError> {
Ok(self.session_manager.get_finished_sessions(date).await?)
}
pub async fn delete_finished_sessions(
&self,
before_date: Date,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.delete_finished_sessions(before_date)
.await?)
}
pub async fn insert_unique_user(
&self,
date: Date,
client_address_bs58: String,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.insert_unique_user(date, client_address_bs58)
.await?)
}
pub async fn get_unique_users_count(&self, date: Date) -> Result<i32, StatsStorageError> {
Ok(self.session_manager.get_unique_users_count(date).await?)
}
pub async fn delete_unique_users(&self, before_date: Date) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.delete_unique_users(before_date)
.await?)
}
pub async fn insert_active_session(
&self,
client_address: DestinationAddressBytes,
session: ActiveSession,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.insert_active_session(
client_address.as_base58_string(),
session.start,
session.typ.to_string().into(),
)
.await?)
}
pub async fn update_active_session_type(
&self,
client_address: DestinationAddressBytes,
session_type: SessionType,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.update_active_session_type(
client_address.as_base58_string(),
session_type.to_string().into(),
)
.await?)
}
pub async fn get_active_session(
&self,
client_address: DestinationAddressBytes,
) -> Result<Option<ActiveSession>, StatsStorageError> {
Ok(self
.session_manager
.get_active_session(client_address.as_base58_string())
.await?
.map(Into::into))
}
pub async fn get_all_active_sessions(&self) -> Result<Vec<ActiveSession>, StatsStorageError> {
Ok(self
.session_manager
.get_all_active_sessions()
.await?
.into_iter()
.map(Into::into)
.collect())
}
pub async fn get_started_sessions_count(
&self,
start_date: Date,
) -> Result<i32, StatsStorageError> {
Ok(self
.session_manager
.get_started_sessions_count(start_date)
.await?)
}
pub async fn get_active_users(&self) -> Result<Vec<String>, StatsStorageError> {
Ok(self.session_manager.get_active_users().await?)
}
pub async fn delete_active_session(
&self,
client_address: DestinationAddressBytes,
) -> Result<(), StatsStorageError> {
Ok(self
.session_manager
.delete_active_session(client_address.as_base58_string())
.await?)
}
pub async fn cleanup_active_sessions(&self) -> Result<(), StatsStorageError> {
Ok(self.session_manager.cleanup_active_sessions().await?)
}
}
common/gateway-stats-storage/src/models.rs
+109
View File
@@ -0,0 +1,109 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_credentials_interface::TicketType;
use sqlx::prelude::FromRow;
use time::{Duration, OffsetDateTime};
#[derive(FromRow)]
pub struct StoredFinishedSession {
duration_ms: i64,
typ: String,
}
impl StoredFinishedSession {
pub fn serialize(&self) -> (u64, String) {
(
self.duration_ms as u64, //we are sure that it fits in a u64, see `fn end_at`
self.typ.clone(),
)
}
}
pub struct FinishedSession {
pub duration: Duration,
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,
typ: String,
}
pub struct ActiveSession {
pub start: OffsetDateTime,
pub typ: SessionType,
}
impl ActiveSession {
pub fn new(start_time: OffsetDateTime) -> Self {
ActiveSession {
start: start_time,
typ: SessionType::Unknown,
}
}
pub fn set_type(&mut self, ticket_type: TicketType) {
self.typ = ticket_type.into();
}
pub fn end_at(self, stop_time: OffsetDateTime) -> Option<FinishedSession> {
let session_duration = stop_time - self.start;
//ensure duration is positive to fit in a u64
//u64::max milliseconds is 500k millenia so no overflow issue
if session_duration > Duration::ZERO {
Some(FinishedSession {
duration: session_duration,
typ: self.typ,
})
} else {
None
}
}
}
impl From<StoredActiveSession> for ActiveSession {
fn from(value: StoredActiveSession) -> Self {
ActiveSession {
start: value.start_time,
typ: SessionType::from_string(&value.typ),
}
}
}
common/gateway-stats-storage/src/sessions.rs
+177
View File
@@ -0,0 +1,177 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use time::{Date, OffsetDateTime};
use crate::models::{StoredActiveSession, StoredFinishedSession};
pub(crate) type Result<T> = std::result::Result<T, sqlx::Error>;
#[derive(Clone)]
pub(crate) struct SessionManager {
connection_pool: sqlx::SqlitePool,
}
impl SessionManager {
/// Creates new instance of the `SessionsManager` with the provided sqlite connection pool.
///
/// # Arguments
///
/// * `connection_pool`: database connection pool to use.
pub(crate) fn new(connection_pool: sqlx::SqlitePool) -> Self {
SessionManager { connection_pool }
}
pub(crate) async fn insert_finished_session(
&self,
date: Date,
duration_ms: i64,
typ: String,
) -> Result<()> {
sqlx::query!(
"INSERT INTO sessions_finished (day, duration_ms, typ) VALUES (?, ?, ?)",
date,
duration_ms,
typ
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn get_finished_sessions(
&self,
date: Date,
) -> Result<Vec<StoredFinishedSession>> {
sqlx::query_as("SELECT duration_ms, typ FROM sessions_finished WHERE day = ?")
.bind(date)
.fetch_all(&self.connection_pool)
.await
}
pub(crate) async fn delete_finished_sessions(&self, before_date: Date) -> Result<()> {
sqlx::query!("DELETE FROM sessions_finished WHERE day <= ? ", before_date)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn insert_unique_user(
&self,
date: Date,
client_address_b58: String,
) -> Result<()> {
sqlx::query!(
"INSERT OR IGNORE INTO sessions_unique_users (day, client_address) VALUES (?, ?)",
date,
client_address_b58,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn get_unique_users_count(&self, date: Date) -> Result<i32> {
Ok(sqlx::query!(
"SELECT COUNT(*) as count FROM sessions_unique_users WHERE day = ?",
date
)
.fetch_one(&self.connection_pool)
.await?
.count)
}
pub(crate) async fn delete_unique_users(&self, before_date: Date) -> Result<()> {
sqlx::query!(
"DELETE FROM sessions_unique_users WHERE day <= ? ",
before_date
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn insert_active_session(
&self,
client_address_b58: String,
start_time: OffsetDateTime,
typ: String,
) -> Result<()> {
sqlx::query!(
"INSERT INTO sessions_active (client_address, start_time, typ) VALUES (?, ?, ?)",
client_address_b58,
start_time,
typ
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn update_active_session_type(
&self,
client_address_b58: String,
typ: String,
) -> Result<()> {
sqlx::query!(
"UPDATE sessions_active SET typ = ? WHERE client_address = ?",
typ,
client_address_b58,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn get_active_session(
&self,
client_address_b58: String,
) -> Result<Option<StoredActiveSession>> {
sqlx::query_as("SELECT start_time, typ FROM sessions_active WHERE client_address = ?")
.bind(client_address_b58)
.fetch_optional(&self.connection_pool)
.await
}
pub(crate) async fn get_all_active_sessions(&self) -> Result<Vec<StoredActiveSession>> {
sqlx::query_as("SELECT start_time, typ FROM sessions_active")
.fetch_all(&self.connection_pool)
.await
}
pub(crate) async fn get_started_sessions_count(&self, start_date: Date) -> Result<i32> {
Ok(sqlx::query!(
"SELECT COUNT(*) as count FROM sessions_active WHERE date(start_time) = ?",
start_date
)
.fetch_one(&self.connection_pool)
.await?
.count)
}
pub(crate) async fn get_active_users(&self) -> Result<Vec<String>> {
Ok(sqlx::query!("SELECT client_address from sessions_active")
.fetch_all(&self.connection_pool)
.await?
.into_iter()
.map(|record| record.client_address)
.collect())
}
pub(crate) async fn delete_active_session(&self, client_address_b58: String) -> Result<()> {
sqlx::query!(
"DELETE FROM sessions_active WHERE client_address = ?",
client_address_b58
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn cleanup_active_sessions(&self) -> Result<()> {
sqlx::query!("DELETE FROM sessions_active")
.execute(&self.connection_pool)
.await?;
Ok(())
}
}
gateway/Cargo.toml
+1
View File
@@ -69,6 +69,7 @@ nym-credentials-interface = { path = "../common/credentials-interface" }
nym-credential-verification = { path = "../common/credential-verification" }
nym-crypto = { path = "../common/crypto" }
nym-gateway-storage = { path = "../common/gateway-storage" }
nym-gateway-stats-storage = { path = "../common/gateway-stats-storage" }
nym-gateway-requests = { path = "../common/gateway-requests" }
nym-mixnet-client = { path = "../common/client-libs/mixnet-client" }
nym-mixnode-common = { path = "../common/mixnode-common" }
gateway/src/config/persistence/paths.rs
+8 -1
View File
@@ -12,6 +12,7 @@ pub const DEFAULT_PRIVATE_SPHINX_KEY_FILENAME: &str = "private_sphinx.pem";
pub const DEFAULT_PUBLIC_SPHINX_KEY_FILENAME: &str = "public_sphinx.pem";
pub const DEFAULT_CLIENTS_STORAGE_FILENAME: &str = "db.sqlite";
pub const DEFAULT_STATS_STORAGE_FILENAME: &str = "stats.sqlite";
pub const DEFAULT_NETWORK_REQUESTER_CONFIG_FILENAME: &str = "network_requester_config.toml";
pub const DEFAULT_NETWORK_REQUESTER_DATA_DIR: &str = "network-requester-data";
@@ -39,6 +40,9 @@ pub struct GatewayPaths {
#[serde(alias = "persistent_storage")]
pub clients_storage: PathBuf,
/// Path to sqlite database containing all persistent stats data.
pub stats_storage: PathBuf,
/// Path to the configuration of the embedded network requester.
#[serde(deserialize_with = "de_maybe_stringified")]
pub network_requester_config: Option<PathBuf>,
@@ -54,7 +58,9 @@ impl GatewayPaths {
pub fn new_default<P: AsRef<Path>>(id: P) -> Self {
GatewayPaths {
keys: KeysPaths::new_default(id.as_ref()),
clients_storage: default_data_directory(id).join(DEFAULT_CLIENTS_STORAGE_FILENAME),
clients_storage: default_data_directory(id.as_ref())
.join(DEFAULT_CLIENTS_STORAGE_FILENAME),
stats_storage: default_data_directory(id).join(DEFAULT_STATS_STORAGE_FILENAME),
// node_description: default_config_filepath(id).join(DEFAULT_DESCRIPTION_FILENAME),
network_requester_config: None,
ip_packet_router_config: None,
@@ -70,6 +76,7 @@ impl GatewayPaths {
public_sphinx_key_file: Default::default(),
},
clients_storage: Default::default(),
stats_storage: Default::default(),
network_requester_config: None,
ip_packet_router_config: None,
}
gateway/src/error.rs
+7
View File
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: GPL-3.0-only
use nym_authenticator::error::AuthenticatorError;
use nym_gateway_stats_storage::error::StatsStorageError;
use nym_gateway_storage::error::StorageError;
use nym_ip_packet_router::error::IpPacketRouterError;
use nym_network_requester::error::{ClientCoreError, NetworkRequesterError};
@@ -115,6 +116,12 @@ pub enum GatewayError {
source: StorageError,
},
#[error("stats storage failure: {source}")]
StatsStorageError {
#[from]
source: StatsStorageError,
},
#[error("Path to network requester configuration file hasn't been specified. Perhaps try to run `setup-network-requester`?")]
UnspecifiedNetworkRequesterConfig,
gateway/src/node/helpers.rs
+9
View File
@@ -5,6 +5,7 @@ use crate::config::Config;
use crate::error::GatewayError;
use nym_crypto::asymmetric::encryption;
use nym_gateway_stats_storage::PersistentStatsStorage;
use nym_gateway_storage::PersistentStorage;
use nym_pemstore::traits::PemStorableKeyPair;
use nym_pemstore::KeyPairPath;
@@ -74,6 +75,14 @@ pub(crate) async fn initialise_main_storage(
Ok(PersistentStorage::init(path, retrieval_limit).await?)
}
pub(crate) async fn initialise_stats_storage(
config: &Config,
) -> Result<PersistentStatsStorage, GatewayError> {
let path = &config.storage_paths.stats_storage;
Ok(PersistentStatsStorage::init(path).await?)
}
pub fn load_keypair<T: PemStorableKeyPair>(
paths: KeyPairPath,
name: impl Into<String>,
gateway/src/node/mod.rs
+25 -14
View File
@@ -12,7 +12,9 @@ use crate::http::HttpApiBuilder;
use crate::node::client_handling::active_clients::ActiveClientsStore;
use crate::node::client_handling::embedded_clients::{LocalEmbeddedClientHandle, MessageRouter};
use crate::node::client_handling::websocket;
use crate::node::helpers::{initialise_main_storage, load_network_requester_config};
use crate::node::helpers::{
initialise_main_storage, initialise_stats_storage, load_network_requester_config,
};
use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler;
use futures::channel::{mpsc, oneshot};
use nym_credential_verification::ecash::{
@@ -41,6 +43,7 @@ pub(crate) mod helpers;
pub(crate) mod mixnet_handling;
pub(crate) mod statistics;
pub use nym_gateway_stats_storage::PersistentStatsStorage;
pub use nym_gateway_storage::{PersistentStorage, Storage};
// TODO: should this struct live here?
@@ -96,6 +99,8 @@ pub async fn create_gateway(
let storage = initialise_main_storage(&config).await?;
let stats_storage = initialise_stats_storage(&config).await?;
let nr_opts = network_requester_config.map(|config| LocalNetworkRequesterOpts {
config: config.clone(),
custom_mixnet_path: custom_mixnet.clone(),
@@ -106,7 +111,7 @@ pub async fn create_gateway(
custom_mixnet_path: custom_mixnet.clone(),
});
Gateway::new(config, nr_opts, ip_opts, storage)
Gateway::new(config, nr_opts, ip_opts, storage, stats_storage)
}
#[derive(Debug, Clone)]
@@ -147,7 +152,9 @@ pub struct Gateway<St = PersistentStorage> {
/// x25519 keypair used for Diffie-Hellman. Currently only used for sphinx key derivation.
sphinx_keypair: Arc<encryption::KeyPair>,
storage: St,
client_storage: St,
stats_storage: PersistentStatsStorage,
wireguard_data: Option<nym_wireguard::WireguardData>,
@@ -163,10 +170,12 @@ impl<St> Gateway<St> {
config: Config,
network_requester_opts: Option<LocalNetworkRequesterOpts>,
ip_packet_router_opts: Option<LocalIpPacketRouterOpts>,
storage: St,
client_storage: St,
stats_storage: PersistentStatsStorage,
) -> Result<Self, GatewayError> {
Ok(Gateway {
storage,
client_storage,
stats_storage,
identity_keypair: Arc::new(load_identity_keys(&config)?),
sphinx_keypair: Arc::new(helpers::load_sphinx_keys(&config)?),
config,
@@ -179,7 +188,7 @@ impl<St> Gateway<St> {
task_client: None,
})
}
#[allow(clippy::too_many_arguments)]
pub fn new_loaded(
config: Config,
network_requester_opts: Option<LocalNetworkRequesterOpts>,
@@ -187,7 +196,8 @@ impl<St> Gateway<St> {
authenticator_opts: Option<LocalAuthenticatorOpts>,
identity_keypair: Arc<identity::KeyPair>,
sphinx_keypair: Arc<encryption::KeyPair>,
storage: St,
client_storage: St,
stats_storage: PersistentStatsStorage,
) -> Self {
Gateway {
config,
@@ -196,7 +206,8 @@ impl<St> Gateway<St> {
authenticator_opts,
identity_keypair,
sphinx_keypair,
storage,
client_storage,
stats_storage,
wireguard_data: None,
session_stats: None,
run_http_server: true,
@@ -240,7 +251,7 @@ impl<St> Gateway<St> {
let connection_handler = ConnectionHandler::new(
packet_processor,
self.storage.clone(),
self.client_storage.clone(),
ack_sender,
active_clients_store,
);
@@ -275,7 +286,7 @@ impl<St> Gateway<St> {
forwarding_channel,
router_tx,
);
let all_peers = self.storage.get_all_wireguard_peers().await?;
let all_peers = self.client_storage.get_all_wireguard_peers().await?;
let used_private_network_ips = all_peers
.iter()
.cloned()
@@ -330,7 +341,7 @@ impl<St> Gateway<St> {
.start_with_shutdown(router_shutdown);
let wg_api = nym_wireguard::start_wireguard(
self.storage.clone(),
self.client_storage.clone(),
all_peers,
shutdown,
wireguard_data,
@@ -377,7 +388,7 @@ impl<St> Gateway<St> {
let shared_state = websocket::CommonHandlerState {
ecash_verifier,
storage: self.storage.clone(),
storage: self.client_storage.clone(),
local_identity: Arc::clone(&self.identity_keypair),
only_coconut_credentials: self.config.gateway.only_coconut_credentials,
bandwidth_cfg: (&self.config).into(),
@@ -415,7 +426,7 @@ impl<St> Gateway<St> {
info!("Starting gateway stats collector...");
let (mut stats_collector, stats_event_sender) =
GatewayStatisticsCollector::new(shared_session_stats);
GatewayStatisticsCollector::new(shared_session_stats, self.stats_storage.clone());
tokio::spawn(async move { stats_collector.run(shutdown).await });
stats_event_sender
}
@@ -654,7 +665,7 @@ impl<St> Gateway<St> {
nyxd_client,
self.identity_keypair.public_key().to_bytes(),
shutdown.fork("EcashVerifier"),
self.storage.clone(),
self.client_storage.clone(),
)
.await?,
);
gateway/src/node/statistics/mod.rs
+25 -4
View File
@@ -2,13 +2,14 @@
// SPDX-License-Identifier: GPL-3.0-only
use futures::{channel::mpsc, StreamExt};
use nym_gateway_stats_storage::PersistentStatsStorage;
use nym_node_http_api::state::metrics::SharedSessionStats;
use nym_statistics_common::events::{StatsEvent, StatsEventReceiver, StatsEventSender};
use nym_task::TaskClient;
use sessions::SessionStatsHandler;
use std::time::Duration;
use time::OffsetDateTime;
use tracing::trace;
use tracing::{error, trace, warn};
pub mod sessions;
@@ -23,21 +24,38 @@ pub(crate) struct GatewayStatisticsCollector {
impl GatewayStatisticsCollector {
pub fn new(
shared_session_stats: SharedSessionStats,
stats_storage: PersistentStatsStorage,
) -> (GatewayStatisticsCollector, StatsEventSender) {
let (stats_event_tx, stats_event_rx) = mpsc::unbounded();
let session_stats = SessionStatsHandler::new(shared_session_stats, stats_storage);
let collector = GatewayStatisticsCollector {
stats_event_rx,
session_stats: SessionStatsHandler::new(shared_session_stats),
session_stats,
};
(collector, stats_event_tx)
}
async fn update_shared_state(&mut self, update_time: OffsetDateTime) {
self.session_stats.update_shared_state(update_time).await;
if let Err(e) = self
.session_stats
.maybe_update_shared_state(update_time)
.await
{
error!("Failed to update session stats - {e}");
}
//here goes additionnal stats handler update
}
async fn on_start(&mut self) {
if let Err(e) = self.session_stats.on_start().await {
error!("Failed to cleanup session stats handler - {e}");
}
//here goes additionnal stats handler start cleanup
}
pub async fn run(&mut self, mut shutdown: TaskClient) {
self.on_start().await;
let mut update_interval = tokio::time::interval(STATISTICS_UPDATE_TIMER_INTERVAL);
while !shutdown.is_shutdown() {
tokio::select! {
@@ -53,7 +71,10 @@ impl GatewayStatisticsCollector {
Some(stat_event) = self.stats_event_rx.next() => {
//dispatching event to proper handler
match stat_event {
StatsEvent::SessionStatsEvent(event) => self.session_stats.handle_event(event),
StatsEvent::SessionStatsEvent(event) => {
if let Err(e) = self.session_stats.handle_event(event).await{
warn!("Session event handling error - {e}");
}},
}
},
gateway/src/node/statistics/sessions.rs
+113 -131
View File
@@ -2,176 +2,158 @@
// SPDX-License-Identifier: GPL-3.0-only
use nym_credentials_interface::TicketType;
use nym_gateway_stats_storage::models::FinishedSession;
use nym_gateway_stats_storage::PersistentStatsStorage;
use nym_gateway_stats_storage::{error::StatsStorageError, models::ActiveSession};
use nym_node_http_api::state::metrics::SharedSessionStats;
use nym_sphinx::DestinationAddressBytes;
use std::collections::{HashMap, HashSet};
use time::{Date, Duration, OffsetDateTime};
use nym_statistics_common::events::SessionEvent;
const FINISHED_SESSIONS_CAP: usize = 1_000_000; //to be on the safe side of memory blowups until persistent storage
#[derive(PartialEq)]
enum SessionType {
Vpn,
Mixnet,
Unknown,
}
impl SessionType {
fn to_string(&self) -> &str {
match self {
Self::Vpn => "vpn",
Self::Mixnet => "mixnet",
Self::Unknown => "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,
}
}
}
struct FinishedSession {
duration: Duration,
typ: SessionType,
}
impl FinishedSession {
fn serialize(&self) -> (u64, String) {
(
self.duration.whole_milliseconds() as u64, //we are sure that it fits in a u64, see `fn end_at`
self.typ.to_string().into(),
)
}
}
struct ActiveSession {
start: OffsetDateTime,
typ: SessionType,
}
impl ActiveSession {
fn new(start_time: OffsetDateTime) -> Self {
ActiveSession {
start: start_time,
typ: SessionType::Unknown,
}
}
fn set_type(&mut self, ticket_type: TicketType) {
self.typ = ticket_type.into();
}
fn end_at(self, stop_time: OffsetDateTime) -> Option<FinishedSession> {
let session_duration = stop_time - self.start;
//ensure duration is positive to fit in a u64
//u64::max milliseconds is 500k millenia so no overflow issue
if session_duration > Duration::ZERO {
Some(FinishedSession {
duration: session_duration,
typ: self.typ,
})
} else {
None
}
}
}
pub(crate) struct SessionStatsHandler {
last_update_day: Date,
storage: PersistentStatsStorage,
current_day: Date,
shared_session_stats: SharedSessionStats,
active_sessions: HashMap<DestinationAddressBytes, ActiveSession>,
unique_users: HashSet<DestinationAddressBytes>,
sessions_started: u32,
finished_sessions: Vec<FinishedSession>,
}
impl SessionStatsHandler {
pub fn new(shared_session_stats: SharedSessionStats) -> Self {
pub fn new(shared_session_stats: SharedSessionStats, storage: PersistentStatsStorage) -> Self {
SessionStatsHandler {
last_update_day: OffsetDateTime::now_utc().date(),
storage,
current_day: OffsetDateTime::now_utc().date(),
shared_session_stats,
active_sessions: Default::default(),
unique_users: Default::default(),
sessions_started: 0,
finished_sessions: Default::default(),
}
}
pub(crate) fn handle_event(&mut self, event: SessionEvent) {
pub(crate) async fn handle_event(
&mut self,
event: SessionEvent,
) -> Result<(), StatsStorageError> {
match event {
SessionEvent::SessionStart { start_time, client } => {
self.handle_session_start(start_time, client);
self.handle_session_start(start_time, client).await
}
SessionEvent::SessionStop { stop_time, client } => {
self.handle_session_stop(stop_time, client);
self.handle_session_stop(stop_time, client).await
}
SessionEvent::EcashTicket {
ticket_type,
client,
} => self.handle_ecash_ticket(ticket_type, client),
} => self.handle_ecash_ticket(ticket_type, client).await,
}
}
fn handle_session_start(
async fn handle_session_start(
&mut self,
start_time: OffsetDateTime,
client: DestinationAddressBytes,
) {
self.sessions_started += 1;
self.unique_users.insert(client);
self.active_sessions
.insert(client, ActiveSession::new(start_time));
}
fn handle_session_stop(&mut self, stop_time: OffsetDateTime, client: DestinationAddressBytes) {
if let Some(session) = self.active_sessions.remove(&client) {
if let Some(finished_session) = session.end_at(stop_time) {
if self.finished_sessions.len() < FINISHED_SESSIONS_CAP {
self.finished_sessions.push(finished_session);
}
}
}
) -> Result<(), StatsStorageError> {
self.storage
.insert_unique_user(self.current_day, client.as_base58_string())
.await?;
self.storage
.insert_active_session(client, ActiveSession::new(start_time))
.await?;
Ok(())
}
fn handle_ecash_ticket(&mut self, ticket_type: TicketType, client: DestinationAddressBytes) {
if let Some(active_session) = self.active_sessions.get_mut(&client) {
if active_session.typ == SessionType::Unknown {
active_session.set_type(ticket_type);
async fn handle_session_stop(
&mut self,
stop_time: OffsetDateTime,
client: DestinationAddressBytes,
) -> Result<(), StatsStorageError> {
if let Some(session) = self.storage.get_active_session(client).await? {
if let Some(finished_session) = session.end_at(stop_time) {
self.storage
.insert_finished_session(self.current_day, finished_session)
.await?;
self.storage.delete_active_session(client).await?;
}
}
Ok(())
}
async fn handle_ecash_ticket(
&mut self,
ticket_type: TicketType,
client: DestinationAddressBytes,
) -> Result<(), StatsStorageError> {
self.storage
.update_active_session_type(client, ticket_type.into())
.await?;
Ok(())
}
pub(crate) async fn on_start(&mut self) -> Result<(), StatsStorageError> {
let yesterday = OffsetDateTime::now_utc().date() - Duration::DAY;
//publish yesterday's data if any
self.publish_stats(yesterday).await?;
//store "active" sessions as duration 0
for active_session in self.storage.get_all_active_sessions().await? {
self.storage
.insert_finished_session(
self.current_day,
FinishedSession {
duration: Duration::ZERO,
typ: active_session.typ,
},
)
.await?
}
//cleanup active sessions
self.storage.cleanup_active_sessions().await?;
//delete old entries
self.delete_old_stats(yesterday - Duration::DAY).await?;
Ok(())
}
//update shared state once a day has passed, with data from the previous day
pub(crate) async fn update_shared_state(&mut self, update_time: OffsetDateTime) {
let update_date = update_time.date();
if update_date != self.last_update_day {
{
let mut shared_state = self.shared_session_stats.write().await;
shared_state.update_time = self.last_update_day;
shared_state.unique_active_users = self.unique_users.len() as u32;
shared_state.session_started = self.sessions_started;
shared_state.sessions = self
.finished_sessions
.iter()
.map(|s| s.serialize())
.collect();
}
self.reset_stats(update_date);
async fn publish_stats(&mut self, stats_date: Date) -> Result<(), StatsStorageError> {
let finished_sessions = self.storage.get_finished_sessions(stats_date).await?;
let user_count = self.storage.get_unique_users_count(stats_date).await?;
let session_started = self.storage.get_started_sessions_count(stats_date).await? as u32;
{
let mut shared_state = self.shared_session_stats.write().await;
shared_state.update_time = stats_date;
shared_state.unique_active_users = user_count as u32;
shared_state.session_started = session_started;
shared_state.sessions = finished_sessions.iter().map(|s| s.serialize()).collect();
}
Ok(())
}
pub(crate) async fn maybe_update_shared_state(
&mut self,
update_time: OffsetDateTime,
) -> Result<(), StatsStorageError> {
let update_date = update_time.date();
if update_date != self.current_day {
self.publish_stats(self.current_day).await?;
self.delete_old_stats(self.current_day - Duration::DAY)
.await?;
self.reset_stats(update_date).await?;
self.current_day = update_date;
}
Ok(())
}
fn reset_stats(&mut self, reset_day: Date) {
self.last_update_day = reset_day;
self.unique_users = self.active_sessions.keys().copied().collect();
self.finished_sessions = Default::default();
self.sessions_started = 0;
async fn reset_stats(&mut self, reset_day: Date) -> Result<(), StatsStorageError> {
//active users reset
let new_active_users = self.storage.get_active_users().await?;
for user in new_active_users {
self.storage.insert_unique_user(reset_day, user).await?;
}
Ok(())
}
async fn delete_old_stats(&mut self, delete_before: Date) -> Result<(), StatsStorageError> {
self.storage.delete_finished_sessions(delete_before).await?;
self.storage.delete_unique_users(delete_before).await?;
Ok(())
}
}
nym-node/src/config/old_configs/mod.rs
+2
View File
@@ -4,7 +4,9 @@
mod old_config_v1;
mod old_config_v2;
mod old_config_v3;
mod old_config_v4;
pub use old_config_v1::try_upgrade_config_v1;
pub use old_config_v2::try_upgrade_config_v2;
pub use old_config_v3::try_upgrade_config_v3;
pub use old_config_v4::try_upgrade_config_v4;
nym-node/src/config/old_configs/old_config_v3.rs
+35 -57
View File
@@ -4,13 +4,7 @@
#![allow(dead_code)]
use crate::{config::*, error::KeyIOFailure};
use entry_gateway::Debug as EntryGatewayConfigDebug;
use exit_gateway::{IpPacketRouter, IpPacketRouterDebug, NetworkRequester, NetworkRequesterDebug};
use mixnode::{Verloc, VerlocDebug};
use nym_client_core_config_types::{
disk_persistence::{ClientKeysPaths, CommonClientPaths},
DebugConfig as ClientDebugConfig,
};
use nym_client_core_config_types::DebugConfig as ClientDebugConfig;
use nym_config::serde_helpers::de_maybe_port;
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_network_requester::{
@@ -19,6 +13,7 @@ use nym_network_requester::{
};
use nym_pemstore::{store_key, store_keypair};
use nym_sphinx_acknowledgements::AckKey;
use old_configs::old_config_v4::*;
use persistence::*;
use rand::rngs::OsRng;
use serde::{Deserialize, Serialize};
@@ -90,12 +85,12 @@ pub enum NodeModeV3 {
ExitGateway,
}
impl From<NodeModeV3> for NodeMode {
impl From<NodeModeV3> for NodeModeV4 {
fn from(config: NodeModeV3) -> Self {
match config {
NodeModeV3::Mixnode => NodeMode::Mixnode,
NodeModeV3::EntryGateway => NodeMode::EntryGateway,
NodeModeV3::ExitGateway => NodeMode::ExitGateway,
NodeModeV3::Mixnode => NodeModeV4::Mixnode,
NodeModeV3::EntryGateway => NodeModeV4::EntryGateway,
NodeModeV3::ExitGateway => NodeModeV4::ExitGateway,
}
}
}
@@ -601,23 +596,6 @@ impl AuthenticatorPathsV3 {
}
}
pub fn to_common_client_paths(&self) -> CommonClientPaths {
CommonClientPaths {
keys: ClientKeysPaths {
private_identity_key_file: self.private_ed25519_identity_key_file.clone(),
public_identity_key_file: self.public_ed25519_identity_key_file.clone(),
private_encryption_key_file: self.private_x25519_diffie_hellman_key_file.clone(),
public_encryption_key_file: self.public_x25519_diffie_hellman_key_file.clone(),
ack_key_file: self.ack_key_file.clone(),
},
gateway_registrations: self.gateway_registrations.clone(),
// not needed for embedded providers
credentials_database: Default::default(),
reply_surb_database: self.reply_surb_database.clone(),
}
}
pub fn ed25519_identity_storage_paths(&self) -> nym_pemstore::KeyPairPath {
nym_pemstore::KeyPairPath::new(
&self.private_ed25519_identity_key_file,
@@ -963,7 +941,7 @@ pub async fn initialise(
pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
path: P,
prev_config: Option<ConfigV3>,
) -> Result<Config, NymNodeError> {
) -> Result<ConfigV4, NymNodeError> {
tracing::debug!("Updating from 1.1.4");
let old_cfg = if let Some(prev_config) = prev_config {
prev_config
@@ -981,21 +959,21 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
.ok_or(NymNodeError::DataDirDerivationFailure)?,
);
let cfg = Config {
let cfg = ConfigV4 {
save_path: old_cfg.save_path,
id: old_cfg.id,
mode: old_cfg.mode.into(),
host: Host {
host: HostV4 {
public_ips: old_cfg.host.public_ips,
hostname: old_cfg.host.hostname,
location: old_cfg.host.location,
},
mixnet: Mixnet {
mixnet: MixnetV4 {
bind_address: old_cfg.mixnet.bind_address,
announce_port: None,
nym_api_urls: old_cfg.mixnet.nym_api_urls,
nyxd_urls: old_cfg.mixnet.nyxd_urls,
debug: MixnetDebug {
debug: MixnetDebugV4 {
packet_forwarding_initial_backoff: old_cfg
.mixnet
.debug
@@ -1009,8 +987,8 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
unsafe_disable_noise: old_cfg.mixnet.debug.unsafe_disable_noise,
},
},
storage_paths: NymNodePaths {
keys: KeysPaths {
storage_paths: NymNodePathsV4 {
keys: KeysPathsV4 {
private_ed25519_identity_key_file: old_cfg
.storage_paths
.keys
@@ -1038,7 +1016,7 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
},
description: old_cfg.storage_paths.description,
},
http: Http {
http: HttpV4 {
bind_address: old_cfg.http.bind_address,
landing_page_assets_path: old_cfg.http.landing_page_assets_path,
access_token: old_cfg.http.access_token,
@@ -1046,13 +1024,13 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
expose_system_hardware: old_cfg.http.expose_system_hardware,
expose_crypto_hardware: old_cfg.http.expose_crypto_hardware,
},
wireguard: Wireguard {
wireguard: WireguardV4 {
enabled: old_cfg.wireguard.enabled,
bind_address: old_cfg.wireguard.bind_address,
private_ip: old_cfg.wireguard.private_ip,
announced_port: old_cfg.wireguard.announced_port,
private_network_prefix: old_cfg.wireguard.private_network_prefix,
storage_paths: WireguardPaths {
storage_paths: WireguardPathsV4 {
private_diffie_hellman_key_file: old_cfg
.wireguard
.storage_paths
@@ -1063,12 +1041,12 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
.public_diffie_hellman_key_file,
},
},
mixnode: MixnodeConfig {
storage_paths: MixnodePaths {},
verloc: Verloc {
mixnode: MixnodeConfigV4 {
storage_paths: MixnodePathsV4 {},
verloc: VerlocV4 {
bind_address: old_cfg.mixnode.verloc.bind_address,
announce_port: None,
debug: VerlocDebug {
debug: VerlocDebugV4 {
packets_per_node: old_cfg.mixnode.verloc.debug.packets_per_node,
connection_timeout: old_cfg.mixnode.verloc.debug.connection_timeout,
packet_timeout: old_cfg.mixnode.verloc.debug.packet_timeout,
@@ -1078,16 +1056,16 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
retry_timeout: old_cfg.mixnode.verloc.debug.retry_timeout,
},
},
debug: mixnode::Debug {
debug: DebugV4 {
node_stats_logging_delay: old_cfg.mixnode.debug.node_stats_logging_delay,
node_stats_updating_delay: old_cfg.mixnode.debug.node_stats_updating_delay,
},
},
entry_gateway: EntryGatewayConfig {
storage_paths: EntryGatewayPaths {
entry_gateway: EntryGatewayConfigV4 {
storage_paths: EntryGatewayPathsV4 {
clients_storage: old_cfg.entry_gateway.storage_paths.clients_storage,
cosmos_mnemonic: old_cfg.entry_gateway.storage_paths.cosmos_mnemonic,
authenticator: AuthenticatorPaths {
authenticator: AuthenticatorPathsV4 {
private_ed25519_identity_key_file: old_cfg
.entry_gateway
.storage_paths
@@ -1129,16 +1107,16 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
bind_address: old_cfg.entry_gateway.bind_address,
announce_ws_port: old_cfg.entry_gateway.announce_ws_port,
announce_wss_port: old_cfg.entry_gateway.announce_wss_port,
debug: EntryGatewayConfigDebug {
debug: EntryGatewayConfigDebugV4 {
message_retrieval_limit: old_cfg.entry_gateway.debug.message_retrieval_limit,
// \/ ADDED
zk_nym_tickets: Default::default(),
},
},
exit_gateway: ExitGatewayConfig {
storage_paths: ExitGatewayPaths {
exit_gateway: ExitGatewayConfigV4 {
storage_paths: ExitGatewayPathsV4 {
clients_storage: exit_gateway_paths.clients_storage,
network_requester: NetworkRequesterPaths {
network_requester: NetworkRequesterPathsV4 {
private_ed25519_identity_key_file: old_cfg
.exit_gateway
.storage_paths
@@ -1175,7 +1153,7 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
.network_requester
.gateway_registrations,
},
ip_packet_router: IpPacketRouterPaths {
ip_packet_router: IpPacketRouterPathsV4 {
private_ed25519_identity_key_file: old_cfg
.exit_gateway
.storage_paths
@@ -1212,7 +1190,7 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
.ip_packet_router
.gateway_registrations,
},
authenticator: AuthenticatorPaths {
authenticator: AuthenticatorPathsV4 {
private_ed25519_identity_key_file: old_cfg
.exit_gateway
.storage_paths
@@ -1252,8 +1230,8 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
},
open_proxy: old_cfg.exit_gateway.open_proxy,
upstream_exit_policy_url: old_cfg.exit_gateway.upstream_exit_policy_url,
network_requester: NetworkRequester {
debug: NetworkRequesterDebug {
network_requester: NetworkRequesterV4 {
debug: NetworkRequesterDebugV4 {
enabled: old_cfg.exit_gateway.network_requester.debug.enabled,
disable_poisson_rate: old_cfg
.exit_gateway
@@ -1263,8 +1241,8 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
client_debug: old_cfg.exit_gateway.network_requester.debug.client_debug,
},
},
ip_packet_router: IpPacketRouter {
debug: IpPacketRouterDebug {
ip_packet_router: IpPacketRouterV4 {
debug: IpPacketRouterDebugV4 {
enabled: old_cfg.exit_gateway.ip_packet_router.debug.enabled,
disable_poisson_rate: old_cfg
.exit_gateway
@@ -1277,7 +1255,7 @@ pub async fn try_upgrade_config_v3<P: AsRef<Path>>(
debug: Default::default(),
},
authenticator: Default::default(),
logging: LoggingSettings {},
logging: LoggingSettingsV4 {},
};
Ok(cfg)
nym-node/src/config/old_configs/old_config_v4.rs
+1422
View File
File diff suppressed because it is too large Load Diff
nym-node/src/config/persistence.rs
+9
View File
@@ -24,6 +24,7 @@ pub const DEFAULT_NYMNODE_DESCRIPTION_FILENAME: &str = "description.toml";
// Entry Gateway:
pub const DEFAULT_CLIENTS_STORAGE_FILENAME: &str = "clients.sqlite";
pub const DEFAULT_STATS_STORAGE_FILENAME: &str = "stats.sqlite";
pub const DEFAULT_MNEMONIC_FILENAME: &str = "cosmos_mnemonic";
// Exit Gateway:
@@ -147,6 +148,9 @@ pub struct EntryGatewayPaths {
/// derived shared keys, available client bandwidths and wireguard peers.
pub clients_storage: PathBuf,
/// Path to sqlite database containing all persistent stats data.
pub stats_storage: PathBuf,
/// Path to file containing cosmos account mnemonic used for zk-nym redemption.
pub cosmos_mnemonic: PathBuf,
@@ -157,6 +161,7 @@ impl EntryGatewayPaths {
pub fn new<P: AsRef<Path>>(data_dir: P) -> Self {
EntryGatewayPaths {
clients_storage: data_dir.as_ref().join(DEFAULT_CLIENTS_STORAGE_FILENAME),
stats_storage: data_dir.as_ref().join(DEFAULT_STATS_STORAGE_FILENAME),
cosmos_mnemonic: data_dir.as_ref().join(DEFAULT_MNEMONIC_FILENAME),
authenticator: AuthenticatorPaths::new(data_dir),
}
@@ -207,6 +212,9 @@ pub struct ExitGatewayPaths {
/// derived shared keys, available client bandwidths and wireguard peers.
pub clients_storage: PathBuf,
/// Path to sqlite database containing all persistent stats data.
pub stats_storage: PathBuf,
pub network_requester: NetworkRequesterPaths,
pub ip_packet_router: IpPacketRouterPaths,
@@ -459,6 +467,7 @@ impl ExitGatewayPaths {
let data_dir = data_dir.as_ref();
ExitGatewayPaths {
clients_storage: data_dir.join(DEFAULT_CLIENTS_STORAGE_FILENAME),
stats_storage: data_dir.join(DEFAULT_STATS_STORAGE_FILENAME),
network_requester: NetworkRequesterPaths::new(data_dir),
ip_packet_router: IpPacketRouterPaths::new(data_dir),
authenticator: AuthenticatorPaths::new(data_dir),
nym-node/src/config/template.rs
+7
View File
@@ -185,6 +185,9 @@ announce_wss_port = {{#if entry_gateway.announce_wss_port }} {{ entry_gateway.an
# derived shared keys, available client bandwidths and wireguard peers.
clients_storage = '{{ entry_gateway.storage_paths.clients_storage }}'
# Path to sqlite database containing all persistent stats data.
stats_storage = '{{ entry_gateway.storage_paths.stats_storage }}'
# Path to file containing cosmos account mnemonic used for zk-nym redemption.
cosmos_mnemonic = '{{ entry_gateway.storage_paths.cosmos_mnemonic }}'
@@ -237,6 +240,10 @@ upstream_exit_policy_url = '{{ exit_gateway.upstream_exit_policy_url }}'
# derived shared keys, available client bandwidths and wireguard peers.
clients_storage = '{{ exit_gateway.storage_paths.clients_storage }}'
# Path to sqlite database containing all persistent stats data.
stats_storage = '{{ exit_gateway.storage_paths.stats_storage }}'
[exit_gateway.storage_paths.network_requester]
# Path to file containing network requester ed25519 identity private key.
private_ed25519_identity_key_file = '{{ exit_gateway.storage_paths.network_requester.private_ed25519_identity_key_file }}'
nym-node/src/config/upgrade_helpers.rs
+2 -1
View File
@@ -10,7 +10,8 @@ use std::path::Path;
async fn try_upgrade_config(path: &Path) -> Result<(), NymNodeError> {
let cfg = try_upgrade_config_v1(path, None).await.ok();
let cfg = try_upgrade_config_v2(path, cfg).await.ok();
match try_upgrade_config_v3(path, cfg).await {
let cfg = try_upgrade_config_v3(path, cfg).await.ok();
match try_upgrade_config_v4(path, cfg).await {
Ok(cfg) => cfg.save(),
Err(e) => {
tracing::error!("Failed to finish upgrade - {e}");
nym-node/src/node/mod.rs
+15
View File
@@ -67,6 +67,7 @@ impl MixnodeData {
pub struct EntryGatewayData {
mnemonic: Zeroizing<bip39::Mnemonic>,
client_storage: nym_gateway::node::PersistentStorage,
stats_storage: nym_gateway::node::PersistentStatsStorage,
sessions_stats: SharedSessionStats,
}
@@ -94,6 +95,11 @@ impl EntryGatewayData {
)
.await
.map_err(nym_gateway::GatewayError::from)?,
stats_storage: nym_gateway::node::PersistentStatsStorage::init(
&config.storage_paths.stats_storage,
)
.await
.map_err(nym_gateway::GatewayError::from)?,
sessions_stats: SharedSessionStats::new(),
})
}
@@ -114,6 +120,7 @@ pub struct ExitGatewayData {
auth_x25519: x25519::PublicKey,
client_storage: nym_gateway::node::PersistentStorage,
stats_storage: nym_gateway::node::PersistentStatsStorage,
}
impl ExitGatewayData {
@@ -262,6 +269,11 @@ impl ExitGatewayData {
.await
.map_err(nym_gateway::GatewayError::from)?;
let stats_storage =
nym_gateway::node::PersistentStatsStorage::init(&config.storage_paths.stats_storage)
.await
.map_err(nym_gateway::GatewayError::from)?;
Ok(ExitGatewayData {
nr_ed25519,
nr_x25519,
@@ -270,6 +282,7 @@ impl ExitGatewayData {
auth_ed25519,
auth_x25519,
client_storage,
stats_storage,
})
}
}
@@ -580,6 +593,7 @@ impl NymNode {
self.ed25519_identity_keys.clone(),
self.x25519_sphinx_keys.clone(),
self.entry_gateway.client_storage.clone(),
self.entry_gateway.stats_storage.clone(),
);
entry_gateway.disable_http_server();
entry_gateway.set_task_client(task_client);
@@ -610,6 +624,7 @@ impl NymNode {
self.ed25519_identity_keys.clone(),
self.x25519_sphinx_keys.clone(),
self.exit_gateway.client_storage.clone(),
self.exit_gateway.stats_storage.clone(),
);
exit_gateway.disable_http_server();
exit_gateway.set_task_client(task_client);