diff --git a/Cargo.lock b/Cargo.lock index 05e435eee6..2a09a35770 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4846,6 +4846,7 @@ dependencies = [ "nym-nonexhaustive-delayqueue", "nym-pemstore", "nym-sphinx", + "nym-statistics-common", "nym-task", "nym-topology", "nym-validator-client", @@ -6255,6 +6256,7 @@ dependencies = [ "nym-socks5-client-core", "nym-socks5-requests", "nym-sphinx", + "nym-statistics-common", "nym-task", "nym-topology", "nym-validator-client", @@ -6596,9 +6598,18 @@ name = "nym-statistics-common" version = "0.1.0" dependencies = [ "futures", + "log", "nym-credentials-interface", + "nym-metrics", "nym-sphinx", + "nym-task", + "serde", + "serde_json", + "si-scale", + "sysinfo", + "thiserror", "time", + "tokio", ] [[package]] diff --git a/common/client-core/Cargo.toml b/common/client-core/Cargo.toml index 9e2f16157c..754eb61440 100644 --- a/common/client-core/Cargo.toml +++ b/common/client-core/Cargo.toml @@ -43,6 +43,7 @@ nym-gateway-requests = { path = "../gateway-requests" } nym-metrics = { path = "../nym-metrics" } nym-nonexhaustive-delayqueue = { path = "../nonexhaustive-delayqueue" } nym-sphinx = { path = "../nymsphinx" } +nym-statistics-common = { path = "../statistics" } nym-pemstore = { path = "../pemstore" } nym-topology = { path = "../topology", features = ["serializable"] } nym-validator-client = { path = "../client-libs/validator-client", default-features = false } diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs index 646300d5aa..8d197439f6 100644 --- a/common/client-core/src/client/base_client/mod.rs +++ b/common/client-core/src/client/base_client/mod.rs @@ -1,8 +1,8 @@ // Copyright 2022-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use super::packet_statistics_control::PacketStatisticsReporter; use super::received_buffer::ReceivedBufferMessage; +use super::statistics_control::StatisticsControl; use super::topology_control::geo_aware_provider::GeoAwareTopologyProvider; use crate::client::base_client::storage::helpers::store_client_keys; use crate::client::base_client::storage::MixnetClientStorage; @@ -12,7 +12,6 @@ use crate::client::key_manager::persistence::KeyStore; use crate::client::key_manager::ClientKeys; use crate::client::mix_traffic::transceiver::{GatewayReceiver, GatewayTransceiver, RemoteGateway}; use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController}; -use crate::client::packet_statistics_control::PacketStatisticsControl; use crate::client::real_messages_control; use crate::client::real_messages_control::RealMessagesController; use crate::client::received_buffer::{ @@ -49,16 +48,20 @@ use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::addressing::nodes::NodeIdentity; use nym_sphinx::params::PacketType; use nym_sphinx::receiver::{ReconstructedMessage, SphinxMessageReceiver}; +use nym_statistics_common::clients::ClientStatsSender; +use nym_statistics_common::StatsReportingConfig; use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths}; use nym_task::{TaskClient, TaskHandle}; use nym_topology::provider_trait::TopologyProvider; use nym_topology::HardcodedTopologyProvider; use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, UserAgent}; use rand::rngs::OsRng; +use sha2::Digest; use std::fmt::Debug; use std::os::raw::c_int as RawFd; use std::path::Path; use std::sync::Arc; +use tokio::sync::mpsc::Sender; use url::Url; #[cfg(all( @@ -184,6 +187,7 @@ pub struct BaseClientBuilder<'a, C, S: MixnetClientStorage> { custom_gateway_transceiver: Option>, shutdown: Option, user_agent: Option, + stats_reporting_config: Option, setup_method: GatewaySetup, } @@ -207,6 +211,7 @@ where custom_gateway_transceiver: None, shutdown: None, user_agent: None, + stats_reporting_config: None, setup_method: GatewaySetup::MustLoad { gateway_id: None }, } } @@ -250,6 +255,12 @@ where self } + #[must_use] + pub fn with_statistics_reporting(mut self, config: StatsReportingConfig) -> Self { + self.stats_reporting_config = Some(config); + self + } + pub fn with_stored_topology>( mut self, file: P, @@ -273,7 +284,7 @@ where self_address: Recipient, topology_accessor: TopologyAccessor, mix_tx: BatchMixMessageSender, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, shutdown: TaskClient, ) { info!("Starting loop cover traffic stream..."); @@ -306,7 +317,7 @@ where client_connection_rx: ConnectionCommandReceiver, shutdown: TaskClient, packet_type: PacketType, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) { info!("Starting real traffic stream..."); @@ -335,7 +346,7 @@ where reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, shutdown: TaskClient, - packet_statistics_control: PacketStatisticsReporter, + metrics_reporter: ClientStatsSender, ) { info!("Starting received messages buffer controller..."); let controller: ReceivedMessagesBufferController = @@ -345,7 +356,7 @@ where mixnet_receiver, reply_key_storage, reply_controller_sender, - packet_statistics_control, + metrics_reporter, ); controller.start_with_shutdown(shutdown) } @@ -586,11 +597,19 @@ where Ok(()) } - fn start_packet_statistics_control(shutdown: TaskClient) -> PacketStatisticsReporter { - info!("Starting packet statistics control..."); - let (packet_statistics_control, packet_stats_reporter) = PacketStatisticsControl::new(); - packet_statistics_control.start_with_shutdown(shutdown); - packet_stats_reporter + fn start_statistics_control( + stats_reporting_config: Option, + client_stats_id: String, + input_sender: Sender, + shutdown: TaskClient, + ) -> ClientStatsSender { + info!("Starting statistics control..."); + StatisticsControl::create_and_start_with_shutdown( + stats_reporting_config, + client_stats_id, + input_sender.clone(), + shutdown.with_suffix("controller"), + ) } fn start_mix_traffic_controller( @@ -720,6 +739,17 @@ where self.user_agent.clone(), ); + let client_stats_id = format!( + "{:x}", + sha2::Sha256::digest(self_address.identity().to_bytes()) + ); + let stats_reporter = Self::start_statistics_control( + self.stats_reporting_config, + client_stats_id, + input_sender.clone(), + shutdown.fork("statistics_control"), + ); + // needs to be started as the first thing to block if required waiting for the gateway Self::start_topology_refresher( topology_provider, @@ -731,9 +761,6 @@ where ) .await?; - let packet_stats_reporter = - Self::start_packet_statistics_control(shutdown.fork("packet_statistics_control")); - let gateway_packet_router = PacketRouter::new( ack_sender, mixnet_messages_sender, @@ -765,7 +792,7 @@ where reply_storage.key_storage(), reply_controller_sender.clone(), shutdown.fork("received_messages_buffer"), - packet_stats_reporter.clone(), + stats_reporter.clone(), ); // The message_sender is the transmitter for any component generating sphinx packets @@ -804,7 +831,7 @@ where client_connection_rx, shutdown.fork("real_traffic_controller"), self.config.debug.traffic.packet_type, - packet_stats_reporter.clone(), + stats_reporter.clone(), ); if !self @@ -819,7 +846,7 @@ where self_address, shared_topology_accessor.clone(), message_sender, - packet_stats_reporter, + stats_reporter.clone(), shutdown.fork("cover_traffic_stream"), ); } @@ -847,6 +874,7 @@ where topology_accessor: shared_topology_accessor, gateway_connection: GatewayConnection { gateway_ws_fd }, }, + stats_reporter, task_handle: shutdown, }) } @@ -858,6 +886,7 @@ pub struct BaseClient { pub client_input: ClientInputStatus, pub client_output: ClientOutputStatus, pub client_state: ClientState, + pub stats_reporter: ClientStatsSender, pub task_handle: TaskHandle, } diff --git a/common/client-core/src/client/cover_traffic_stream.rs b/common/client-core/src/client/cover_traffic_stream.rs index 101a0a7754..8efea56e5a 100644 --- a/common/client-core/src/client/cover_traffic_stream.rs +++ b/common/client-core/src/client/cover_traffic_stream.rs @@ -2,7 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 use crate::client::mix_traffic::BatchMixMessageSender; -use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; use crate::client::topology_control::TopologyAccessor; use crate::{config, spawn_future}; use futures::task::{Context, Poll}; @@ -13,6 +12,7 @@ use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::cover::generate_loop_cover_packet; use nym_sphinx::params::{PacketSize, PacketType}; use nym_sphinx::utils::sample_poisson_duration; +use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender}; use rand::{rngs::OsRng, CryptoRng, Rng}; use std::pin::Pin; use std::sync::Arc; @@ -63,7 +63,7 @@ where packet_type: PacketType, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, } impl Stream for LoopCoverTrafficStream @@ -109,7 +109,7 @@ impl LoopCoverTrafficStream { topology_access: TopologyAccessor, traffic_config: config::Traffic, cover_config: config::CoverTraffic, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { let rng = OsRng; @@ -198,9 +198,9 @@ impl LoopCoverTrafficStream { } } } else { - self.stats_tx.report(PacketStatisticsEvent::CoverPacketSent( - cover_traffic_packet_size.size(), - )); + self.stats_tx.report( + PacketStatisticsEvent::CoverPacketSent(cover_traffic_packet_size.size()).into(), + ); } // TODO: I'm not entirely sure whether this is really required, because I'm not 100% diff --git a/common/client-core/src/client/mod.rs b/common/client-core/src/client/mod.rs index f8226da387..cd6b0dd688 100644 --- a/common/client-core/src/client/mod.rs +++ b/common/client-core/src/client/mod.rs @@ -7,9 +7,9 @@ pub(crate) mod helpers; pub mod inbound_messages; pub mod key_manager; pub mod mix_traffic; -pub mod packet_statistics_control; pub mod real_messages_control; pub mod received_buffer; pub mod replies; +pub mod statistics_control; pub mod topology_control; pub(crate) mod transmission_buffer; diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs index d7876c64ab..ae74c61536 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs @@ -1,9 +1,9 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; - use super::action_controller::{AckActionSender, Action}; +use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender}; + use futures::StreamExt; use log::*; use nym_gateway_client::AcknowledgementReceiver; @@ -19,7 +19,7 @@ pub(super) struct AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: AckActionSender, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, } impl AcknowledgementListener { @@ -27,7 +27,7 @@ impl AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: AckActionSender, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { AcknowledgementListener { ack_key, @@ -40,7 +40,7 @@ impl AcknowledgementListener { async fn on_ack(&mut self, ack_content: Vec) { trace!("Received an ack"); self.stats_tx - .report(PacketStatisticsEvent::AckReceived(ack_content.len())); + .report(PacketStatisticsEvent::AckReceived(ack_content.len()).into()); let frag_id = match recover_identifier(&self.ack_key, &ack_content) .map(FragmentIdentifier::try_from_bytes) @@ -57,13 +57,13 @@ impl AcknowledgementListener { if frag_id == COVER_FRAG_ID { trace!("Received an ack for a cover message - no need to do anything"); self.stats_tx - .report(PacketStatisticsEvent::CoverAckReceived(ack_content.len())); + .report(PacketStatisticsEvent::CoverAckReceived(ack_content.len()).into()); return; } trace!("Received {} from the mix network", frag_id); self.stats_tx - .report(PacketStatisticsEvent::RealAckReceived(ack_content.len())); + .report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into()); self.action_sender .unbounded_send(Action::new_remove(frag_id)) .unwrap(); diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs index ec35429b9f..9a6c237c2c 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs @@ -8,7 +8,6 @@ use self::{ sent_notification_listener::SentNotificationListener, }; use crate::client::inbound_messages::InputMessageReceiver; -use crate::client::packet_statistics_control::PacketStatisticsReporter; use crate::client::real_messages_control::message_handler::MessageHandler; use crate::client::replies::reply_controller::ReplyControllerSender; use crate::spawn_future; @@ -24,6 +23,7 @@ use nym_sphinx::{ chunking::fragment::{Fragment, FragmentIdentifier}, Delay as SphinxDelay, }; +use nym_statistics_common::clients::ClientStatsSender; use rand::{CryptoRng, Rng}; use std::{ sync::{Arc, Weak}, @@ -209,7 +209,7 @@ where connectors: AcknowledgementControllerConnectors, message_handler: MessageHandler, reply_controller_sender: ReplyControllerSender, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { let (retransmission_tx, retransmission_rx) = mpsc::unbounded(); diff --git a/common/client-core/src/client/real_messages_control/mod.rs b/common/client-core/src/client/real_messages_control/mod.rs index d4914f1440..9607b7a9b0 100644 --- a/common/client-core/src/client/real_messages_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/mod.rs @@ -35,7 +35,7 @@ use crate::client::replies::reply_controller; use crate::config; pub(crate) use acknowledgement_control::{AckActionSender, Action}; -use super::packet_statistics_control::PacketStatisticsReporter; +use nym_statistics_common::clients::ClientStatsSender; pub(crate) mod acknowledgement_control; pub(crate) mod message_handler; @@ -145,7 +145,7 @@ impl RealMessagesController { reply_controller_receiver: ReplyControllerReceiver, lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { let rng = OsRng; diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index f373fa0756..d39c1731b6 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -3,7 +3,6 @@ use self::sending_delay_controller::SendingDelayController; use crate::client::mix_traffic::BatchMixMessageSender; -use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; use crate::client::transmission_buffer::TransmissionBuffer; @@ -19,6 +18,7 @@ use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::params::PacketSize; use nym_sphinx::preparer::PreparedFragment; use nym_sphinx::utils::sample_poisson_duration; +use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender}; use nym_task::connections::{ ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane, }; @@ -115,8 +115,8 @@ where /// Report queue lengths so that upstream can backoff sending data, and keep connections open. lane_queue_lengths: LaneQueueLengths, - /// Channel used for sending statistics events to `PacketStatisticsControl`. - stats_tx: PacketStatisticsReporter, + /// Channel used for sending metrics events (specifically `PacketStatistics` events) to the metrics tracker. + stats_tx: ClientStatsSender, } #[derive(Debug)] @@ -175,7 +175,7 @@ where topology_access: TopologyAccessor, lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { OutQueueControl { config, @@ -277,7 +277,7 @@ where } else { PacketStatisticsEvent::CoverPacketSent(packet_size) }; - self.stats_tx.report(event); + self.stats_tx.report(event.into()); } // notify ack controller about sending our message only after we actually managed to push it @@ -373,13 +373,13 @@ where TransmissionLane::Retransmission => Some(PacketStatisticsEvent::RetransmissionQueued), }; if let Some(stat_event) = stat_event { - self.stats_tx.report(stat_event); + self.stats_tx.report(stat_event.into()); } // To avoid comparing apples to oranges when presenting the fraction of packets that are // retransmissions, we also need to keep track to the total number of real messages queued, // even though we also track the actual number of messages sent later in the pipeline. self.stats_tx - .report(PacketStatisticsEvent::RealPacketQueued); + .report(PacketStatisticsEvent::RealPacketQueued.into()); Some(real_next) } diff --git a/common/client-core/src/client/received_buffer.rs b/common/client-core/src/client/received_buffer.rs index 021fd75d1a..8869e722db 100644 --- a/common/client-core/src/client/received_buffer.rs +++ b/common/client-core/src/client/received_buffer.rs @@ -1,9 +1,8 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::client::{ - packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}, - replies::{reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys}, +use crate::client::replies::{ + reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys, }; use crate::spawn_future; use futures::channel::mpsc; @@ -20,6 +19,7 @@ use nym_sphinx::anonymous_replies::{encryption_key::EncryptionKeyDigest, SurbEnc use nym_sphinx::message::{NymMessage, PlainMessage}; use nym_sphinx::params::ReplySurbKeyDigestAlgorithm; use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage}; +use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender}; use std::collections::HashSet; use std::sync::Arc; @@ -46,7 +46,7 @@ struct ReceivedMessagesBufferInner { // and every now and then remove ids older than X recently_reconstructed: HashSet, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, } impl ReceivedMessagesBufferInner { @@ -61,16 +61,12 @@ impl ReceivedMessagesBufferInner { // received and sent packets due to the sphinx layers being removed by the exit gateway // before it reaches the mixnet client. self.stats_tx - .report(PacketStatisticsEvent::CoverPacketReceived( - fragment_data_size, - )); + .report(PacketStatisticsEvent::CoverPacketReceived(fragment_data_size).into()); return None; } self.stats_tx - .report(PacketStatisticsEvent::RealPacketReceived( - fragment_data_size, - )); + .report(PacketStatisticsEvent::RealPacketReceived(fragment_data_size).into()); let fragment = match self.message_receiver.recover_fragment(fragment_data) { Err(err) => { @@ -163,7 +159,7 @@ impl ReceivedMessagesBuffer { local_encryption_keypair: Arc, reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, - stats_tx: PacketStatisticsReporter, + stats_tx: ClientStatsSender, ) -> Self { ReceivedMessagesBuffer { inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner { @@ -504,13 +500,13 @@ impl ReceivedMessagesBufferControll mixnet_packet_receiver: MixnetMessageReceiver, reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, - packet_statistics_reporter: PacketStatisticsReporter, + metrics_reporter: ClientStatsSender, ) -> Self { let received_buffer = ReceivedMessagesBuffer::new( local_encryption_keypair, reply_key_storage, reply_controller_sender, - packet_statistics_reporter, + metrics_reporter, ); ReceivedMessagesBufferController { diff --git a/common/client-core/src/client/statistics_control.rs b/common/client-core/src/client/statistics_control.rs new file mode 100644 index 0000000000..df027f13fb --- /dev/null +++ b/common/client-core/src/client/statistics_control.rs @@ -0,0 +1,154 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! # Statistics collection and reporting. +//! +//! Modular metrics collection and reporting system. submodules can be added to collect different types of metrics. +//! On creation the Statistics controller will start a task that will listen for incoming stats events and +//! multiplex them out to the appropriate metrics module based on type. +//! +//! Adding A new module you need to write a new module that implements the `StatsObj` trait and add it to +//! the `stats` hashmap in the `StatisticsControl` struct during it's initialization in the `new` function in +//! this file. + +#![warn(clippy::expect_used)] +#![warn(clippy::unwrap_used)] +#![warn(clippy::todo)] +#![warn(clippy::dbg_macro)] + +use std::time::Duration; + +use nym_sphinx::addressing::Recipient; +use nym_statistics_common::{ + clients::{ClientStatsController, ClientStatsReceiver, ClientStatsSender}, + StatsReportingConfig, +}; +use nym_task::connections::TransmissionLane; + +use crate::{ + client::inbound_messages::{InputMessage, InputMessageSender}, + spawn_future, +}; + +/// Time interval between reporting statistics to the given provider if it exist +const STATS_REPORT_INTERVAL_SECS: u64 = 300; +/// Time interval between reporting statistics locally (logging/task_client) +const LOCAL_REPORT_INTERVAL: Duration = Duration::from_secs(2); +/// Interval for taking snapshots of the statistics +const SNAPSHOT_INTERVAL: Duration = Duration::from_millis(500); + +/// Launches and manages metrics collection and reporting. +/// +/// This is designed to be generic to allow for multiple types of metrics to be collected and +/// reported. +pub(crate) struct StatisticsControl { + /// Keep store the different types of metrics collectors + stats: ClientStatsController, + + /// Incoming packet stats events from other tasks + stats_rx: ClientStatsReceiver, + + /// Channel to send stats report through the mixnet + report_tx: InputMessageSender, + + /// Service-provider address to send stats reports + reporting_address: Option, +} + +impl StatisticsControl { + pub(crate) fn create( + reporting_config: Option, + client_stats_id: String, + report_tx: InputMessageSender, + ) -> (Self, ClientStatsSender) { + let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel(); + let (reporting_address, client_type) = match reporting_config { + Some(cfg) => (Some(cfg.reporting_address), cfg.reporting_type), + None => (None, "".into()), + }; + + let stats = ClientStatsController::new(client_stats_id, client_type); + + ( + StatisticsControl { + stats, + stats_rx, + reporting_address, + report_tx, + }, + ClientStatsSender::new(Some(stats_tx)), + ) + } + + async fn report_stats(&mut self, recipient: Recipient) { + let stats_report = self.stats.build_report(); + + let report_message = InputMessage::new_regular( + recipient, + stats_report.into(), + TransmissionLane::General, + None, + ); + if let Err(err) = self.report_tx.send(report_message).await { + log::error!("Failed to report client stats: {:?}", err); + } else { + self.stats.reset(); + } + } + + async fn run_with_shutdown(&mut self, mut task_client: nym_task::TaskClient) { + log::debug!("Started StatisticsControl with graceful shutdown support"); + + let stats_report_interval = Duration::from_secs(STATS_REPORT_INTERVAL_SECS); + let mut stats_report_interval = tokio::time::interval(stats_report_interval); + let mut local_report_interval = tokio::time::interval(LOCAL_REPORT_INTERVAL); + let mut snapshot_interval = tokio::time::interval(SNAPSHOT_INTERVAL); + + loop { + tokio::select! { + stats_event = self.stats_rx.recv() => match stats_event { + Some(stats_event) => self.stats.handle_event(stats_event), + None => { + log::trace!("StatisticsControl: shutting down due to closed stats channel"); + break; + } + }, + _ = snapshot_interval.tick() => { + self.stats.snapshot(); + } + _ = stats_report_interval.tick(), if self.reporting_address.is_some() => { + // SAFTEY : this branch executes only if reporting is not none, so unwrapp is fine + #[allow(clippy::unwrap_used)] + self.report_stats(self.reporting_address.unwrap()).await; + } + + _ = local_report_interval.tick() => { + self.stats.local_report(&mut task_client); + } + _ = task_client.recv_with_delay() => { + log::trace!("StatisticsControl: Received shutdown"); + break; + }, + } + } + task_client.recv_timeout().await; + log::debug!("StatisticsControl: Exiting"); + } + + pub(crate) fn start_with_shutdown(mut self, task_client: nym_task::TaskClient) { + spawn_future(async move { + self.run_with_shutdown(task_client).await; + }) + } + + pub(crate) fn create_and_start_with_shutdown( + reporting_config: Option, + client_stats_id: String, + report_tx: InputMessageSender, + task_client: nym_task::TaskClient, + ) -> ClientStatsSender { + let (controller, sender) = Self::create(reporting_config, client_stats_id, report_tx); + controller.start_with_shutdown(task_client); + sender + } +} diff --git a/common/statistics/Cargo.toml b/common/statistics/Cargo.toml index 3daba412d5..9a8242f34a 100644 --- a/common/statistics/Cargo.toml +++ b/common/statistics/Cargo.toml @@ -11,7 +11,16 @@ license.workspace = true [dependencies] futures = { workspace = true } +log = { workspace = true } +sysinfo = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } time = { workspace = true } +tokio = { workspace = true } +si-scale = { workspace = true } nym-sphinx = { path = "../nymsphinx" } nym-credentials-interface = { path = "../credentials-interface" } +nym-metrics = { path = "../nym-metrics" } +nym-task = { path = "../task" } diff --git a/common/statistics/src/clients/gateway_conn_statistics.rs b/common/statistics/src/clients/gateway_conn_statistics.rs new file mode 100644 index 0000000000..961c7a0f7f --- /dev/null +++ b/common/statistics/src/clients/gateway_conn_statistics.rs @@ -0,0 +1,83 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! # Gateway Connection statistics +//! +//! Metrics collected by the client while establishing and maintaining connections to the gateway. + +use super::ClientStatsEvents; +use std::collections::VecDeque; + +use nym_metrics::{inc, inc_by}; +use serde::{Deserialize, Serialize}; + +#[derive(Default, Debug, Clone, Serialize, Deserialize)] +pub(crate) struct GatewayStats { + // Sent + real_packets_sent: u64, + real_packets_sent_size: usize, + + /// failed connection statistics + failures: VecDeque<()>, // TODO +} + +impl GatewayStats { + fn handle(&mut self, event: GatewayStatsEvent) { + match event { + GatewayStatsEvent::RealPacketSent(packet_size) => { + self.real_packets_sent += 1; + self.real_packets_sent_size += packet_size; + inc!("real_packets_sent"); + inc_by!("real_packets_sent_size", packet_size); + } + } + } + + fn summary(&self) -> (String, String) { + ( + format!("packets sent: {}", self.real_packets_sent), + "packets received: todo".to_owned(), + ) + } +} + +impl From for ClientStatsEvents { + fn from(event: GatewayStatsEvent) -> ClientStatsEvents { + ClientStatsEvents::GatewayConn(event) + } +} + +/// Event space for Gateway Connection Events +#[derive(Debug)] +pub enum GatewayStatsEvent { + /// The real packets sent. Recall that acks are sent by the gateway, so it's not included here. + RealPacketSent(usize), +} + +/// Gateway Statistics Tracking +#[derive(Default)] +pub struct GatewayStatsControl { + // Keep track of packet statistics over time + stats: GatewayStats, +} + +impl GatewayStatsControl { + pub(crate) fn handle_event(&mut self, event: GatewayStatsEvent) { + self.stats.handle(event) + } + + pub(crate) fn report(&self) -> GatewayStats { + self.stats.clone() + } + + pub(crate) fn local_report(&self) { + self.report_counters(); + } + + fn report_counters(&self) { + log::trace!("packet statistics: {:?}", &self.stats); + let (summary_sent, summary_recv) = self.stats.summary(); + log::debug!("{}", summary_sent); + log::debug!("{}", summary_recv); + } +} diff --git a/common/statistics/src/clients/mod.rs b/common/statistics/src/clients/mod.rs new file mode 100644 index 0000000000..9f94cda230 --- /dev/null +++ b/common/statistics/src/clients/mod.rs @@ -0,0 +1,135 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::report::{ClientStatsReport, OsInformation}; + +use nym_task::TaskClient; +use time::{OffsetDateTime, Time}; +use tokio::sync::mpsc::UnboundedSender; + +/// Active gateway connection statistics. +pub mod gateway_conn_statistics; + +/// Nym API connection statistics. +pub mod nym_api_statistics; + +/// Packet count based statistics. +pub mod packet_statistics; + +/// Channel receiving generic stats events to be used by a statistics aggregator. +pub type ClientStatsReceiver = tokio::sync::mpsc::UnboundedReceiver; + +/// Channel allowing generic statistics events to be reported to a stats event aggregator +#[derive(Clone)] +pub struct ClientStatsSender { + stats_tx: Option>, +} + +impl ClientStatsSender { + /// Create a new statistics Sender + pub fn new(stats_tx: Option>) -> Self { + ClientStatsSender { stats_tx } + } + + /// Report a statistics event using the sender. + pub fn report(&mut self, event: ClientStatsEvents) { + if let Some(tx) = self.stats_tx.as_mut() { + if let Err(err) = tx.send(event) { + log::error!("Failed to send stats event: {:?}", err); + } + } + } +} + +/// Client Statistics events (static for now) +pub enum ClientStatsEvents { + /// Packet count events + PacketStatistics(packet_statistics::PacketStatisticsEvent), + /// Gateway Connection events + GatewayConn(gateway_conn_statistics::GatewayStatsEvent), + /// Nym API connection events + NymApi(nym_api_statistics::NymApiStatsEvent), +} + +/// Controls stats event handling and reporting +pub struct ClientStatsController { + //static infos + last_update_time: OffsetDateTime, + client_id: String, + client_type: String, + os_information: OsInformation, + + // stats collection modules + packet_stats: packet_statistics::PacketStatisticsControl, + gateway_conn_stats: gateway_conn_statistics::GatewayStatsControl, + nym_api_stats: nym_api_statistics::NymApiStatsControl, +} + +impl ClientStatsController { + /// Creates a ClientStatsController given a client_id + pub fn new(client_id: String, client_type: String) -> Self { + ClientStatsController { + last_update_time: ClientStatsController::get_update_time(), + client_id, + client_type, + os_information: OsInformation::new(), + packet_stats: Default::default(), + gateway_conn_stats: Default::default(), + nym_api_stats: Default::default(), + } + } + /// Returns a static ClientStatsReport that can be sent somewhere + pub fn build_report(&self) -> ClientStatsReport { + ClientStatsReport { + last_update_time: self.last_update_time, + client_id: self.client_id.clone(), + client_type: self.client_type.clone(), + os_information: self.os_information.clone(), + packet_stats: self.packet_stats.report(), + gateway_conn_stats: self.gateway_conn_stats.report(), + nym_api_stats: self.nym_api_stats.report(), + } + } + + /// Handle and dispatch incoming stats event + pub fn handle_event(&mut self, stats_event: ClientStatsEvents) { + match stats_event { + ClientStatsEvents::PacketStatistics(event) => self.packet_stats.handle_event(event), + ClientStatsEvents::GatewayConn(event) => self.gateway_conn_stats.handle_event(event), + ClientStatsEvents::NymApi(event) => self.nym_api_stats.handle_event(event), + } + } + + /// Reset the metrics to their initial state. + /// + /// Used to periodically reset the metrics in accordance with periodic reporting strategy + pub fn reset(&mut self) { + self.nym_api_stats = Default::default(); + self.gateway_conn_stats = Default::default(); + //no periodic reset for packet stats + + self.last_update_time = ClientStatsController::get_update_time(); + } + + /// snapshot the current state of the metrics for module that needs it + pub fn snapshot(&mut self) { + //no snapshot for gateway_conn_stats + //no snapshot for nym_api_stats + self.packet_stats.snapshot(); + } + + pub fn local_report(&mut self, task_client: &mut TaskClient) { + self.packet_stats.local_report(task_client); + self.gateway_conn_stats.local_report(); + self.nym_api_stats.local_report(); + } + + fn get_update_time() -> OffsetDateTime { + let now = OffsetDateTime::now_utc(); + #[allow(clippy::unwrap_used)] + //Safety : 0 is always a valid number of seconds, hours and minutes comes from a valid source + let new_time = Time::from_hms(now.hour(), now.minute(), 0).unwrap(); + //allows a bigger anonymity by hiding exact sending time + now.replace_time(new_time) + } +} diff --git a/common/statistics/src/clients/nym_api_statistics.rs b/common/statistics/src/clients/nym_api_statistics.rs new file mode 100644 index 0000000000..9c3ee609c0 --- /dev/null +++ b/common/statistics/src/clients/nym_api_statistics.rs @@ -0,0 +1,83 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! # API Connection statistics +//! +//! Metrics collected by the client while attempting to pull config from the API. + +use super::ClientStatsEvents; +use std::collections::VecDeque; + +use nym_metrics::{inc, inc_by}; +use serde::{Deserialize, Serialize}; + +#[derive(Default, Debug, Clone, Serialize, Deserialize)] +pub(crate) struct NymApiStats { + // Sent + real_packets_sent: u64, + real_packets_sent_size: usize, + + /// API connection failure statistics + failures: VecDeque<()>, // TODO +} + +impl NymApiStats { + fn handle(&mut self, event: NymApiStatsEvent) { + match event { + NymApiStatsEvent::RealPacketSent(packet_size) => { + self.real_packets_sent += 1; + self.real_packets_sent_size += packet_size; + inc!("real_packets_sent"); + inc_by!("real_packets_sent_size", packet_size); + } + } + } + + fn summary(&self) -> (String, String) { + ( + format!("packets sent: {}", self.real_packets_sent,), + "packets received: todo".to_owned(), + ) + } +} + +/// Event space for Nym API statistics tracking +#[derive(Debug)] +pub enum NymApiStatsEvent { + /// The real packets sent. Recall that acks are sent by the Api, so it's not included here. + RealPacketSent(usize), +} + +impl From for ClientStatsEvents { + fn from(event: NymApiStatsEvent) -> ClientStatsEvents { + ClientStatsEvents::NymApi(event) + } +} + +/// Nym API statistics tracking object +#[derive(Default)] +pub struct NymApiStatsControl { + // Keep track of packet statistics over time + stats: NymApiStats, +} + +impl NymApiStatsControl { + pub(crate) fn handle_event(&mut self, event: NymApiStatsEvent) { + self.stats.handle(event) + } + + pub(crate) fn report(&self) -> NymApiStats { + self.stats.clone() + } + + pub(crate) fn local_report(&self) { + self.report_counters(); + } + + fn report_counters(&self) { + log::trace!("packet statistics: {:?}", &self.stats); + let (summary_sent, summary_recv) = self.stats.summary(); + log::debug!("{}", summary_sent); + log::debug!("{}", summary_recv); + } +} diff --git a/common/client-core/src/client/packet_statistics_control.rs b/common/statistics/src/clients/packet_statistics.rs similarity index 68% rename from common/client-core/src/client/packet_statistics_control.rs rename to common/statistics/src/clients/packet_statistics.rs index bfdca3c38a..5d6d1f9c1b 100644 --- a/common/client-core/src/client/packet_statistics_control.rs +++ b/common/statistics/src/clients/packet_statistics.rs @@ -1,3 +1,7 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use super::ClientStatsEvents; use core::fmt; use std::{ collections::VecDeque, @@ -5,45 +9,17 @@ use std::{ }; use nym_metrics::{inc, inc_by}; +use serde::{Deserialize, Serialize}; use si_scale::helpers::bibytes2; -// Metrics server -use futures::future::{FusedFuture, OptionFuture}; -use futures::FutureExt; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use http_body_util::Full; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use hyper::body::Bytes; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use hyper::server::conn::http1; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use hyper::service::service_fn; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use hyper::{Request, Response}; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use hyper_util::rt::TokioIo; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use std::convert::Infallible; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -#[cfg(feature = "metrics-server")] -use std::net::SocketAddr; -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -use tokio::net::TcpListener; - -use crate::spawn_future; - -// Time interval between reporting packet statistics -const PACKET_REPORT_INTERVAL_SECS: u64 = 2; -// Interval for taking snapshots of the packet statistics -const SNAPSHOT_INTERVAL_MS: u64 = 500; // When computing rates, we include snapshots that are up to this old. We set it to some odd number // a tad larger than an integer number of snapshot intervals, so that we don't have to worry about // threshold effects. // Also, set it larger than the packet report interval so that we don't miss notable singular events const RECORDING_WINDOW_MS: u64 = 2300; -#[derive(Default, Debug, Clone)] -struct PacketStatistics { +#[derive(Default, Debug, Clone, Serialize, Deserialize)] +pub(crate) struct PacketStatistics { // Sent real_packets_sent: u64, real_packets_sent_size: usize, @@ -73,7 +49,7 @@ struct PacketStatistics { } impl PacketStatistics { - fn handle_event(&mut self, event: PacketStatisticsEvent) { + fn handle(&mut self, event: PacketStatisticsEvent) { match event { PacketStatisticsEvent::RealPacketSent(packet_size) => { self.real_packets_sent += 1; @@ -366,56 +342,46 @@ impl PacketRates { } } +/// Event Space used for counting the Packet types used in a connection. #[derive(Debug)] -pub(crate) enum PacketStatisticsEvent { - // The real packets sent. Recall that acks are sent by the gateway, so it's not included here. +pub enum PacketStatisticsEvent { + /// The real packets sent. Recall that acks are sent by the gateway, so it's not included here. RealPacketSent(usize), - // The cover packets sent + /// The cover packets sent CoverPacketSent(usize), - // Real packets received + /// Real packets received RealPacketReceived(usize), - // Cover packets received + /// Cover packets received CoverPacketReceived(usize), - // Ack of any type received. This is mostly used as a consistency check, and should be the sum - // of real and cover acks received. + /// Ack of any type received. This is mostly used as a consistency check, and should be the sum + /// of real and cover acks received. AckReceived(usize), - // Out of the total acks received, this is the subset of those that were real + /// Out of the total acks received, this is the subset of those that were real RealAckReceived(usize), - // Out of the total acks received, this is the subset of those that were for cover traffic + /// Out of the total acks received, this is the subset of those that were for cover traffic CoverAckReceived(usize), - // Types of packets queued + /// Types of packets queued RealPacketQueued, + /// Types of packets queued RetransmissionQueued, + /// Types of packets queued ReplySurbRequestQueued, + /// Types of packets queued AdditionalReplySurbRequestQueued, } -type PacketStatisticsReceiver = tokio::sync::mpsc::UnboundedReceiver; - -#[derive(Clone)] -pub(crate) struct PacketStatisticsReporter { - stats_tx: tokio::sync::mpsc::UnboundedSender, -} - -impl PacketStatisticsReporter { - pub(crate) fn new(stats_tx: tokio::sync::mpsc::UnboundedSender) -> Self { - Self { stats_tx } - } - - pub(crate) fn report(&self, event: PacketStatisticsEvent) { - self.stats_tx.send(event).unwrap_or_else(|err| { - log::error!("Failed to report packet stat: {:?}", err); - }); +impl From for ClientStatsEvents { + fn from(event: PacketStatisticsEvent) -> ClientStatsEvents { + ClientStatsEvents::PacketStatistics(event) } } -pub(crate) struct PacketStatisticsControl { - // Incoming packet stats events from other tasks - stats_rx: PacketStatisticsReceiver, - +/// Statistics tracking for Packet based I/O +#[derive(Default)] +pub struct PacketStatisticsControl { // Keep track of packet statistics over time stats: PacketStatistics, @@ -428,18 +394,28 @@ pub(crate) struct PacketStatisticsControl { } impl PacketStatisticsControl { - pub(crate) fn new() -> (Self, PacketStatisticsReporter) { - let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel(); + pub(crate) fn handle_event(&mut self, event: PacketStatisticsEvent) { + self.stats.handle(event) + } - ( - Self { - stats_rx, - stats: PacketStatistics::default(), - history: VecDeque::new(), - rates: VecDeque::new(), - }, - PacketStatisticsReporter::new(stats_tx), - ) + pub(crate) fn snapshot(&mut self) { + self.update_history(); + self.update_rates(); + } + + pub(crate) fn report(&self) -> PacketStatistics { + self.stats.clone() + } + + pub(crate) fn local_report(&mut self, task_client: &mut nym_task::TaskClient) { + let rates = self.report_rates(); + self.check_for_notable_events(); + self.report_counters(); + + // Report our current bandwidth used to e.g a GUI client + if let Some(rates) = rates { + task_client.send_status_msg(Box::new(MixnetBandwidthStatisticsEvent::new(rates))); + } } // Add the current stats to the history, and remove old ones. @@ -536,129 +512,4 @@ impl PacketStatisticsControl { // IDEA: if there is a burst of acks, that could indicate tokio task starvation. } - - pub(crate) async fn run_with_shutdown(&mut self, mut task_client: nym_task::TaskClient) { - log::debug!("Started PacketStatisticsControl with graceful shutdown support"); - - let report_interval = Duration::from_secs(PACKET_REPORT_INTERVAL_SECS); - let mut report_interval = tokio::time::interval(report_interval); - let snapshot_interval = Duration::from_millis(SNAPSHOT_INTERVAL_MS); - let mut snapshot_interval = tokio::time::interval(snapshot_interval); - - cfg_if::cfg_if! { - if #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] { - log::warn!("Metrics server is not supported on wasm32-unknown-unknown"); - let listener: Option = None; - } else if #[cfg(feature = "metrics-server")] { - let mut metrics_port = 18000; - let listener: Option; - loop { - let addr = SocketAddr::from(([0, 0, 0, 0], metrics_port)); - match TcpListener::bind(addr).await { - Ok(l) => { - log::info!("###############################"); - log::info!("Metrics endpoint is at: {:?}", l.local_addr()); - log::info!("###############################"); - listener = Some(l); - break; - }, - Err(err) => { - log::warn!("Failed to bind metrics server: {:?}", err); - metrics_port += 1; - } - }; - } - } else { - log::info!("Metrics server is disabled!"); - let listener: Option = None; - } - } - - loop { - // it seems at some point tokio changed its select precondition evaluation, - // and it's no longer checked before the future is evaluated. - let accept_future: OptionFuture<_> = listener - .as_ref() - .map(|l| l.accept()) - .map(FutureExt::fuse) - .into(); - - tokio::select! { - stats_event = self.stats_rx.recv() => match stats_event { - Some(stats_event) => { - log::trace!("PacketStatisticsControl: Received stats event"); - self.stats.handle_event(stats_event); - }, - None => { - log::trace!("PacketStatisticsControl: stopping since stats channel was closed"); - break; - } - }, - // conditional will disable the branch if we're in wasm32-unknown-unknown - // use `_` to calm down clippy when running for wasm - _result = accept_future, if !accept_future.is_terminated() => { - cfg_if::cfg_if! { - if #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] { - if let Some(Ok((stream, _))) = _result { - let io = TokioIo::new(stream); - - tokio::task::spawn(async move { - if let Err(err) = http1::Builder::new() - .serve_connection(io, service_fn(serve_metrics)) - .await - { - log::warn!("Error serving connection: {:?}", err); - } - }); - } else { - log::warn!("Error accepting connection"); - } - } - } - } - _ = snapshot_interval.tick() => { - self.update_history(); - self.update_rates(); - } - _ = report_interval.tick() => { - let rates = self.report_rates(); - self.check_for_notable_events(); - self.report_counters(); - - // Report our current bandwidth used to e.g a GUI client - if let Some(rates) = rates { - task_client.send_status_msg(Box::new(MixnetBandwidthStatisticsEvent::new(rates))); - } - } - _ = task_client.recv_with_delay() => { - log::trace!("PacketStatisticsControl: Received shutdown"); - break; - }, - } - } - log::debug!("PacketStatisticsControl: Exiting"); - } - - pub(crate) fn start_with_shutdown(mut self, task_client: nym_task::TaskClient) { - spawn_future(async move { - self.run_with_shutdown(task_client).await; - }) - } -} - -#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] -async fn serve_metrics( - _: Request, -) -> Result>, Infallible> { - use nym_metrics::metrics; - - Ok(Response::new(Full::new(Bytes::from(metrics!())))) -} - -#[cfg(all(target_arch = "wasm32", target_os = "unknown"))] -struct WasmEmpty; - -#[cfg(all(target_arch = "wasm32", target_os = "unknown"))] -impl WasmEmpty { - async fn accept(&self) {} } diff --git a/common/statistics/src/error.rs b/common/statistics/src/error.rs new file mode 100644 index 0000000000..2db30c0ea5 --- /dev/null +++ b/common/statistics/src/error.rs @@ -0,0 +1,15 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use thiserror::Error; + +/// Error types occurring while processing statistics events and reporting. +#[derive(Debug, Error)] +#[allow(missing_docs)] +pub enum StatsError { + #[error("Failed to (de)serialize stats report : {0}")] + ReportJsonSerialization(#[from] serde_json::Error), +} + +/// Result of a statistics operation. +pub type Result = core::result::Result; diff --git a/common/statistics/src/events.rs b/common/statistics/src/events.rs deleted file mode 100644 index a9b81defd6..0000000000 --- a/common/statistics/src/events.rs +++ /dev/null @@ -1,54 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -use futures::channel::mpsc; -use nym_credentials_interface::TicketType; -use nym_sphinx::DestinationAddressBytes; -use time::OffsetDateTime; - -pub type StatsEventSender = mpsc::UnboundedSender; -pub type StatsEventReceiver = mpsc::UnboundedReceiver; -pub enum StatsEvent { - SessionStatsEvent(SessionEvent), -} - -impl StatsEvent { - pub fn new_session_start(client: DestinationAddressBytes) -> StatsEvent { - StatsEvent::SessionStatsEvent(SessionEvent::SessionStart { - start_time: OffsetDateTime::now_utc(), - client, - }) - } - - pub fn new_session_stop(client: DestinationAddressBytes) -> StatsEvent { - StatsEvent::SessionStatsEvent(SessionEvent::SessionStop { - stop_time: OffsetDateTime::now_utc(), - client, - }) - } - - pub fn new_ecash_ticket( - client: DestinationAddressBytes, - ticket_type: TicketType, - ) -> StatsEvent { - StatsEvent::SessionStatsEvent(SessionEvent::EcashTicket { - ticket_type, - client, - }) - } -} - -pub enum SessionEvent { - SessionStart { - start_time: OffsetDateTime, - client: DestinationAddressBytes, - }, - SessionStop { - stop_time: OffsetDateTime, - client: DestinationAddressBytes, - }, - EcashTicket { - ticket_type: TicketType, - client: DestinationAddressBytes, - }, -} diff --git a/common/statistics/src/gateways.rs b/common/statistics/src/gateways.rs new file mode 100644 index 0000000000..398fda559a --- /dev/null +++ b/common/statistics/src/gateways.rs @@ -0,0 +1,89 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_credentials_interface::TicketType; +use nym_sphinx::DestinationAddressBytes; +use time::OffsetDateTime; + +/// Channel for receiving incoming Stats events +pub type GatewayStatsReceiver = tokio::sync::mpsc::UnboundedReceiver; + +/// Channel allowing for generic statistics events to be reported to a stats event aggregator. +#[derive(Clone)] +pub struct GatewayStatsReporter { + stats_tx: tokio::sync::mpsc::UnboundedSender, +} + +impl GatewayStatsReporter { + /// Construct a new gateway statistics event reporter + pub fn new(stats_tx: tokio::sync::mpsc::UnboundedSender) -> Self { + Self { stats_tx } + } + + /// Report a gateway statistivs event using the reporter + pub fn report(&self, event: GatewayStatsEvent) { + self.stats_tx.send(event).unwrap_or_else(|err| { + log::error!("Failed to report gateway stat event : {err}"); + }); + } +} + +/// Gateway Statistics events +pub enum GatewayStatsEvent { + /// Events in the lifecycle of an established client tunnel + SessionStatsEvent(SessionEvent), +} + +impl GatewayStatsEvent { + /// A new session between this gateway and the client remote has successfully opened + pub fn new_session_start(client: DestinationAddressBytes) -> GatewayStatsEvent { + GatewayStatsEvent::SessionStatsEvent(SessionEvent::SessionStart { + start_time: OffsetDateTime::now_utc(), + client, + }) + } + + /// An existing session with the client remote has ended + pub fn new_session_stop(client: DestinationAddressBytes) -> GatewayStatsEvent { + GatewayStatsEvent::SessionStatsEvent(SessionEvent::SessionStop { + stop_time: OffsetDateTime::now_utc(), + client, + }) + } + + /// A new ecash ticket has been added / requested + pub fn new_ecash_ticket( + client: DestinationAddressBytes, + ticket_type: TicketType, + ) -> GatewayStatsEvent { + GatewayStatsEvent::SessionStatsEvent(SessionEvent::EcashTicket { + ticket_type, + client, + }) + } +} + +/// Events in the lifecycle of an established client tunnel +pub enum SessionEvent { + /// A new session between this gateway and the client remote has successfully opened + SessionStart { + /// The timestamp of the session open event + start_time: OffsetDateTime, + /// Address of the remote client opening the connection + client: DestinationAddressBytes, + }, + /// An existing session with the client remote has ended + SessionStop { + /// Timestamp of the session end event + stop_time: OffsetDateTime, + /// Address of the remote client opening the connection + client: DestinationAddressBytes, + }, + /// A new ecash ticket has been added / requested + EcashTicket { + /// Type of ecash ticket that has been created as part of the session + ticket_type: TicketType, + /// Address of the remote client opening the connection + client: DestinationAddressBytes, + }, +} diff --git a/common/statistics/src/lib.rs b/common/statistics/src/lib.rs index 222251db6c..4c3a8856d3 100644 --- a/common/statistics/src/lib.rs +++ b/common/statistics/src/lib.rs @@ -1,4 +1,69 @@ // Copyright 2024 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only +// SPDX-License-Identifier: Apache-2.0 -pub mod events; +//! Nym Statistics +//! +//! This crate contains basic statistics utilities and abstractions to be re-used and +//! applied throughout both the client and gateway implementations. + +#![warn(clippy::expect_used)] +#![warn(clippy::unwrap_used)] +#![warn(clippy::todo)] +#![warn(clippy::dbg_macro)] + +use nym_sphinx::addressing::Recipient; + +/// Client specific statistics interfaces and events. +pub mod clients; +/// Statistics related errors. +pub mod error; +/// Gateway specific statistics interfaces and events. +pub mod gateways; +/// Statistics reporting abstractions and implementations. +pub mod report; + +/// Stats Reporting config to use when stats reporting is wanted +#[derive(Clone, Debug)] +pub struct StatsReportingConfig { + /// Client address to report to + pub reporting_address: Recipient, + + /// Type of client reporting (vpn_client, authenticator, native-client) + pub reporting_type: String, +} + +impl StatsReportingConfig { + /// Create a StatsReportingConfig for a native client + pub fn new_native(reporting_address: Recipient) -> Self { + StatsReportingConfig { + reporting_address, + reporting_type: NATIVE_CLIENT.to_string(), + } + } + /// Create a StatsReportingConfig for a vpn client + pub fn new_vpn(reporting_address: Recipient) -> Self { + StatsReportingConfig { + reporting_address, + reporting_type: VPN_CLIENT.to_string(), + } + } + /// Create a StatsReportingConfig for a socks5 client + pub fn new_socks5(reporting_address: Recipient) -> Self { + StatsReportingConfig { + reporting_address, + reporting_type: SOCKS5_CLIENT.to_string(), + } + } + /// Create a StatsReportingConfig for an unspecified client + pub fn new_unknown(reporting_address: Recipient) -> Self { + StatsReportingConfig { + reporting_address, + reporting_type: UNKNOWN.to_string(), + } + } +} + +const VPN_CLIENT: &str = "vpn_client"; +const NATIVE_CLIENT: &str = "native_client"; +const SOCKS5_CLIENT: &str = "socks5_client"; +const UNKNOWN: &str = "unknown"; diff --git a/common/statistics/src/report.rs b/common/statistics/src/report.rs new file mode 100644 index 0000000000..0ca15f9d80 --- /dev/null +++ b/common/statistics/src/report.rs @@ -0,0 +1,64 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::clients::{ + gateway_conn_statistics::GatewayStats, nym_api_statistics::NymApiStats, + packet_statistics::PacketStatistics, +}; + +use super::error::StatsError; + +use serde::{Deserialize, Serialize}; +use sysinfo::System; +use time::OffsetDateTime; + +/// 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) 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, +} + +impl From for Vec { + fn from(value: ClientStatsReport) -> Self { + // safety, no custom serialisation + #[allow(clippy::unwrap_used)] + let report_json = serde_json::to_string(&value).unwrap(); + report_json.as_bytes().to_vec() + } +} + +impl TryFrom<&[u8]> for ClientStatsReport { + type Error = StatsError; + fn try_from(value: &[u8]) -> Result { + Ok(serde_json::from_slice(value)?) + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct OsInformation { + pub(crate) os_type: String, + pub(crate) os_version: Option, + pub(crate) os_arch: Option, +} + +impl OsInformation { + pub fn new() -> Self { + OsInformation { + os_type: System::distribution_id(), + os_version: System::long_os_version(), + os_arch: System::cpu_arch(), + } + } +} + +impl Default for OsInformation { + fn default() -> Self { + Self::new() + } +} diff --git a/gateway/src/node/client_handling/active_clients.rs b/gateway/src/node/client_handling/active_clients.rs index 82cdf98dc3..fbf83ce92c 100644 --- a/gateway/src/node/client_handling/active_clients.rs +++ b/gateway/src/node/client_handling/active_clients.rs @@ -5,8 +5,8 @@ use super::websocket::message_receiver::{IsActiveRequestSender, MixMessageSender use crate::node::client_handling::embedded_clients::LocalEmbeddedClientHandle; use dashmap::DashMap; use nym_sphinx::DestinationAddressBytes; -use nym_statistics_common::events; -use nym_statistics_common::events::StatsEventSender; +use nym_statistics_common::gateways; +use nym_statistics_common::gateways::GatewayStatsReporter; use std::sync::Arc; use tracing::warn; @@ -37,7 +37,7 @@ impl ActiveClient { #[derive(Clone)] pub(crate) struct ActiveClientsStore { inner: Arc>, - stats_event_sender: StatsEventSender, + stats_event_reporter: GatewayStatsReporter, } #[derive(Clone)] @@ -51,10 +51,10 @@ pub(crate) struct ClientIncomingChannels { impl ActiveClientsStore { /// Creates new instance of `ActiveClientsStore` to store in-memory handles to all currently connected clients. - pub(crate) fn new(stats_event_sender: StatsEventSender) -> Self { + pub(crate) fn new(stats_event_reporter: GatewayStatsReporter) -> Self { ActiveClientsStore { inner: Arc::new(DashMap::new()), - stats_event_sender, + stats_event_reporter, } } @@ -130,12 +130,8 @@ impl ActiveClientsStore { /// * `client`: address of the client for which to remove the handle. pub(crate) fn disconnect(&self, client: DestinationAddressBytes) { self.inner.remove(&client); - if let Err(e) = self - .stats_event_sender - .unbounded_send(events::StatsEvent::new_session_stop(client)) - { - warn!("Failed to send session stop event to collector : {e}") - }; + self.stats_event_reporter + .report(gateways::GatewayStatsEvent::new_session_stop(client)); } /// Insert new client handle into the store. @@ -157,12 +153,8 @@ impl ActiveClientsStore { if self.inner.insert(client, entry).is_some() { panic!("inserted a duplicate remote client") } - if let Err(e) = self - .stats_event_sender - .unbounded_send(events::StatsEvent::new_session_start(client)) - { - warn!("Failed to send session start event to collector : {e}") - }; + self.stats_event_reporter + .report(gateways::GatewayStatsEvent::new_session_start(client)); } /// Inserts a handle to the embedded client diff --git a/gateway/src/node/client_handling/websocket/common_state.rs b/gateway/src/node/client_handling/websocket/common_state.rs index a07f0e768d..ec0a3eb1d1 100644 --- a/gateway/src/node/client_handling/websocket/common_state.rs +++ b/gateway/src/node/client_handling/websocket/common_state.rs @@ -3,7 +3,7 @@ use nym_credential_verification::{ecash::EcashManager, BandwidthFlushingBehaviourConfig}; use nym_crypto::asymmetric::identity; -use nym_statistics_common::events::StatsEventSender; +use nym_statistics_common::gateways::GatewayStatsReporter; use std::sync::Arc; // I can see this being possible expanded with say storage or client store @@ -14,5 +14,5 @@ pub(crate) struct CommonHandlerState { pub(crate) local_identity: Arc, pub(crate) only_coconut_credentials: bool, pub(crate) bandwidth_cfg: BandwidthFlushingBehaviourConfig, - pub(crate) stats_event_sender: StatsEventSender, + pub(crate) stats_event_reporter: GatewayStatsReporter, } diff --git a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs index 80d99a8890..1b60fb6956 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs @@ -26,7 +26,7 @@ use nym_gateway_requests::{ }; use nym_gateway_storage::{error::StorageError, Storage}; use nym_sphinx::forwarding::packet::MixPacket; -use nym_statistics_common::events; +use nym_statistics_common::gateways; use nym_task::TaskClient; use nym_validator_client::coconut::EcashApiError; use rand::{random, CryptoRng, Rng}; @@ -259,11 +259,9 @@ where trace!("available total bandwidth: {available_total}"); if let Ok(ticket_type) = maybe_ticket_type { - if let Err(e) = self.inner.shared_state.stats_event_sender.unbounded_send( - events::StatsEvent::new_ecash_ticket(self.client.address, ticket_type), - ) { - warn!("Failed to send session stop event to collector : {e}") - }; + self.inner.shared_state.stats_event_reporter.report( + gateways::GatewayStatsEvent::new_ecash_ticket(self.client.address, ticket_type), + ); } else { error!("Somehow verified a ticket with an unknown ticket type"); } diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index b5ea09c2c6..e01683d2ad 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -27,7 +27,7 @@ use nym_mixnet_client::forwarder::{MixForwardingSender, PacketForwarder}; use nym_network_defaults::NymNetworkDetails; use nym_network_requester::{LocalGateway, NRServiceProviderBuilder, RequestFilter}; use nym_node_http_api::state::metrics::SharedSessionStats; -use nym_statistics_common::events::{self, StatsEventSender}; +use nym_statistics_common::gateways::{self, GatewayStatsReporter}; use nym_task::{TaskClient, TaskHandle, TaskManager}; use nym_topology::NetworkAddress; use nym_types::gateway::GatewayNodeDetailsResponse; @@ -415,7 +415,7 @@ impl Gateway { active_clients_store: ActiveClientsStore, shutdown: TaskClient, ecash_verifier: Arc>, - stats_event_sender: StatsEventSender, + stats_event_reporter: GatewayStatsReporter, ) where St: Storage + Send + Sync + Clone + 'static, { @@ -432,7 +432,7 @@ impl Gateway { local_identity: Arc::clone(&self.identity_keypair), only_coconut_credentials: self.config.gateway.only_coconut_credentials, bandwidth_cfg: (&self.config).into(), - stats_event_sender, + stats_event_reporter, }; websocket::Listener::new(listening_address, shared_state).start( @@ -462,7 +462,7 @@ impl Gateway { &self, shared_session_stats: SharedSessionStats, shutdown: TaskClient, - ) -> events::StatsEventSender { + ) -> gateways::GatewayStatsReporter { info!("Starting gateway stats collector..."); let (mut stats_collector, stats_event_sender) = diff --git a/gateway/src/node/statistics/mod.rs b/gateway/src/node/statistics/mod.rs index 6cc943cba7..8b60c934a7 100644 --- a/gateway/src/node/statistics/mod.rs +++ b/gateway/src/node/statistics/mod.rs @@ -1,10 +1,11 @@ // Copyright 2022 - Nym Technologies SA // 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_statistics_common::gateways::{ + GatewayStatsEvent, GatewayStatsReceiver, GatewayStatsReporter, +}; use nym_task::TaskClient; use sessions::SessionStatsHandler; use std::time::Duration; @@ -16,7 +17,7 @@ pub mod sessions; const STATISTICS_UPDATE_TIMER_INTERVAL: Duration = Duration::from_secs(3600); //update timer, no need to check everytime pub(crate) struct GatewayStatisticsCollector { - stats_event_rx: StatsEventReceiver, + stats_event_rx: GatewayStatsReceiver, session_stats: SessionStatsHandler, //here goes additionnal stats handler } @@ -25,15 +26,16 @@ impl GatewayStatisticsCollector { pub fn new( shared_session_stats: SharedSessionStats, stats_storage: PersistentStatsStorage, - ) -> (GatewayStatisticsCollector, StatsEventSender) { - let (stats_event_tx, stats_event_rx) = mpsc::unbounded(); + ) -> (GatewayStatisticsCollector, GatewayStatsReporter) { + let (stats_event_tx, stats_event_rx) = tokio::sync::mpsc::unbounded_channel(); let session_stats = SessionStatsHandler::new(shared_session_stats, stats_storage); let collector = GatewayStatisticsCollector { stats_event_rx, session_stats, }; - (collector, stats_event_tx) + let reporter = GatewayStatsReporter::new(stats_event_tx); + (collector, reporter) } async fn update_shared_state(&mut self, update_time: OffsetDateTime) { @@ -68,10 +70,10 @@ impl GatewayStatisticsCollector { self.update_shared_state(now).await; }, - Some(stat_event) = self.stats_event_rx.next() => { + Some(stat_event) = self.stats_event_rx.recv() => { //dispatching event to proper handler match stat_event { - StatsEvent::SessionStatsEvent(event) => { + GatewayStatsEvent::SessionStatsEvent(event) => { if let Err(e) = self.session_stats.handle_event(event).await{ warn!("Session event handling error - {e}"); }}, diff --git a/gateway/src/node/statistics/sessions.rs b/gateway/src/node/statistics/sessions.rs index e9394f203f..eca940a21d 100644 --- a/gateway/src/node/statistics/sessions.rs +++ b/gateway/src/node/statistics/sessions.rs @@ -10,7 +10,7 @@ use nym_sphinx::DestinationAddressBytes; use sha2::{Digest, Sha256}; use time::{Date, Duration, OffsetDateTime}; -use nym_statistics_common::events::SessionEvent; +use nym_statistics_common::gateways::SessionEvent; pub(crate) struct SessionStatsHandler { storage: PersistentStatsStorage, diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index 8ed4845e9a..6d8f360eff 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -23,6 +23,7 @@ nym-credential-storage = { path = "../../../common/credential-storage" } nym-credential-utils = { path = "../../../common/credential-utils" } nym-network-defaults = { path = "../../../common/network-defaults" } nym-sphinx = { path = "../../../common/nymsphinx" } +nym-statistics-common = { path = "../../../common/statistics" } nym-task = { path = "../../../common/task" } nym-topology = { path = "../../../common/topology" } nym-socks5-client-core = { path = "../../../common/socks5-client-core" } diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index edf299904d..7a70bf5b6c 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -54,6 +54,7 @@ pub struct MixnetClientBuilder { custom_shutdown: Option, force_tls: bool, user_agent: Option, + stats_reporting_config: Option, // TODO: incorporate it properly into `MixnetClientStorage` (I will need it in wasm anyway) gateway_endpoint_config_path: Option, @@ -93,6 +94,7 @@ impl MixnetClientBuilder { custom_gateway_transceiver: None, force_tls: false, user_agent: None, + stats_reporting_config: None, }) } } @@ -120,6 +122,7 @@ where custom_shutdown: None, force_tls: false, user_agent: None, + stats_reporting_config: None, gateway_endpoint_config_path: None, storage, } @@ -138,6 +141,7 @@ where custom_shutdown: self.custom_shutdown, force_tls: self.force_tls, user_agent: self.user_agent, + stats_reporting_config: self.stats_reporting_config, gateway_endpoint_config_path: self.gateway_endpoint_config_path, storage, } @@ -231,6 +235,15 @@ where self } + #[must_use] + pub fn with_statistics_reporting( + mut self, + config: nym_statistics_common::StatsReportingConfig, + ) -> Self { + self.stats_reporting_config = Some(config); + self + } + /// Use custom mixnet sender that might not be the default websocket gateway connection. /// only for advanced use #[must_use] @@ -259,6 +272,7 @@ where client.wait_for_gateway = self.wait_for_gateway; client.force_tls = self.force_tls; client.user_agent = self.user_agent; + client.stats_reporting_config = self.stats_reporting_config; Ok(client) } @@ -308,6 +322,8 @@ where custom_shutdown: Option, user_agent: Option, + + stats_reporting_config: Option, } impl DisconnectedMixnetClient @@ -357,6 +373,7 @@ where force_tls: false, custom_shutdown: None, user_agent: None, + stats_reporting_config: None, }) } @@ -577,6 +594,10 @@ where base_builder = base_builder.with_user_agent(user_agent); } + if let Some(stats_reporting_config) = self.stats_reporting_config { + base_builder = base_builder.with_statistics_reporting(stats_reporting_config); + } + if let Some(topology_provider) = self.custom_topology_provider { base_builder = base_builder.with_topology_provider(topology_provider); }