feat: [torrust#1523] add new metric: number of inactive peers

The metric is added to the `torrent-repository` package.

The metric in Prometheus format:

```
torrent_repository_peers_inactive_total{} 0
```

It was not included as a new label in the number of peers because it can't be calculated from current events.

New inactivity events could have been added but the solution was much
more complex than this and having two metrics counting peers is not so
bad.

The discarded alternative was addinga new label por satte (`active`,
`inactive`).

Author: josecelano

Cargo.lock

@@ -18,6 +18,7 @@ version.workspace = true
 [dependencies]
 aquatic_udp_protocol = "0"
 bittorrent-primitives = "0.1.0"
+chrono = { version = "0", default-features = false, features = ["clock"] }
 crossbeam-skiplist = "0"
 futures = "0"
 serde = { version = "1.0.219", features = ["derive"] }

packages/torrent-repository/Cargo.toml

@@ -1,5 +1,6 @@
 pub mod event;
 pub mod metrics;
+pub mod peers_inactivity_update;
 pub mod repository;
 
 use metrics::Metrics;
@@ -23,6 +24,7 @@ const TORRENT_REPOSITORY_PEERS_UPDATED_TOTAL: &str = "torrent_repository_peers_u
 
 const TORRENT_REPOSITORY_PEER_CONNECTIONS_TOTAL: &str = "torrent_repository_peer_connections_total";
 const TORRENT_REPOSITORY_UNIQUE_PEERS_TOTAL: &str = "torrent_repository_unique_peers_total"; // todo: not implemented yet
+const TORRENT_REPOSITORY_PEERS_INACTIVE_TOTAL: &str = "torrent_repository_peers_inactive_total";
 
 #[must_use]
 pub fn describe_metrics() -> Metrics {
@@ -88,5 +90,11 @@ pub fn describe_metrics() -> Metrics {
         Some(&MetricDescription::new("The total number of unique peers.")),
     );
 
+    metrics.metric_collection.describe_gauge(
+        &metric_name!(TORRENT_REPOSITORY_PEERS_INACTIVE_TOTAL),
+        Some(Unit::Count),
+        Some(&MetricDescription::new("The total number of inactive peers.")),
+    );
+
     metrics
 }

packages/torrent-repository/src/statistics/mod.rs

@@ -0,0 +1,72 @@
+//! Job that runs a task on intervals to update peers' inactivity metrics.
+use std::sync::Arc;
+
+use chrono::Utc;
+use tokio::task::JoinHandle;
+use torrust_tracker_clock::clock::Time;
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::metric_name;
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+use tracing::instrument;
+
+use super::repository::Repository;
+use crate::statistics::TORRENT_REPOSITORY_PEERS_INACTIVE_TOTAL;
+use crate::{CurrentClock, Swarms};
+
+#[must_use]
+#[instrument(skip(swarms, stats_repository))]
+pub fn start_job(
+    swarms: &Arc<Swarms>,
+    stats_repository: &Arc<Repository>,
+    inactivity_cutoff: DurationSinceUnixEpoch,
+) -> JoinHandle<()> {
+    let weak_swarms = std::sync::Arc::downgrade(swarms);
+    let weak_stats_repository = std::sync::Arc::downgrade(stats_repository);
+
+    let interval_in_secs = 15; // todo: make this configurable
+
+    tokio::spawn(async move {
+        let interval = std::time::Duration::from_secs(interval_in_secs);
+        let mut interval = tokio::time::interval(interval);
+        interval.tick().await;
+
+        loop {
+            tokio::select! {
+                _ = tokio::signal::ctrl_c() => {
+                    tracing::info!("Stopping peers inactivity metrics update job ...");
+                    break;
+                }
+                _ = interval.tick() => {
+                    if let (Some(swarms), Some(stats_repository)) = (weak_swarms.upgrade(), weak_stats_repository.upgrade()) {
+                        let start_time = Utc::now().time();
+
+                        tracing::debug!("Updating peers inactivity metrics (executed every {} secs) ...", interval_in_secs);
+
+                        let inactive_peers_total = swarms.count_inactive_peers(inactivity_cutoff).await;
+
+                        tracing::info!(inactive_peers_total = inactive_peers_total);
+
+                        #[allow(clippy::cast_precision_loss)]
+                        let inactive_peers_total = inactive_peers_total as f64;
+
+                        let _unused = stats_repository
+                            .set_gauge(
+                                &metric_name!(TORRENT_REPOSITORY_PEERS_INACTIVE_TOTAL),
+                                &LabelSet::default(),
+                                inactive_peers_total,
+                                CurrentClock::now(),
+                            )
+                            .await;
+
+                        tracing::debug!(
+                            "Peers inactivity metrics updated in {} ms",
+                            (Utc::now().time() - start_time).num_milliseconds()
+                        );
+                    } else {
+                        break;
+                    }
+                }
+            }
+        }
+    })
+}

packages/torrent-repository/src/statistics/peers_inactivity_update.rs

@@ -57,6 +57,31 @@ impl Repository {
         result
     }
 
+    /// # Errors
+    ///
+    /// This function will return an error if the metric collection fails to
+    /// set the gauge.
+    pub async fn set_gauge(
+        &self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        value: f64,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        let mut stats_lock = self.stats.write().await;
+
+        let result = stats_lock.set_gauge(metric_name, labels, value, now);
+
+        drop(stats_lock);
+
+        match result {
+            Ok(()) => {}
+            Err(ref err) => tracing::error!("Failed to set the gauge: {}", err),
+        }
+
+        result
+    }
+
     /// # Errors
     ///
     /// This function will return an error if the metric collection fails to

packages/torrent-repository/src/statistics/repository.rs

@@ -118,6 +118,14 @@ impl Swarm {
         (seeders, leechers)
     }
 
+    #[must_use]
+    pub fn count_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) -> usize {
+        self.peers
+            .iter()
+            .filter(|(_, peer)| peer::ReadInfo::get_updated(&**peer) <= current_cutoff)
+            .count()
+    }
+
     #[must_use]
     pub fn len(&self) -> usize {
         self.peers.len()
@@ -435,6 +443,22 @@ mod tests {
         assert_eq!(swarm.peers_excluding(&peer2.peer_addr, None), [Arc::new(peer1)]);
     }
 
+    #[tokio::test]
+    async fn it_should_count_inactive_peers() {
+        let mut swarm = Swarm::new(&sample_info_hash(), 0, None);
+        let mut downloads_increased = false;
+        let one_second = DurationSinceUnixEpoch::new(1, 0);
+
+        // Insert the peer
+        let last_update_time = DurationSinceUnixEpoch::new(1_669_397_478_934, 0);
+        let peer = PeerBuilder::default().last_updated_on(last_update_time).build();
+        swarm.upsert_peer(peer.into(), &mut downloads_increased).await;
+
+        let inactive_peers_total = swarm.count_inactive_peers(last_update_time + one_second);
+
+        assert_eq!(inactive_peers_total, 1);
+    }
+
     #[tokio::test]
     async fn it_should_remove_inactive_peers() {
         let mut swarm = Swarm::new(&sample_info_hash(), 0, None);

packages/torrent-repository/src/swarm.rs

@@ -248,6 +248,18 @@ impl Swarms {
         }
     }
 
+    /// Counts the number of inactive peers across all torrents.
+    pub async fn count_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) -> usize {
+        let mut inactive_peers_total = 0;
+
+        for swarm_handle in &self.swarms {
+            let swarm = swarm_handle.value().lock().await;
+            inactive_peers_total += swarm.count_inactive_peers(current_cutoff);
+        }
+
+        inactive_peers_total
+    }
+
     /// Removes inactive peers from all torrent entries.
     ///
     /// A peer is considered inactive if its last update timestamp is older than
