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

Author: josecelano

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

@@ -0,0 +1,104 @@
+//! Job that runs a task on intervals to update peers' activity 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, TORRENT_REPOSITORY_TORRENTS_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 activity metrics update job (ctrl-c signal received) ...");
+                    break;
+                }
+                _ = interval.tick() => {
+                    if let (Some(swarms), Some(stats_repository)) = (weak_swarms.upgrade(), weak_stats_repository.upgrade()) {
+                        update_activity_metrics(interval_in_secs, &swarms, &stats_repository, inactivity_cutoff).await;
+                    } else {
+                        tracing::info!("Stopping peers activity metrics update job (can't upgrade weak pointers) ...");
+                        break;
+                    }
+                }
+            }
+        }
+    })
+}
+
+async fn update_activity_metrics(
+    interval_in_secs: u64,
+    swarms: &Arc<Swarms>,
+    stats_repository: &Arc<Repository>,
+    inactivity_cutoff: DurationSinceUnixEpoch,
+) {
+    let start_time = Utc::now().time();
+
+    tracing::debug!(
+        "Updating peers and torrents activity metrics (executed every {} secs) ...",
+        interval_in_secs
+    );
+
+    let activity_metadata = swarms.get_activity_metadata(inactivity_cutoff).await;
+
+    activity_metadata.log();
+
+    update_inactive_peers_total(stats_repository, activity_metadata.inactive_peers_total).await;
+    update_inactive_torrents_total(stats_repository, activity_metadata.inactive_torrents_total).await;
+
+    tracing::debug!(
+        "Peers and torrents activity metrics updated in {} ms",
+        (Utc::now().time() - start_time).num_milliseconds()
+    );
+}
+
+async fn update_inactive_peers_total(stats_repository: &Arc<Repository>, inactive_peers_total: usize) {
+    #[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;
+}
+
+async fn update_inactive_torrents_total(stats_repository: &Arc<Repository>, inactive_torrents_total: usize) {
+    #[allow(clippy::cast_precision_loss)]
+    let inactive_torrents_total = inactive_torrents_total as f64;
+
+    let _unused = stats_repository
+        .set_gauge(
+            &metric_name!(TORRENT_REPOSITORY_TORRENTS_INACTIVE_TOTAL),
+            &LabelSet::default(),
+            inactive_torrents_total,
+            CurrentClock::now(),
+        )
+        .await;
+}

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

@@ -1,6 +1,6 @@
+pub mod activity_metrics_updater;
 pub mod event;
 pub mod metrics;
-pub mod peers_inactivity_update;
 pub mod repository;
 
 use metrics::Metrics;
@@ -15,6 +15,7 @@ const TORRENT_REPOSITORY_TORRENTS_REMOVED_TOTAL: &str = "torrent_repository_torr
 
 const TORRENT_REPOSITORY_TORRENTS_TOTAL: &str = "torrent_repository_torrents_total";
 const TORRENT_REPOSITORY_TORRENTS_DOWNLOADS_TOTAL: &str = "torrent_repository_torrents_downloads_total";
+const TORRENT_REPOSITORY_TORRENTS_INACTIVE_TOTAL: &str = "torrent_repository_torrents_inactive_total";
 
 // Peers metrics
 
@@ -56,6 +57,12 @@ pub fn describe_metrics() -> Metrics {
         Some(&MetricDescription::new("The total number of torrent downloads.")),
     );
 
