feature: expand nym-node prometheus metrics (#5298)

* fixed bearer auth for prometheus route

* basic prometheus metrics

* added rates on global values

* improved structure on the prometheus metrics

* added additional metrics for ingress websockets and egress mixnet connections

* some channel business metrics

* fixed metrics registration and added additional variants

* added counter for number of disk persisted packets

* counter for pending egress packets

* counter for pending egress forward packets

* clippy

nym-node/src/node/http/error.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 std::io;
 use std::net::SocketAddr;

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

@@ -10,7 +10,12 @@ use crate::node::http::state::metrics::MetricsAppState;
 use axum::extract::FromRef;
 use axum::routing::get;
 use axum::Router;
+use nym_http_api_common::middleware::bearer_auth::AuthLayer;
 use nym_node_requests::routes::api::v1::metrics;
+use nym_node_requests::routes::api::v1::metrics::prometheus_absolute;
+use std::sync::Arc;
+use tracing::info;
+use zeroize::Zeroizing;
 
 pub mod legacy_mixing;
 pub mod packets_stats;
@@ -21,16 +26,23 @@ pub mod wireguard;
 
 #[derive(Debug, Clone, Default)]
 pub struct Config {
-    //
+    pub bearer_token: Option<Arc<Zeroizing<String>>>,
 }
 
 #[allow(deprecated)]
-pub(super) fn routes<S>(_config: Config) -> Router<S>
+pub(super) fn routes<S>(config: Config) -> Router<S>
 where
     S: Send + Sync + 'static + Clone,
     MetricsAppState: FromRef<S>,
 {
-    Router::new()
+    if config.bearer_token.is_none() {
+        info!(
+            "bearer token hasn't been set. '{}' route will not be exposed",
+            prometheus_absolute()
+        )
+    }
+
+    let router = Router::new()
         .route(
             metrics::LEGACY_MIXING,
             get(legacy_mixing::legacy_mixing_stats),
@@ -38,6 +50,17 @@ where
         .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))
+        .route(metrics::VERLOC, get(verloc_stats));
+
+    let auth_middleware = config.bearer_token.map(AuthLayer::new);
+
+    // don't expose prometheus route without bearer token set
+    if let Some(auth_middleware) = auth_middleware {
+        router.route(
+            metrics::PROMETHEUS,
+            get(prometheus_metrics).route_layer(auth_middleware),
+        )
+    } else {
+        router
+    }
 }

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

@@ -1,12 +1,6 @@
 // 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::State;
-use axum::http::StatusCode;
-use axum_extra::TypedHeader;
-use headers::authorization::Bearer;
-use headers::Authorization;
 use nym_metrics::metrics;
 
 /// Returns `prometheus` compatible metrics
@@ -25,21 +19,7 @@ use nym_metrics::metrics;
         ("prometheus_token" = [])
     )
 )]
-pub(crate) async fn prometheus_metrics<'a>(
-    TypedHeader(authorization): TypedHeader<Authorization<Bearer>>,
-    State(state): State<MetricsAppState>,
-) -> Result<String, StatusCode> {
-    if authorization.token().is_empty() {
-        return Err(StatusCode::UNAUTHORIZED);
-    }
-    // TODO: is 500 the correct error code here?
-    let Some(metrics_key) = state.prometheus_access_token else {
-        return Err(StatusCode::INTERNAL_SERVER_ERROR);
-    };
-
-    if metrics_key != authorization.token() {
-        return Err(StatusCode::UNAUTHORIZED);
-    }
-
-    Ok(metrics!())
+// the AuthLayer is protecting access to this endpoint
+pub(crate) async fn prometheus_metrics() -> String {
+    metrics!()
 }

nym-node/src/node/http/router/mod.rs

@@ -20,6 +20,8 @@ use nym_node_requests::api::SignedHostInformation;
 use nym_node_requests::routes;
 use std::net::SocketAddr;
 use std::path::Path;
+use std::sync::Arc;
+use zeroize::Zeroizing;
 
 pub mod api;
 pub mod landing_page;
@@ -115,6 +117,11 @@ impl HttpServerConfig {
         self.api.v1_config.authenticator.details = Some(authenticator);
         self
     }
+
+    pub fn with_prometheus_bearer_token(mut self, bearer_token: Option<String>) -> Self {
+        self.api.v1_config.metrics.bearer_token = bearer_token.map(|b| Arc::new(Zeroizing::new(b)));
+        self
+    }
 }
 
 pub struct NymNodeRouter {

nym-node/src/node/http/state/metrics.rs

@@ -9,8 +9,6 @@ pub use nym_verloc::measurements::metrics::SharedVerlocStats;
 
 #[derive(Clone)]
 pub struct MetricsAppState {
-    pub(crate) prometheus_access_token: Option<String>,
-
     pub(crate) metrics: NymNodeMetrics,
 
     pub(crate) verloc: SharedVerlocStats,

nym-node/src/node/http/state/mod.rs

@@ -24,17 +24,7 @@ impl AppState {
             // does it have to be?
             // also no.
             startup_time: Instant::now(),
-            metrics: MetricsAppState {
-                prometheus_access_token: None,
-                metrics,
-                verloc,
-            },
+            metrics: MetricsAppState { metrics, verloc },
         }
     }
-
-    #[must_use]
-    pub fn with_metrics_key(mut self, bearer_token: impl Into<Option<String>>) -> Self {
-        self.metrics.prometheus_access_token = bearer_token.into();
-        self
-    }
 }

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

