feat: [#1358] basic scaffolding for events in torrent-repository pkg

TODO:

- Run the event listener for the torrent-repository package when the tracker starts.
- Inject enven sender in `Swarms` and `Swarm` type to send events.
- Trigger events and process them to update the metrics.
- Expose the metrics via the `metrics` API endpoint.
- ...

Author: josecelano

Cargo.lock

@@ -19,10 +19,13 @@ version.workspace = true
 aquatic_udp_protocol = "0"
 bittorrent-primitives = "0.1.0"
 crossbeam-skiplist = "0"
+serde = "1.0.219"
 thiserror = "2.0.12"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
 torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
+torrust-tracker-events = { version = "3.0.0-develop", path = "../events" }
+torrust-tracker-metrics = { version = "3.0.0-develop", path = "../metrics" }
 torrust-tracker-primitives = { version = "3.0.0-develop", path = "../primitives" }
 tracing = "0"
 

packages/torrent-repository/Cargo.toml

@@ -0,0 +1,44 @@
+use std::net::SocketAddr;
+
+use aquatic_udp_protocol::PeerId;
+use bittorrent_primitives::info_hash::InfoHash;
+use torrust_tracker_primitives::peer::PeerAnnouncement;
+
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum Event {
+    TorrentAdded {
+        info_hash: InfoHash,
+        announcement: PeerAnnouncement,
+    },
+    TorrentRemoved {
+        info_hash: InfoHash,
+    },
+    PeerAdded {
+        announcement: PeerAnnouncement,
+    },
+    PeerRemoved {
+        socket_addr: SocketAddr,
+        peer_id: PeerId,
+    },
+}
+
+pub mod sender {
+    use std::sync::Arc;
+
+    use super::Event;
+
+    pub type Sender = Option<Arc<dyn torrust_tracker_events::sender::Sender<Event = Event>>>;
+    pub type Broadcaster = torrust_tracker_events::broadcaster::Broadcaster<Event>;
+}
+
+pub mod receiver {
+    use super::Event;
+
+    pub type Receiver = Box<dyn torrust_tracker_events::receiver::Receiver<Event = Event>>;
+}
+
+pub mod bus {
+    use crate::event::Event;
+
+    pub type EventBus = torrust_tracker_events::bus::EventBus<Event>;
+}

packages/torrent-repository/src/event.rs

@@ -1,3 +1,5 @@
+pub mod event;
+pub mod statistics;
 pub mod swarm;
 pub mod swarms;
 
@@ -19,6 +21,8 @@ pub(crate) type CurrentClock = clock::Working;
 #[allow(dead_code)]
 pub(crate) type CurrentClock = clock::Stopped;
 
+pub const TORRENT_REPOSITORY_LOG_TARGET: &str = "TORRENT_REPOSITORY";
+
 pub trait LockTrackedTorrent {
     fn lock_or_panic(&self) -> MutexGuard<'_, Swarm>;
 }

packages/torrent-repository/src/lib.rs

@@ -0,0 +1,21 @@
+use std::sync::Arc;
+
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use crate::event::Event;
+use crate::statistics::repository::Repository;
+
+/// # Panics
+///
+/// This function panics if the client IP address is not the same as the IP
+/// version of the event.
+pub async fn handle_event(_event: Event, stats_repository: &Arc<Repository>, _now: DurationSinceUnixEpoch) {
+    /*match event {
+        Event::TorrentAdded { .. } => {}
+        Event::TorrentRemoved { .. } => {}
+        Event::PeerAdded { .. } => {}
+        Event::PeerRemoved { .. } => {}
+    }*/
+
+    tracing::debug!("metrics: {:?}", stats_repository.get_metrics().await);
+}

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

@@ -0,0 +1,57 @@
+use std::sync::Arc;
+
+use tokio::task::JoinHandle;
+use torrust_tracker_clock::clock::Time;
+use torrust_tracker_events::receiver::RecvError;
+
+use super::handler::handle_event;
+use crate::event::receiver::Receiver;
+use crate::statistics::repository::Repository;
+use crate::{CurrentClock, TORRENT_REPOSITORY_LOG_TARGET};
+
+#[must_use]
+pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> JoinHandle<()> {
+    let stats_repository = repository.clone();
+
+    tracing::info!(target: TORRENT_REPOSITORY_LOG_TARGET, "Starting torrent repository event listener");
+
+    tokio::spawn(async move {
+        dispatch_events(receiver, stats_repository).await;
+
+        tracing::info!(target: TORRENT_REPOSITORY_LOG_TARGET, "Torrent repository listener finished");
+    })
+}
+
+async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
+    let shutdown_signal = tokio::signal::ctrl_c();
+
+    tokio::pin!(shutdown_signal);
+
+    loop {
+        tokio::select! {
+            biased;
+
+            _ = &mut shutdown_signal => {
+                tracing::info!(target: TORRENT_REPOSITORY_LOG_TARGET, "Received Ctrl+C, shutting down torrent repository event listener.");
+                break;
+            }
+
+            result = receiver.recv() => {
+                match result {
+                    Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
+                    Err(e) => {
+                        match e {
+                            RecvError::Closed => {
+                                tracing::info!(target: TORRENT_REPOSITORY_LOG_TARGET, "Torrent repository event receiver closed.");
+                                break;
+                            }
+                            RecvError::Lagged(n) => {
+                                tracing::warn!(target: TORRENT_REPOSITORY_LOG_TARGET, "Torrent repository event receiver lagged by {} events.", n);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

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

@@ -0,0 +1,2 @@
+pub mod handler;
+pub mod listener;

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

@@ -0,0 +1,39 @@
+use serde::Serialize;
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::metric::MetricName;
+use torrust_tracker_metrics::metric_collection::{Error, MetricCollection};
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+/// Metrics collected by the torrent repository.
+#[derive(Debug, Clone, PartialEq, Default, Serialize)]
+pub struct Metrics {
+    /// A collection of metrics.
+    pub metric_collection: MetricCollection,
+}
+
+impl Metrics {
+    /// # Errors
+    ///
+    /// Returns an error if the metric does not exist and it cannot be created.
+    pub fn increase_counter(
+        &mut self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        self.metric_collection.increase_counter(metric_name, labels, now)
+    }
+
+    /// # Errors
+    ///
+    /// Returns an error if the metric does not exist and it cannot be created.
+    pub fn set_gauge(
+        &mut self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        value: f64,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        self.metric_collection.set_gauge(metric_name, labels, value, now)
+    }
+}

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

@@ -0,0 +1,56 @@
+pub mod event;
+pub mod metrics;
+pub mod repository;
+
+use metrics::Metrics;
+use torrust_tracker_metrics::metric::description::MetricDescription;
+use torrust_tracker_metrics::metric_name;
+use torrust_tracker_metrics::unit::Unit;
+
+/*
+
+Metrics name convention for number of downloads (not all of them are available):
+
+```
+{crate_prefix}_{lifespan}_{server_scope}_{swarm_scope}_{metric_name}
+```
+
+Where:
+
+- `crate_prefix`: The prefix of the crate (e.g. torrent_repository).
+- `lifespan`: The lifespan of the metric; `per_session` or `persistent`.
+- `server_scope`: The scope of the server; `all_instances` (all servers) or `per_instance` (one server).
+- `swarm_scope`: The scope of the metric; `per_torrent` (one torrent) or `all_torrents` (all torrents).
+- `metric_name`: The name of the metric; `downloads_total` (total downloads).
+
+Example:
+
+```
+torrent_repository_per_session_all_instances_all_torrents_downloads_total
+```
+
+- `crate_prefix`: `torrent_repository``
+- `lifespan`: `per_session`
+- `server_scope`: `all_instances`
+- `swarm_scope`: `all_torrents`
+- `metric_name`: `downloads_total`
+
+*/
+
+const PER_SESSION_ALL_INSTANCES_ALL_TORRENTS_DOWNLOADS_TOTAL: &str =
+    "torrent_repository_per_session_all_instances_all_torrents_downloads_total";
+
+#[must_use]
+pub fn describe_metrics() -> Metrics {
+    let mut metrics = Metrics::default();
+
+    metrics.metric_collection.describe_counter(
+        &metric_name!(PER_SESSION_ALL_INSTANCES_ALL_TORRENTS_DOWNLOADS_TOTAL),
+        Some(Unit::Count),
+        Some(&MetricDescription::new(
+            "The total number of downloads for all torrents and all servers, since the tracker started.",
+        )),
+    );
+
+    metrics
+}

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

@@ -0,0 +1,54 @@
+use std::sync::Arc;
+
+use tokio::sync::{RwLock, RwLockReadGuard};
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::metric::MetricName;
+use torrust_tracker_metrics::metric_collection::Error;
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use super::describe_metrics;
+use super::metrics::Metrics;
+
+/// A repository for the torrent repository metrics.
+#[derive(Clone)]
+pub struct Repository {
+    pub stats: Arc<RwLock<Metrics>>,
+}
+
+impl Default for Repository {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Repository {
+    #[must_use]
+    pub fn new() -> Self {
+        let stats = Arc::new(RwLock::new(describe_metrics()));
+
+        Self { stats }
+    }
+
+    pub async fn get_metrics(&self) -> RwLockReadGuard<'_, Metrics> {
+        self.stats.read().await
+    }
+
+    /// # Errors
+    ///
+    /// This function will return an error if the metric collection fails to
+    /// increase the counter.
+    pub async fn increase_counter(
+        &self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        let mut stats_lock = self.stats.write().await;
+
+        let result = stats_lock.increase_counter(metric_name, labels, now);
+
+        drop(stats_lock);
+
+        result
+    }
+}