+    metrics.metric_collection.describe_gauge(
+        &metric_name!(TORRENT_REPOSITORY_TORRENTS_INACTIVE_TOTAL),
+        Some(Unit::Count),
+        Some(&MetricDescription::new("The total number of inactive torrents.")),
+    );
+
     // Peers metrics
 
     metrics.metric_collection.describe_counter(

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

@@ -126,6 +126,17 @@ impl Swarm {
             .count()
     }
 
+    #[must_use]
+    pub fn get_activity_metadata(&self, current_cutoff: DurationSinceUnixEpoch) -> ActivityMetadata {
+        let inactive_peers_total = self.count_inactive_peers(current_cutoff);
+
+        let active_peers_total = self.len() - inactive_peers_total;
+
+        let is_active = active_peers_total > 0;
+
+        ActivityMetadata::new(is_active, active_peers_total, inactive_peers_total)
+    }
+
     #[must_use]
     pub fn len(&self) -> usize {
         self.peers.len()
@@ -296,6 +307,30 @@ impl Swarm {
     }
 }
 
+#[derive(Clone)]
+pub struct ActivityMetadata {
+    /// Indicates if the swarm is active. It's inactive if there are no active
+    /// peers.
+    pub is_active: bool,
+
+    /// The number of active peers in the swarm.
+    pub active_peers_total: usize,
+
+    /// The number of inactive peers in the swarm.
+    pub inactive_peers_total: usize,
+}
+
+impl ActivityMetadata {
+    #[must_use]
+    pub fn new(is_active: bool, active_peers_total: usize, inactive_peers_total: usize) -> Self {
+        Self {
+            is_active,
+            active_peers_total,
+            inactive_peers_total,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
 

packages/torrent-repository/src/swarm.rs

@@ -248,6 +248,32 @@ impl Swarms {
         }
     }
 
+    pub async fn get_activity_metadata(&self, current_cutoff: DurationSinceUnixEpoch) -> AggregateActivityMetadata {
+        let mut active_peers_total = 0;
+        let mut inactive_peers_total = 0;
+        let mut active_torrents_total = 0;
+
+        for swarm_handle in &self.swarms {
+            let swarm = swarm_handle.value().lock().await;
+
+            let activity_metadata = swarm.get_activity_metadata(current_cutoff);
+
+            if activity_metadata.is_active {
+                active_torrents_total += 1;
+            }
+
+            active_peers_total += activity_metadata.active_peers_total;
+            inactive_peers_total += activity_metadata.inactive_peers_total;
+        }
+
+        AggregateActivityMetadata {
+            active_peers_total,
+            inactive_peers_total,
+            active_torrents_total,
+            inactive_torrents_total: self.len() - active_torrents_total,
+        }
+    }
+
     /// 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;
@@ -446,6 +472,31 @@ impl Swarms {
 #[derive(thiserror::Error, Debug, Clone)]
 pub enum Error {}
 
+#[derive(Clone, Debug, Default)]
+pub struct AggregateActivityMetadata {
+    /// The number of active peers in all swarms.
+    pub active_peers_total: usize,
+
+    /// The number of inactive peers in all swarms.
+    pub inactive_peers_total: usize,
+
+    /// The number of active torrents.
+    pub active_torrents_total: usize,
+
+    /// The number of inactive torrents.
+    pub inactive_torrents_total: usize,
+}
+
+impl AggregateActivityMetadata {
+    pub fn log(&self) {
+        tracing::info!(
+            active_peers_total = self.active_peers_total,
+            inactive_peers_total = self.inactive_peers_total,
+            active_torrents_total = self.active_torrents_total,
+            inactive_torrents_total = self.inactive_torrents_total
+        );
+    }
+}
 #[cfg(test)]
 mod tests {
 

packages/torrent-repository/src/swarms.rs

@@ -28,7 +28,7 @@ use tracing::instrument;
 
 use crate::bootstrap::jobs::manager::JobManager;
 use crate::bootstrap::jobs::{
-    self, health_check_api, http_tracker, peers_inactivity_update, torrent_cleanup, tracker_apis, udp_tracker,
+    self, health_check_api, http_tracker, activity_metrics_updater, torrent_cleanup, tracker_apis, udp_tracker,
 };
 use crate::bootstrap::{self};
 use crate::container::AppContainer;
@@ -267,7 +267,7 @@ 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);
+        let handle = activity_metrics_updater::start_job(config, app_container);
 
         job_manager.push("peers_inactivity_update", handle);
     } else {

src/app.rs

@@ -1,4 +1,4 @@
-//! Job that runs a task on intervals to update peers' inactivity metrics.
+//! Job that runs a task on intervals to update peers' activity metrics.
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -11,7 +11,7 @@ use crate::CurrentClock;
 
 #[must_use]
 pub fn start_job(config: &Configuration, app_container: &Arc<AppContainer>) -> JoinHandle<()> {
-    torrust_tracker_torrent_repository::statistics::peers_inactivity_update::start_job(
+    torrust_tracker_torrent_repository::statistics::activity_metrics_updater::start_job(
         &app_container.torrent_repository_container.swarms.clone(),
         &app_container.torrent_repository_container.stats_repository.clone(),
         peer_inactivity_cutoff_timestamp(config.core.tracker_policy.max_peer_timeout),

…ootstrap/jobs/peers_inactivity_update.rs …otstrap/jobs/activity_metrics_updater.rssrc/bootstrap/jobs/peers_inactivity_update.rs renamed to src/bootstrap/jobs/activity_metrics_updater.rs src/bootstrap/jobs/peers_inactivity_update.rs renamed to src/bootstrap/jobs/activity_metrics_updater.rs

@@ -6,11 +6,11 @@
 //! 2. Launch all the application services as concurrent jobs.
 //!
 //! This modules contains all the functions needed to start those jobs.
+pub mod activity_metrics_updater;
 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;