@@ -44,7 +44,7 @@ impl MetricsAggregator {
         self.event_sender.clone()
     }
 
-    pub fn register_handler<H>(&mut self, handler: H, update_interval: Duration)
+    pub fn register_handler<H>(&mut self, handler: H, update_interval: impl Into<Option<Duration>>)
     where
         H: MetricsHandler,
     {

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

@@ -1,2 +1,2 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
+// SPDX-License-Identifier: GPL-3.0-only

nym-node/src/node/metrics/handler/client_sessions.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::metrics::handler::{
     MetricsHandler, OnStartMetricsHandler, OnUpdateMetricsHandler,
@@ -10,6 +10,8 @@ use nym_gateway_stats_storage::error::StatsStorageError;
 use nym_gateway_stats_storage::models::{TicketType, ToSessionType};
 use nym_node_metrics::entry::{ActiveSession, ClientSessions, FinishedSession};
 use nym_node_metrics::events::GatewaySessionEvent;
+use nym_node_metrics::prometheus_wrapper::PrometheusMetric::EntryClientSessionsDurations;
+use nym_node_metrics::prometheus_wrapper::PROMETHEUS_METRICS;
 use nym_node_metrics::NymNodeMetrics;
 use nym_sphinx_types::DestinationAddressBytes;
 use time::{Date, Duration, OffsetDateTime};
@@ -53,6 +55,13 @@ impl GatewaySessionStatsHandler {
     ) -> Result<(), StatsStorageError> {
         if let Some(session) = self.storage.get_active_session(client).await? {
             if let Some(finished_session) = session.end_at(stop_time) {
+                PROMETHEUS_METRICS.observe_histogram(
+                    EntryClientSessionsDurations {
+                        typ: finished_session.typ.to_string(),
+                    },
+                    finished_session.duration.as_secs_f64(),
+                );
+
                 self.storage
                     .insert_finished_session(self.current_day, finished_session)
                     .await?;

nym-node/src/node/metrics/handler/global_prometheus_updater/at_last_update.rs

@@ -0,0 +1,219 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use nym_node_metrics::mixnet::{EgressMixingStats, IngressMixingStats, MixingStats};
+use nym_node_metrics::wireguard::WireguardStats;
+use nym_node_metrics::NymNodeMetrics;
+use time::OffsetDateTime;
+
+// used to calculate traffic rates
+#[derive(Debug)]
+pub(crate) struct AtLastUpdate {
+    time: OffsetDateTime,
+
+    mixnet: LastMixnet,
+    wireguard: LastWireguard,
+}
+
+impl AtLastUpdate {
+    pub(crate) fn is_initial(&self) -> bool {
+        self.time == OffsetDateTime::UNIX_EPOCH
+    }
+
+    pub(crate) fn rates(&self, previous: &Self) -> RateSinceUpdate {
+        let delta_secs = (self.time - previous.time).as_seconds_f64();
+
+        RateSinceUpdate {
+            mixnet: self.mixnet.rates(&previous.mixnet, delta_secs),
+            wireguard: self.wireguard.rates(&previous.wireguard, delta_secs),
+        }
+    }
+}
+
+impl Default for AtLastUpdate {
+    fn default() -> Self {
+        AtLastUpdate {
+            time: OffsetDateTime::now_utc(),
+            mixnet: Default::default(),
+            wireguard: Default::default(),
+        }
+    }
+}
+
+impl From<&NymNodeMetrics> for AtLastUpdate {
+    fn from(metrics: &NymNodeMetrics) -> Self {
+        AtLastUpdate {
+            time: OffsetDateTime::now_utc(),
+            mixnet: (&metrics.mixnet).into(),
+            wireguard: (&metrics.wireguard).into(),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct LastMixnet {
+    ingres: LastMixnetIngress,
+    egress: LastMixnetEgress,
+}
+
+impl LastMixnet {
+    fn rates(&self, previous: &Self, time_delta_secs: f64) -> MixnetRateSinceUpdate {
+        MixnetRateSinceUpdate {
+            ingress: self.ingres.rates(&previous.ingres, time_delta_secs),
+            egress: self.egress.rates(&previous.egress, time_delta_secs),
+        }
+    }
+}
+
+impl From<&MixingStats> for LastMixnet {
+    fn from(value: &MixingStats) -> Self {
+        LastMixnet {
+            ingres: (&value.ingress).into(),
+            egress: (&value.egress).into(),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct LastMixnetIngress {
+    forward_hop_packets_received: usize,
+    final_hop_packets_received: usize,
+    malformed_packets_received: usize,
+    excessive_delay_packets: usize,
+    forward_hop_packets_dropped: usize,
+    final_hop_packets_dropped: usize,
+}
+
+impl LastMixnetIngress {
+    fn rates(&self, previous: &Self, time_delta_secs: f64) -> MixnetIngressRateSinceUpdate {
+        let forward_hop_packets_received_delta =
+            self.forward_hop_packets_received - previous.forward_hop_packets_received;
+        let final_hop_packets_received_delta =
+            self.final_hop_packets_received - previous.final_hop_packets_received;
+        let malformed_packets_received_delta =
+            self.malformed_packets_received - previous.malformed_packets_received;
+        let excessive_delay_packets_delta =
+            self.excessive_delay_packets - previous.excessive_delay_packets;
+        let forward_hop_packets_dropped_delta =
+            self.forward_hop_packets_dropped - previous.forward_hop_packets_dropped;
+        let final_hop_packets_dropped_delta =
+            self.final_hop_packets_dropped - previous.final_hop_packets_dropped;
+
+        MixnetIngressRateSinceUpdate {
+            forward_hop_packets_received_sec: forward_hop_packets_received_delta as f64
+                / time_delta_secs,
+            final_hop_packets_received_sec: final_hop_packets_received_delta as f64
+                / time_delta_secs,
+            malformed_packets_received_sec: malformed_packets_received_delta as f64
+                / time_delta_secs,
+            excessive_delay_packets_sec: excessive_delay_packets_delta as f64 / time_delta_secs,
+            forward_hop_packets_dropped_sec: forward_hop_packets_dropped_delta as f64
+                / time_delta_secs,
+            final_hop_packets_dropped_sec: final_hop_packets_dropped_delta as f64 / time_delta_secs,
+        }
+    }
+}
+
+impl From<&IngressMixingStats> for LastMixnetIngress {
+    fn from(value: &IngressMixingStats) -> Self {
+        LastMixnetIngress {
+            forward_hop_packets_received: value.forward_hop_packets_received(),
+            final_hop_packets_received: value.final_hop_packets_received(),
+            malformed_packets_received: value.malformed_packets_received(),
+            excessive_delay_packets: value.excessive_delay_packets(),
+            forward_hop_packets_dropped: value.forward_hop_packets_dropped(),
+            final_hop_packets_dropped: value.final_hop_packets_dropped(),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct LastMixnetEgress {
+    forward_hop_packets_sent: usize,
+    ack_packets_sent: usize,
+    forward_hop_packets_dropped: usize,
+}
+
+impl LastMixnetEgress {
+    fn rates(&self, previous: &Self, time_delta_secs: f64) -> MixnetEgressRateSinceUpdate {
+        let forward_hop_packets_sent_delta =
+            self.forward_hop_packets_sent - previous.forward_hop_packets_sent;
+        let ack_packets_sent_delta = self.ack_packets_sent - previous.ack_packets_sent;
+        let forward_hop_packets_dropped_delta =
+            self.forward_hop_packets_dropped - previous.forward_hop_packets_dropped;
+
+        MixnetEgressRateSinceUpdate {
+            forward_hop_packets_sent_sec: forward_hop_packets_sent_delta as f64 / time_delta_secs,
+            ack_packets_sent_sec: ack_packets_sent_delta as f64 / time_delta_secs,
+            forward_hop_packets_dropped_sec: forward_hop_packets_dropped_delta as f64
+                / time_delta_secs,
+        }
+    }
+}
+
+impl From<&EgressMixingStats> for LastMixnetEgress {
+    fn from(value: &EgressMixingStats) -> Self {
+        LastMixnetEgress {
+            forward_hop_packets_sent: value.forward_hop_packets_sent(),
+            ack_packets_sent: value.ack_packets_sent(),
+            forward_hop_packets_dropped: value.forward_hop_packets_dropped(),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct LastWireguard {
+    bytes_tx: usize,
+    bytes_rx: usize,
+}
+
+impl LastWireguard {
+    fn rates(&self, previous: &Self, time_delta_secs: f64) -> WireguardRateSinceUpdate {
+        let bytes_tx_delta = self.bytes_tx - previous.bytes_tx;
+        let bytes_rx_delta = self.bytes_rx - previous.bytes_rx;
+
+        WireguardRateSinceUpdate {
+            bytes_tx_sec: bytes_tx_delta as f64 / time_delta_secs,
+            bytes_rx_sec: bytes_rx_delta as f64 / time_delta_secs,
+        }
+    }
+}
+
+impl From<&WireguardStats> for LastWireguard {
+    fn from(value: &WireguardStats) -> Self {
+        LastWireguard {
+            bytes_tx: value.bytes_tx(),
+            bytes_rx: value.bytes_rx(),
+        }
+    }
+}
+
+pub(crate) struct RateSinceUpdate {
+    pub(crate) mixnet: MixnetRateSinceUpdate,
+    pub(crate) wireguard: WireguardRateSinceUpdate,
+}
+
+pub(crate) struct MixnetRateSinceUpdate {
+    pub(crate) ingress: MixnetIngressRateSinceUpdate,
+    pub(crate) egress: MixnetEgressRateSinceUpdate,
+}
+
+pub(crate) struct MixnetIngressRateSinceUpdate {
+    pub(crate) forward_hop_packets_received_sec: f64,
+    pub(crate) final_hop_packets_received_sec: f64,
+    pub(crate) malformed_packets_received_sec: f64,
+    pub(crate) excessive_delay_packets_sec: f64,
+    pub(crate) forward_hop_packets_dropped_sec: f64,
+    pub(crate) final_hop_packets_dropped_sec: f64,
+}
+
+pub(crate) struct MixnetEgressRateSinceUpdate {
+    pub(crate) forward_hop_packets_sent_sec: f64,
+    pub(crate) ack_packets_sent_sec: f64,
+    pub(crate) forward_hop_packets_dropped_sec: f64,
+}
+
+pub(crate) struct WireguardRateSinceUpdate {
+    pub(crate) bytes_tx_sec: f64,
+    pub(crate) bytes_rx_sec: f64,
+}

nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs

@@ -0,0 +1,223 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::node::metrics::handler::global_prometheus_updater::at_last_update::AtLastUpdate;
+use crate::node::metrics::handler::{
+    MetricsHandler, OnStartMetricsHandler, OnUpdateMetricsHandler,
+};
+use async_trait::async_trait;
+use nym_node_metrics::prometheus_wrapper::{
+    NymNodePrometheusMetrics, PrometheusMetric, PROMETHEUS_METRICS,
+};
+use nym_node_metrics::NymNodeMetrics;
+
+mod at_last_update;
+
+// it can be anything, we just need a unique type_id to register our handler
+pub struct GlobalPrometheusData;
+
+pub struct PrometheusGlobalNodeMetricsRegistryUpdater {
+    metrics: NymNodeMetrics,
+    prometheus_wrapper: &'static NymNodePrometheusMetrics,
+    at_last_update: AtLastUpdate,
+}
+
+impl PrometheusGlobalNodeMetricsRegistryUpdater {
+    pub(crate) fn new(metrics: NymNodeMetrics) -> Self {
+        Self {
+            metrics,
+            prometheus_wrapper: &PROMETHEUS_METRICS,
+            at_last_update: Default::default(),
+        }
+    }
+}
+
+#[async_trait]
+impl OnStartMetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater {}
+
+#[async_trait]
+impl OnUpdateMetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater {
+    async fn on_update(&mut self) {
+        let entry_guard = self.metrics.entry.client_sessions().await;
+        use PrometheusMetric::*;
+
+        // # MIXNET
+        // ## INGRESS
+        self.prometheus_wrapper.set(
+            MixnetIngressForwardPacketsReceived,
+            self.metrics.mixnet.ingress.forward_hop_packets_received() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetIngressFinalHopPacketsReceived,
+            self.metrics.mixnet.ingress.final_hop_packets_received() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetIngressMalformedPacketsReceived,
+            self.metrics.mixnet.ingress.malformed_packets_received() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetIngressExcessiveDelayPacketsReceived,
+            self.metrics.mixnet.ingress.excessive_delay_packets() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetEgressForwardPacketsDropped,
+            self.metrics.mixnet.ingress.forward_hop_packets_dropped() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetIngressFinalHopPacketsDropped,
+            self.metrics.mixnet.ingress.final_hop_packets_dropped() as i64,
+        );
+
+        // ## EGRESS
+        self.prometheus_wrapper.set(
+            MixnetEgressStoredOnDiskFinalHopPackets,
+            self.metrics.mixnet.egress.disk_persisted_packets() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetEgressForwardPacketsSent,
+            self.metrics.mixnet.egress.forward_hop_packets_sent() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetEgressAckSent,
+            self.metrics.mixnet.egress.ack_packets_sent() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetEgressForwardPacketsDropped,
+            self.metrics.mixnet.egress.forward_hop_packets_dropped() as i64,
+        );
+
+        // # ENTRY
+        self.prometheus_wrapper.set(
+            EntryClientUniqueUsers,
+            entry_guard.unique_users.len() as i64,
+        );
+        self.prometheus_wrapper.set(
+            EntryClientSessionsStarted,
+            entry_guard.sessions_started as i64,
+        );
+        self.prometheus_wrapper.set(
+            EntryClientSessionsFinished,
+            entry_guard.finished_sessions.len() as i64,
+        );
+
+        // # WIREGUARD
+        self.prometheus_wrapper
+            .set(WireguardBytesRx, self.metrics.wireguard.bytes_rx() as i64);
+        self.prometheus_wrapper
+            .set(WireguardBytesTx, self.metrics.wireguard.bytes_tx() as i64);
+        self.prometheus_wrapper.set(
+            WireguardTotalPeers,
+            self.metrics.wireguard.total_peers() as i64,
+        );
+        self.prometheus_wrapper.set(
+            WireguardActivePeers,
+            self.metrics.wireguard.active_peers() as i64,
+        );
+
+        // # NETWORK
+        self.prometheus_wrapper.set(
+            NetworkActiveIngressMixnetConnections,
+            self.metrics
+                .network
+                .active_ingress_mixnet_connections_count() as i64,
+        );
+        self.prometheus_wrapper.set(
+            NetworkActiveIngressWebSocketConnections,
+            self.metrics
+                .network
+                .active_ingress_websocket_connections_count() as i64,
+        );
+        self.prometheus_wrapper.set(
+            NetworkActiveIngressWebSocketConnections,
+            self.metrics
+                .network
+                .active_egress_mixnet_connections_count() as i64,
+        );
+
+        // # PROCESS
+        self.prometheus_wrapper.set(
+            ProcessForwardHopPacketsBeingDelayed,
+            self.metrics
+                .process
+                .forward_hop_packets_being_delayed_count() as i64,
+        );
+        self.prometheus_wrapper.set(
+            ProcessPacketForwarderQueueSize,
+            self.metrics.process.packet_forwarder_queue_size() as i64,
+        );
+        self.prometheus_wrapper.set(
+            ProcessFinalHopPacketsPendingDelivery,
+            self.metrics
+                .process
+                .final_hop_packets_pending_delivery_count() as i64,
+        );
+        self.prometheus_wrapper.set(
+            ProcessForwardHopPacketsPendingDelivery,
+            self.metrics
+                .process
+                .forward_hop_packets_pending_delivery_count() as i64,
+        );
+
+        let updated = AtLastUpdate::from(&self.metrics);
+
+        // # RATES
+        if !self.at_last_update.is_initial() {
+            let diff = updated.rates(&self.at_last_update);
+
+            self.prometheus_wrapper.set_float(
+                MixnetIngressForwardPacketsReceivedRate,
+                diff.mixnet.ingress.forward_hop_packets_received_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetIngressFinalHopPacketsReceivedRate,
+                diff.mixnet.ingress.final_hop_packets_received_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetIngressMalformedPacketsReceivedRate,
+                diff.mixnet.ingress.malformed_packets_received_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetIngressExcessiveDelayPacketsReceivedRate,
+                diff.mixnet.ingress.excessive_delay_packets_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetIngressForwardPacketsDroppedRate,
+                diff.mixnet.ingress.forward_hop_packets_dropped_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetIngressFinalHopPacketsDroppedRate,
+                diff.mixnet.ingress.final_hop_packets_dropped_sec,
+            );
+
+            // ## EGRESS
+            self.prometheus_wrapper.set_float(
+                MixnetEgressForwardPacketsSendRate,
+                diff.mixnet.egress.forward_hop_packets_sent_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetEgressAckSendRate,
+                diff.mixnet.egress.ack_packets_sent_sec,
+            );
+            self.prometheus_wrapper.set_float(
+                MixnetEgressForwardPacketsDroppedRate,
+                diff.mixnet.egress.forward_hop_packets_dropped_sec,
+            );
+
+            // # WIREGUARD
+            self.prometheus_wrapper
+                .set_float(WireguardBytesRxRate, diff.wireguard.bytes_rx_sec);
+            self.prometheus_wrapper
+                .set_float(WireguardBytesTxRate, diff.wireguard.bytes_tx_sec);
+        }
+        self.at_last_update = updated;
+    }
+}
+
+#[async_trait]
+impl MetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater {
+    type Events = GlobalPrometheusData;
+
+    async fn handle_event(&mut self, _event: Self::Events) {
+        panic!("this should have never been called! MetricsHandler has been incorrectly called on PrometheusNodeMetricsRegistryUpdater")
+    }
+}

nym-node/src/node/metrics/handler/legacy_packet_data.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::metrics::handler::{
     MetricsHandler, OnStartMetricsHandler, OnUpdateMetricsHandler,

nym-node/src/node/metrics/handler/mod.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 async_trait::async_trait;
 use std::any;
@@ -9,8 +9,11 @@ use tokio::time::Instant;
 use tracing::trace;
 
 pub(crate) mod client_sessions;
+pub(crate) mod global_prometheus_updater;
 pub(crate) mod legacy_packet_data;
 pub(crate) mod mixnet_data_cleaner;
+pub(crate) mod pending_egress_packets_updater;
+pub(crate) mod prometheus_events_handler;
 
 pub(crate) trait RegistrableHandler:
     Downcast + OnStartMetricsHandler + OnUpdateMetricsHandler + Send + Sync + 'static
@@ -63,23 +66,23 @@ pub(crate) trait OnStartMetricsHandler {
 
 #[async_trait]
 pub(crate) trait OnUpdateMetricsHandler {
-    async fn on_update(&mut self);
+    async fn on_update(&mut self) {}
 }
 
 pub(crate) struct HandlerWrapper<T> {
     handler: Box<dyn MetricsHandler<Events = T>>,
-    update_interval: Duration,
+    update_interval: Option<Duration>,
     last_updated: Instant,
 }
 
 impl<T> HandlerWrapper<T> {
-    pub fn new<U>(update_interval: Duration, handler: U) -> Self
+    pub fn new<U>(update_interval: impl Into<Option<Duration>>, handler: U) -> Self
     where
         U: MetricsHandler<Events = T>,
     {
         HandlerWrapper {
             handler: Box::new(handler),
-            update_interval,
+            update_interval: update_interval.into(),
             last_updated: Instant::now(),
         }
     }
@@ -107,11 +110,15 @@ impl<T> OnStartMetricsHandler for HandlerWrapper<T> {
 #[async_trait]
 impl<T> OnUpdateMetricsHandler for HandlerWrapper<T> {
     async fn on_update(&mut self) {
+        let Some(update_interval) = self.update_interval else {
+            return;
+        };
+
         let name = any::type_name::<T>();
         trace!("on update for handler for events of type {name}");
 
         let elapsed = self.last_updated.elapsed();
-        if elapsed < self.update_interval {
+        if elapsed < update_interval {
             trace!("too soon for updates");
             return;
         }

nym-node/src/node/metrics/handler/pending_egress_packets_updater.rs

@@ -0,0 +1,60 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::node::metrics::handler::{
+    MetricsHandler, OnStartMetricsHandler, OnUpdateMetricsHandler,
+};
+use async_trait::async_trait;
+use nym_gateway::node::ActiveClientsStore;
+use nym_mixnet_client::client::ActiveConnections;
+use nym_node_metrics::NymNodeMetrics;
+
+// it can be anything, we just need a unique type_id to register our handler
+pub struct PendingEgressPackets;
+
+pub struct PendingEgressPacketsUpdater {
+    metrics: NymNodeMetrics,
+    active_websocket_clients: ActiveClientsStore,
+    active_mixnet_connections: ActiveConnections,
+}
+
+impl PendingEgressPacketsUpdater {
+    pub(crate) fn new(
+        metrics: NymNodeMetrics,
+        active_clients: ActiveClientsStore,
+        active_mixnet_connections: ActiveConnections,
+    ) -> Self {
+        PendingEgressPacketsUpdater {
+            metrics,
+            active_websocket_clients: active_clients,
+            active_mixnet_connections,
+        }
+    }
+}
+
+#[async_trait]
+impl OnStartMetricsHandler for PendingEgressPacketsUpdater {}
+
+#[async_trait]
+impl OnUpdateMetricsHandler for PendingEgressPacketsUpdater {
+    async fn on_update(&mut self) {
+        let pending_final = self.active_websocket_clients.pending_packets();
+        self.metrics
+            .process
+            .update_final_hop_packets_pending_delivery(pending_final);
+
+        let pending_forward = self.active_mixnet_connections.pending_packets();
+        self.metrics
+            .process
+            .update_forward_hop_packets_pending_delivery(pending_forward)
+    }
+}
+
+#[async_trait]
+impl MetricsHandler for PendingEgressPacketsUpdater {
+    type Events = PendingEgressPackets;
+
+    async fn handle_event(&mut self, _event: Self::Events) {
+        panic!("this should have never been called! MetricsHandler has been incorrectly called on PendingEgressPacketsUpdater")
+    }
+}

nym-node/src/node/metrics/handler/prometheus_events_handler.rs

@@ -0,0 +1,6 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+// pub struct PrometheusEventsHandler {
+//     //
+// }

nym-node/src/node/mixnet/handler.rs

@@ -112,7 +112,14 @@ impl ConnectionHandler {
                     .await
                 {
                     Err(err) => error!("Failed to store client data - {err}"),
-                    Ok(_) => trace!("Stored packet for {client}"),
+                    Ok(_) => {
+                        self.shared
+                            .metrics
+                            .mixnet
+                            .egress
+                            .add_disk_persisted_packet();
+                        trace!("Stored packet for {client}")
+                    }
                 }
             }
             Ok(_) => trace!("Pushed received packet to {client}"),

nym-node/src/node/mixnet/packet_forwarding.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 futures::StreamExt;
 use nym_mixnet_client::forwarder::{
@@ -80,7 +80,7 @@ impl<C> PacketForwarder<C> {
         C: SendWithoutResponse,
     {
         let delayed_packet = packet.into_inner();
-        self.forward_packet(delayed_packet)
+        self.forward_packet(delayed_packet);
     }
 
     fn handle_new_packet(&mut self, new_packet: PacketToForward)
@@ -102,6 +102,18 @@ impl<C> PacketForwarder<C> {
         }
     }
 
+    fn update_queue_len_metric(&self) {
+        self.metrics
+            .process
+            .update_forward_hop_packets_being_delayed(self.delay_queue.len());
+    }
+
+    fn update_channel_size_metric(&self, channel_size: usize) {
+        self.metrics
+            .process
+            .update_packet_forwarder_queue_size(channel_size)
+    }
+
     pub async fn run(&mut self)
     where
         C: SendWithoutResponse,
@@ -125,18 +137,21 @@ impl<C> PacketForwarder<C> {
                     // and hence it can't happen that ALL senders are dropped
                     #[allow(clippy::unwrap_used)]
                     self.handle_new_packet(new_packet.unwrap());
+                    let channel_len = self.packet_sender.len();
                     if processed % 1000 == 0 {
-                        let queue_len = self.packet_sender.len();
-                        match queue_len {
+                        match channel_len {
                             n if n > 200 => error!("there are currently {n} mix packets waiting to get forwarded!"),
                             n if n > 50 => warn!("there are currently {n} mix packets waiting to get forwarded"),
                             n => trace!("there are currently {n} mix packets waiting to get forwarded"),
                         }
                     }
-
+                    self.update_channel_size_metric(channel_len);
                     processed += 1;
                 }
             }
+
+            // update the metrics on either new packet being inserted or packet being removed
+            self.update_queue_len_metric();
         }
         trace!("PacketForwarder: Exiting");
     }

nym-node/src/node/mod.rs

@@ -21,8 +21,10 @@ use crate::node::http::{HttpServerConfig, NymNodeHttpServer, NymNodeRouter};
 use crate::node::metrics::aggregator::MetricsAggregator;
 use crate::node::metrics::console_logger::ConsoleLogger;
 use crate::node::metrics::handler::client_sessions::GatewaySessionStatsHandler;
+use crate::node::metrics::handler::global_prometheus_updater::PrometheusGlobalNodeMetricsRegistryUpdater;
 use crate::node::metrics::handler::legacy_packet_data::LegacyMixingStatsUpdater;
 use crate::node::metrics::handler::mixnet_data_cleaner::MixnetMetricsCleaner;
+use crate::node::metrics::handler::pending_egress_packets_updater::PendingEgressPacketsUpdater;
 use crate::node::mixnet::packet_forwarding::PacketForwarder;
 use crate::node::mixnet::shared::ProcessingConfig;
 use crate::node::mixnet::SharedFinalHopData;
@@ -30,6 +32,7 @@ use crate::node::shared_topology::NymNodeTopologyProvider;
 use nym_bin_common::bin_info;
 use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_gateway::node::{ActiveClientsStore, GatewayTasksBuilder};
+use nym_mixnet_client::client::ActiveConnections;
 use nym_mixnet_client::forwarder::MixForwardingSender;
 use nym_network_requester::{
     set_active_gateway, setup_fs_gateways_storage, store_gateway_details, CustomGatewayDetails,
@@ -743,7 +746,8 @@ impl NymNode {
             .with_authenticator_details(auth_details)
             .with_used_exit_policy(exit_policy_details)
             .with_description(self.description.clone())
-            .with_auxiliary_details(auxiliary_details);
+            .with_auxiliary_details(auxiliary_details)
+            .with_prometheus_bearer_token(self.config.http.access_token.clone());
 
         if self.config.http.expose_system_info {
             config = config.with_system_info(get_system_info(
@@ -764,8 +768,7 @@ impl NymNode {
             config.api.v1_config.node.roles.ip_packet_router_enabled = true;
         }
 
-        let app_state = AppState::new(self.metrics.clone(), self.verloc_stats.clone())
-            .with_metrics_key(self.config.http.access_token.clone());
+        let app_state = AppState::new(self.metrics.clone(), self.verloc_stats.clone());
 
         Ok(NymNodeRouter::new(config, app_state)
             .build_server(&self.config.http.bind_address)
@@ -836,7 +839,12 @@ impl NymNode {
         tokio::spawn(async move { verloc_measurer.run().await });
     }
 
-    pub(crate) fn setup_metrics_backend(&self, shutdown: TaskClient) -> MetricEventsSender {
+    pub(crate) fn setup_metrics_backend(
+        &self,
+        active_clients_store: ActiveClientsStore,
+        active_egress_mixnet_connections: ActiveConnections,
+        shutdown: TaskClient,
+    ) -> MetricEventsSender {
         info!("setting up node metrics...");
 
         // aggregator (to listen for any metrics events)
@@ -862,12 +870,35 @@ impl NymNode {
             self.config.metrics.debug.clients_sessions_update_rate,
         );
 
-        // handler for periodically cleaning up stale recipient/sender darta
+        // handler for periodically cleaning up stale recipient/sender data
         metrics_aggregator.register_handler(
             MixnetMetricsCleaner::new(self.metrics.clone()),
             self.config.metrics.debug.stale_mixnet_metrics_cleaner_rate,
         );
 
+        // handler for updating the value of forward/final hop packets pending delivery
+        metrics_aggregator.register_handler(
+            PendingEgressPacketsUpdater::new(
+                self.metrics.clone(),
+                active_clients_store,
+                active_egress_mixnet_connections,
+            ),
+            self.config.metrics.debug.pending_egress_packets_update_rate,
+        );
+
+        // handler for updating the prometheus registry from the global atomic metrics counters
+        // such as number of packets received
+        metrics_aggregator.register_handler(
+            PrometheusGlobalNodeMetricsRegistryUpdater::new(self.metrics.clone()),
+            self.config
+                .metrics
+                .debug
+                .global_prometheus_counters_update_rate,
+        );
+
+        // handler for handling prometheus metrics events
+        // metrics_aggregator.register_handler(PrometheusEventsHandler{}, None);
+
         // note: we're still measuring things such as number of mixed packets,
         // but since they're stored as atomic integers, they are incremented directly at source
         // rather than going through event pipeline
@@ -901,7 +932,7 @@ impl NymNode {
         &self,
         active_clients_store: &ActiveClientsStore,
         shutdown: TaskClient,
-    ) -> MixForwardingSender {
+    ) -> (MixForwardingSender, ActiveConnections) {
         let processing_config = ProcessingConfig::new(&self.config);
 
         // we're ALWAYS listening for mixnet packets, either for forward or final hops (or both)
@@ -918,7 +949,13 @@ impl NymNode {
             self.config.mixnet.debug.initial_connection_timeout,
             self.config.mixnet.debug.maximum_connection_buffer_size,
         );
-        let mixnet_client = nym_mixnet_client::Client::new(mixnet_client_config);
+        let mixnet_client = nym_mixnet_client::Client::new(
+            mixnet_client_config,
+            self.metrics
+                .network
+                .active_egress_mixnet_connections_counter(),
+        );
+        let active_connections = mixnet_client.active_connections();
 
         let mut packet_forwarder = PacketForwarder::new(
             mixnet_client,
@@ -943,7 +980,7 @@ impl NymNode {
         );
 
         mixnet::Listener::new(self.config.mixnet.bind_address, shared).start();
-        mix_packet_sender
+        (mix_packet_sender, active_connections)
     }
 
     pub(crate) async fn run(mut self) -> Result<(), NymNodeError> {
@@ -973,14 +1010,19 @@ impl NymNode {
 
         self.start_verloc_measurements(task_manager.subscribe_named("verloc-measurements"));
 
-        let metrics_sender = self.setup_metrics_backend(task_manager.subscribe_named("metrics"));
         let active_clients_store = ActiveClientsStore::new();
 
-        let mix_packet_sender = self.start_mixnet_listener(
+        let (mix_packet_sender, active_egress_mixnet_connections) = self.start_mixnet_listener(
             &active_clients_store,
             task_manager.subscribe_named("mixnet-traffic"),
         );
 
+        let metrics_sender = self.setup_metrics_backend(
+            active_clients_store.clone(),
+            active_egress_mixnet_connections,
+            task_manager.subscribe_named("metrics"),
+        );
+
         self.start_gateway_tasks(
             metrics_sender,
             active_clients_store,

nym-node/src/node/shared_topology.rs

@@ -3,6 +3,7 @@
 
 use async_trait::async_trait;
 use nym_gateway::node::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent};
+use nym_node_metrics::prometheus_wrapper::{PrometheusMetric, PROMETHEUS_METRICS};
 use nym_topology::{gateway, NymTopology, TopologyProvider};
 use std::sync::Arc;
 use std::time::Duration;
@@ -93,6 +94,10 @@ impl TopologyProvider for NymNodeTopologyProvider {
         if let Some(cached) = guard.cached_topology() {
             return Some(cached);
         }
+
+        // the observation will be included on drop
+        let _timer =
+            PROMETHEUS_METRICS.start_timer(PrometheusMetric::ProcessTopologyQueryResolutionLatency);
         guard.update_cache().await
     }
 }