feature: wireguard metrics (#5278)

* experimental log

* introduce wireguard metrics updates

* add wireguard traffic rates to console logger

* missing import

* changed order of displayed values

* expose bytes information via rest endpoint

* clippy

Cargo.lock

@@ -6804,6 +6804,7 @@ dependencies = [
  "nym-crypto",
  "nym-gateway-storage",
  "nym-network-defaults",
+ "nym-node-metrics",
  "nym-task",
  "nym-wireguard-types",
  "thiserror",

common/wireguard/Cargo.toml

@@ -36,3 +36,4 @@ nym-gateway-storage = { path = "../gateway-storage" }
 nym-network-defaults = { path = "../network-defaults" }
 nym-task = { path = "../task" }
 nym-wireguard-types = { path = "../wireguard-types" }
+nym-node-metrics = { path = "../../nym-node/nym-node-metrics" }

common/wireguard/src/lib.rs

@@ -85,6 +85,7 @@ pub struct WireguardData {
 #[cfg(target_os = "linux")]
 pub async fn start_wireguard(
     storage: nym_gateway_storage::GatewayStorage,
+    metrics: nym_node_metrics::NymNodeMetrics,
     all_peers: Vec<nym_gateway_storage::models::WireguardPeer>,
     task_client: nym_task::TaskClient,
     wireguard_data: WireguardData,
@@ -175,6 +176,7 @@ pub async fn start_wireguard(
     let wg_api = std::sync::Arc::new(WgApiWrapper::new(wg_api));
     let mut controller = PeerController::new(
         storage,
+        metrics,
         wg_api.clone(),
         host,
         peer_bandwidth_managers,

common/wireguard/src/peer_controller.rs

@@ -16,7 +16,9 @@ use nym_credential_verification::{
     ClientBandwidth,
 };
 use nym_gateway_storage::GatewayStorage;
+use nym_node_metrics::NymNodeMetrics;
 use nym_wireguard_types::DEFAULT_PEER_TIMEOUT_CHECK;
+use std::time::{Duration, SystemTime};
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::{mpsc, RwLock};
 use tokio_stream::{wrappers::IntervalStream, StreamExt};
@@ -65,6 +67,11 @@ pub struct QueryBandwidthControlResponse {
 
 pub struct PeerController {
     storage: GatewayStorage,
+
+    // we have "all" metrics of a node, but they're behind a single Arc pointer,
+    // so the overhead is minimal
+    metrics: NymNodeMetrics,
+
     // used to receive commands from individual handles too
     request_tx: mpsc::Sender<PeerControlRequest>,
     request_rx: mpsc::Receiver<PeerControlRequest>,
@@ -76,8 +83,10 @@ pub struct PeerController {
 }
 
 impl PeerController {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         storage: GatewayStorage,
+        metrics: NymNodeMetrics,
         wg_api: Arc<WgApiWrapper>,
         initial_host_information: Host,
         bw_storage_managers: HashMap<Key, (Option<SharedBandwidthStorageManager>, Peer)>,
@@ -123,6 +132,7 @@ impl PeerController {
             request_rx,
             timeout_check_interval,
             task_client,
+            metrics,
         }
     }
 
@@ -257,6 +267,46 @@ impl PeerController {
         }))
     }
 
+    fn update_metrics(&self, new_host: &Host) {
+        let now = SystemTime::now();
+        const ACTIVITY_THRESHOLD: Duration = Duration::from_secs(60);
+
+        let total_peers = new_host.peers.len();
+        let mut active_peers = 0;
+        let mut total_rx = 0;
+        let mut total_tx = 0;
+
+        for peer in new_host.peers.values() {
+            total_rx += peer.rx_bytes;
+            total_tx += peer.tx_bytes;
+
+            // if a peer hasn't performed a handshake in last minute,
+            // I think it's reasonable to assume it's no longer active
+            let Some(last_handshake) = peer.last_handshake else {
+                continue;
+            };
+            let Ok(elapsed) = now.duration_since(last_handshake) else {
+                continue;
+            };
+            if elapsed < ACTIVITY_THRESHOLD {
+                active_peers += 1;
+            }
+        }
+
+        self.metrics.wireguard.update(
+            // if the conversion fails it means we're running not running on a 64bit system
+            // and that's a reason enough for this failure.
+            total_rx.try_into().expect(
+                "failed to convert bytes from u64 to usize - are you running on non 64bit system?",
+            ),
+            total_tx.try_into().expect(
+                "failed to convert bytes from u64 to usize - are you running on non 64bit system?",
+            ),
+            total_peers,
+            active_peers,
+        );
+    }
+
     pub async fn run(&mut self) {
         info!("started wireguard peer controller");
         loop {
@@ -266,6 +316,8 @@ impl PeerController {
                         log::error!("Can't read wireguard kernel data");
                         continue;
                     };
+                    self.update_metrics(&host);
+
                     *self.host_information.write().await = host;
                 }
                 _ = self.task_client.recv() => {

gateway/src/node/mod.rs

@@ -37,6 +37,7 @@ mod internal_service_providers;
 pub use client_handling::active_clients::ActiveClientsStore;
 pub use nym_gateway_stats_storage::PersistentStatsStorage;
 pub use nym_gateway_storage::{error::GatewayStorageError, GatewayStorage};
+use nym_node_metrics::NymNodeMetrics;
 pub use nym_sdk::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent};
 
 #[derive(Debug, Clone)]
@@ -81,6 +82,8 @@ pub struct GatewayTasksBuilder {
 
     metrics_sender: MetricEventsSender,
 
+    metrics: NymNodeMetrics,
+
     mnemonic: Arc<Zeroizing<bip39::Mnemonic>>,
 
     shutdown: TaskClient,
@@ -102,12 +105,14 @@ impl Drop for GatewayTasksBuilder {
 }
 
 impl GatewayTasksBuilder {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         config: Config,
         identity: Arc<ed25519::KeyPair>,
         storage: GatewayStorage,
         mix_packet_sender: MixForwardingSender,
         metrics_sender: MetricEventsSender,
+        metrics: NymNodeMetrics,
         mnemonic: Arc<Zeroizing<bip39::Mnemonic>>,
         shutdown: TaskClient,
     ) -> GatewayTasksBuilder {
@@ -121,6 +126,7 @@ impl GatewayTasksBuilder {
             storage,
             mix_packet_sender,
             metrics_sender,
+            metrics,
             mnemonic,
             shutdown,
             ecash_manager: None,
@@ -443,6 +449,7 @@ impl GatewayTasksBuilder {
     pub async fn try_start_wireguard(
         &mut self,
     ) -> Result<Arc<nym_wireguard::WgApiWrapper>, Box<dyn std::error::Error + Send + Sync>> {
+        let _ = self.metrics.clone();
         unimplemented!("wireguard is not supported on this platform")
     }
 
@@ -460,6 +467,7 @@ impl GatewayTasksBuilder {
 
         let wg_handle = nym_wireguard::start_wireguard(
             self.storage.clone(),
+            self.metrics.clone(),
             all_peers,
             self.shutdown.fork("wireguard"),
             wireguard_data,

nym-node/nym-node-metrics/src/lib.rs

@@ -4,6 +4,7 @@
 use crate::entry::EntryStats;
 use crate::mixnet::MixingStats;
 use crate::network::NetworkStats;
+use crate::wireguard::WireguardStats;
 use std::ops::Deref;
 use std::sync::Arc;
 
@@ -11,6 +12,7 @@ pub mod entry;
 pub mod events;
 pub mod mixnet;
 pub mod network;
+pub mod wireguard;
 
 #[derive(Clone, Default)]
 pub struct NymNodeMetrics {
@@ -34,6 +36,7 @@ impl Deref for NymNodeMetrics {
 pub struct NymNodeMetricsInner {
     pub mixnet: MixingStats,
     pub entry: EntryStats,
+    pub wireguard: WireguardStats,
 
     pub network: NetworkStats,
 }

nym-node/nym-node-metrics/src/wireguard.rs

@@ -0,0 +1,44 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[derive(Default)]
+pub struct WireguardStats {
+    bytes_rx: AtomicUsize,
+    bytes_tx: AtomicUsize,
+
+    total_peers: AtomicUsize,
+    active_peers: AtomicUsize,
+}
+
+impl WireguardStats {
+    pub fn bytes_rx(&self) -> usize {
+        self.bytes_rx.load(Ordering::Relaxed)
+    }
+
+    pub fn bytes_tx(&self) -> usize {
+        self.bytes_tx.load(Ordering::Relaxed)
+    }
+
+    pub fn total_peers(&self) -> usize {
+        self.total_peers.load(Ordering::Relaxed)
+    }
+
+    pub fn active_peers(&self) -> usize {
+        self.active_peers.load(Ordering::Relaxed)
+    }
+
+    pub fn update(
+        &self,
+        bytes_rx: usize,
+        bytes_tx: usize,
+        total_peers: usize,
+        active_peers: usize,
+    ) {
+        self.bytes_rx.store(bytes_rx, Ordering::Relaxed);
+        self.bytes_tx.store(bytes_tx, Ordering::Relaxed);
+        self.total_peers.store(total_peers, Ordering::Relaxed);
+        self.active_peers.store(active_peers, Ordering::Relaxed);
+    }
+}

nym-node/nym-node-requests/src/api/v1/metrics/models.rs

@@ -4,6 +4,19 @@
 pub use mixing::*;
 pub use session::*;
 pub use verloc::*;
+pub use wireguard::*;
+
+pub mod wireguard {
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+    #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+    pub struct WireguardStats {
+        pub bytes_tx: usize,
+
+        pub bytes_rx: usize,
+    }
+}
 
 pub mod packets {
     use serde::{Deserialize, Serialize};

nym-node/nym-node-requests/src/lib.rs

@@ -66,12 +66,18 @@ pub mod routes {
 
                 pub const LEGACY_MIXING: &str = "/mixing";
                 pub const PACKETS_STATS: &str = "/packets-stats";
+                pub const WIREGUARD_STATS: &str = "/wireguard-stats";
                 pub const SESSIONS: &str = "/sessions";
                 pub const VERLOC: &str = "/verloc";
                 pub const PROMETHEUS: &str = "/prometheus";
 
                 absolute_route!(legacy_mixing_absolute, metrics_absolute(), LEGACY_MIXING);
                 absolute_route!(packets_stats_absolute, metrics_absolute(), PACKETS_STATS);
+                absolute_route!(
+                    wireguard_stats_absolute,
+                    metrics_absolute(),
+                    WIREGUARD_STATS
+                );
                 absolute_route!(sessions_absolute, metrics_absolute(), SESSIONS);
                 absolute_route!(verloc_absolute, metrics_absolute(), VERLOC);
                 absolute_route!(prometheus_absolute, metrics_absolute(), PROMETHEUS);

nym-node/src/node/http/router/api/v1/metrics/mod.rs

@@ -5,6 +5,7 @@ use crate::node::http::api::v1::metrics::packets_stats::packets_stats;
 use crate::node::http::api::v1::metrics::prometheus::prometheus_metrics;
 use crate::node::http::api::v1::metrics::sessions::sessions_stats;
 use crate::node::http::api::v1::metrics::verloc::verloc_stats;
+use crate::node::http::api::v1::metrics::wireguard::wireguard_stats;
 use crate::node::http::state::metrics::MetricsAppState;
 use axum::extract::FromRef;
 use axum::routing::get;
@@ -16,6 +17,7 @@ pub mod packets_stats;
 pub mod prometheus;
 pub mod sessions;
 pub mod verloc;
+pub mod wireguard;
 
 #[derive(Debug, Clone, Default)]
 pub struct Config {
@@ -34,6 +36,7 @@ where
             get(legacy_mixing::legacy_mixing_stats),
         )
         .route(metrics::PACKETS_STATS, get(packets_stats))
+        .route(metrics::WIREGUARD_STATS, get(wireguard_stats))
         .route(metrics::SESSIONS, get(sessions_stats))
         .route(metrics::VERLOC, get(verloc_stats))
         .route(metrics::PROMETHEUS, get(prometheus_metrics))

nym-node/src/node/http/router/api/v1/metrics/packets_stats.rs

@@ -1,5 +1,5 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only
 
 use crate::node::http::state::metrics::MetricsAppState;
 use axum::extract::{Query, State};

nym-node/src/node/http/router/api/v1/metrics/wireguard.rs

@@ -0,0 +1,40 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::node::http::state::metrics::MetricsAppState;
+use axum::extract::{Query, State};
+use nym_http_api_common::{FormattedResponse, OutputParams};
+use nym_node_metrics::NymNodeMetrics;
+use nym_node_requests::api::v1::metrics::models::WireguardStats;
+
+/// If applicable, returns wireguard statistics information of this node.
+/// This information is **PURELY** self-reported and in no way validated.
+#[utoipa::path(
+    get,
+    path = "/wireguard-stats",
+    context_path = "/api/v1/metrics",
+    tag = "Metrics",
+    responses(
+        (status = 200, content(
+            ("application/json" = WireguardStats),
+            ("application/yaml" = WireguardStats)
+        ))
+    ),
+    params(OutputParams),
+)]
+pub(crate) async fn wireguard_stats(
+    Query(output): Query<OutputParams>,
+    State(metrics_state): State<MetricsAppState>,
+) -> WireguardStatsResponse {
+    let output = output.output.unwrap_or_default();
+    output.to_response(build_response(&metrics_state.metrics))
+}
+
+fn build_response(metrics: &NymNodeMetrics) -> WireguardStats {
+    WireguardStats {
+        bytes_tx: metrics.wireguard.bytes_tx(),
+        bytes_rx: metrics.wireguard.bytes_rx(),
+    }
+}
+
+pub type WireguardStatsResponse = FormattedResponse<WireguardStats>;

nym-node/src/node/metrics/console_logger.rs

@@ -25,6 +25,9 @@ struct AtLastUpdate {
 
     // EGRESS
     ack_packets_sent: usize,
+
+    wg_tx: usize,
+    wg_rx: usize,
 }
 
 impl AtLastUpdate {
@@ -35,6 +38,8 @@ impl AtLastUpdate {
             final_hop_packets_received: 0,
             forward_hop_packets_sent: 0,
             ack_packets_sent: 0,
+            wg_tx: 0,
+            wg_rx: 0,
         }
     }
 }
@@ -70,6 +75,9 @@ impl ConsoleLogger {
         let forward_sent = self.metrics.mixnet.egress.forward_hop_packets_sent();
         let acks = self.metrics.mixnet.egress.ack_packets_sent();
 
+        let wg_tx = self.metrics.wireguard.bytes_tx();
+        let wg_rx = self.metrics.wireguard.bytes_rx();
+
         let forward_received_rate =
             (forward_received - self.at_last_update.forward_hop_packets_received) as f64
                 / delta_secs;
@@ -79,6 +87,9 @@ impl ConsoleLogger {
             (forward_sent - self.at_last_update.forward_hop_packets_sent) as f64 / delta_secs;
         let acks_rate = (acks - self.at_last_update.ack_packets_sent) as f64 / delta_secs;
 
+        let wg_tx_rate = (wg_tx - self.at_last_update.wg_tx) as f64 / delta_secs;
+        let wg_rx_rate = (wg_rx - self.at_last_update.wg_rx) as f64 / delta_secs;
+
         info!("↑↓ Packets sent [total] / sent [acks] / received [mix] / received [gw]: {} ({}) / {} ({}) / {} ({}) / {} ({})",
             forward_sent.human_count_bare(),
             forward_sent_rate.human_throughput_bare(),
@@ -90,11 +101,24 @@ impl ConsoleLogger {
             final_rate.human_throughput_bare(),
         );
 
+        // only log wireguard if we have transmitted ANY bytes
+        if self.at_last_update.wg_rx != 0 {
+            info!(
+                "↑↓ Wireguard tx/rx: {} ({}) / {} ({})",
+                wg_tx.human_count_bytes(),
+                wg_tx_rate.human_throughput_bytes(),
+                wg_rx.human_count_bytes(),
+                wg_rx_rate.human_throughput_bytes()
+            )
+        }
+
         self.at_last_update.time = now;
         self.at_last_update.forward_hop_packets_received = forward_received;
         self.at_last_update.final_hop_packets_received = final_received;
         self.at_last_update.forward_hop_packets_sent = forward_sent;
         self.at_last_update.ack_packets_sent = acks;
+        self.at_last_update.wg_tx = wg_tx;
+        self.at_last_update.wg_rx = wg_rx;
 
         // TODO: add websocket-client traffic
     }

nym-node/src/node/mod.rs

@@ -589,6 +589,7 @@ impl NymNode {
             self.entry_gateway.client_storage.clone(),
             mix_packet_sender,
             metrics_sender,
+            self.metrics.clone(),
             self.entry_gateway.mnemonic.clone(),
             task_client,
         );