@@ -705,6 +717,22 @@ mod tests {
                 assert!(swarms.get(&info_hash).is_none());
             }
 
+            #[tokio::test]
+            async fn it_should_count_inactive_peers() {
+                let swarms = Arc::new(Swarms::default());
+
+                let info_hash = sample_info_hash();
+                let mut peer = sample_peer();
+                peer.updated = DurationSinceUnixEpoch::new(0, 0);
+
+                swarms.handle_announcement(&info_hash, &peer, None).await.unwrap();
+
+                // Cut off time is 1 second after the peer was updated
+                let inactive_peers_total = swarms.count_inactive_peers(peer.updated.add(Duration::from_secs(1))).await;
+
+                assert_eq!(inactive_peers_total, 1);
+            }
+
             #[tokio::test]
             async fn it_should_remove_peers_that_have_not_been_updated_after_a_cutoff_time() {
                 let swarms = Arc::new(Swarms::default());

packages/torrent-repository/src/swarms.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 
 use torrust_tracker_clock::clock::Time;
 use torrust_tracker_configuration::Core;
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
 use super::repository::in_memory::InMemoryTorrentRepository;
 use super::repository::persisted::DatabasePersistentTorrentRepository;
@@ -103,10 +104,13 @@ impl TorrentsManager {
     }
 
     async fn remove_inactive_peers(&self) {
-        let current_cutoff = CurrentClock::now_sub(&Duration::from_secs(u64::from(self.config.tracker_policy.max_peer_timeout)))
-            .unwrap_or_default();
+        self.in_memory_torrent_repository
+            .remove_inactive_peers(self.current_cutoff())
+            .await;
+    }
 
-        self.in_memory_torrent_repository.remove_inactive_peers(current_cutoff).await;
+    fn current_cutoff(&self) -> DurationSinceUnixEpoch {
+        CurrentClock::now_sub(&Duration::from_secs(u64::from(self.config.tracker_policy.max_peer_timeout))).unwrap_or_default()
     }
 
     async fn remove_peerless_torrents(&self) {

packages/tracker-core/src/torrent/manager.rs

@@ -27,7 +27,9 @@ use torrust_tracker_configuration::{Configuration, HttpTracker, UdpTracker};
 use tracing::instrument;
 
 use crate::bootstrap::jobs::manager::JobManager;
-use crate::bootstrap::jobs::{self, health_check_api, http_tracker, torrent_cleanup, tracker_apis, udp_tracker};
+use crate::bootstrap::jobs::{
+    self, health_check_api, http_tracker, peers_inactivity_update, torrent_cleanup, tracker_apis, udp_tracker,
+};
 use crate::bootstrap::{self};
 use crate::container::AppContainer;
 
@@ -79,8 +81,11 @@ async fn start_jobs(config: &Configuration, app_container: &Arc<AppContainer>) -
 
     start_the_udp_instances(config, app_container, &mut job_manager).await;
     start_the_http_instances(config, app_container, &mut job_manager).await;
-    start_the_http_api(config, app_container, &mut job_manager).await;
+
     start_torrent_cleanup(config, app_container, &mut job_manager);
+    start_peers_inactivity_update(config, app_container, &mut job_manager);
+
+    start_the_http_api(config, app_container, &mut job_manager).await;
     start_health_check_api(config, app_container, &mut job_manager).await;
 
     job_manager
@@ -260,6 +265,16 @@ fn start_torrent_cleanup(config: &Configuration, app_container: &Arc<AppContaine
     }
 }
 
+fn start_peers_inactivity_update(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
+    if config.core.tracker_usage_statistics {
+        let handle = peers_inactivity_update::start_job(config, app_container);
+
+        job_manager.push("peers_inactivity_update", handle);
+    } else {
+        tracing::info!("Peers inactivity update job is disabled.");
+    }
+}
+
 async fn start_health_check_api(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     let handle = health_check_api::start_job(&config.health_check_api, app_container.registar.entries()).await;
 

src/app.rs

@@ -10,6 +10,7 @@ pub mod health_check_api;
 pub mod http_tracker;
 pub mod http_tracker_core;
 pub mod manager;
+pub mod peers_inactivity_update;
 pub mod torrent_cleanup;
 pub mod torrent_repository;
 pub mod tracker_apis;