refactor: [torrust#1228] split statistics mod into UDO and HTTP statistics

Author: josecelano

src/packages/http_tracker_core/mod.rs

@@ -0,0 +1 @@
+pub mod statistics;

src/packages/http_tracker_core/statistics/event/handler.rs

@@ -0,0 +1,123 @@
+use crate::packages::http_tracker_core::statistics::event::Event;
+use crate::packages::http_tracker_core::statistics::repository::Repository;
+
+pub async fn handle_event(event: Event, stats_repository: &Repository) {
+    match event {
+        // TCP4
+        Event::Tcp4Announce => {
+            stats_repository.increase_tcp4_announces().await;
+            stats_repository.increase_tcp4_connections().await;
+        }
+        Event::Tcp4Scrape => {
+            stats_repository.increase_tcp4_scrapes().await;
+            stats_repository.increase_tcp4_connections().await;
+        }
+
+        // TCP6
+        Event::Tcp6Announce => {
+            stats_repository.increase_tcp6_announces().await;
+            stats_repository.increase_tcp6_connections().await;
+        }
+        Event::Tcp6Scrape => {
+            stats_repository.increase_tcp6_scrapes().await;
+            stats_repository.increase_tcp6_connections().await;
+        }
+    }
+
+    tracing::debug!("stats: {:?}", stats_repository.get_stats().await);
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::packages::http_tracker_core::statistics::event::handler::handle_event;
+    use crate::packages::http_tracker_core::statistics::event::Event;
+    use crate::packages::http_tracker_core::statistics::repository::Repository;
+
+    #[tokio::test]
+    async fn should_increase_the_tcp4_announces_counter_when_it_receives_a_tcp4_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp4Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp4_announces_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp4_connections_counter_when_it_receives_a_tcp4_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp4Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp4_connections_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp4_scrapes_counter_when_it_receives_a_tcp4_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp4Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp4_scrapes_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp4_connections_counter_when_it_receives_a_tcp4_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp4Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp4_connections_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp6_announces_counter_when_it_receives_a_tcp6_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp6Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp6_announces_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp6_connections_counter_when_it_receives_a_tcp6_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp6Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp6_connections_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp6_scrapes_counter_when_it_receives_a_tcp6_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp6Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp6_scrapes_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_tcp6_connections_counter_when_it_receives_a_tcp6_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Tcp6Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.tcp6_connections_handled, 1);
+    }
+}

src/packages/http_tracker_core/statistics/event/listener.rs

@@ -0,0 +1,11 @@
+use tokio::sync::mpsc;
+
+use super::handler::handle_event;
+use super::Event;
+use crate::packages::http_tracker_core::statistics::repository::Repository;
+
+pub async fn dispatch_events(mut receiver: mpsc::Receiver<Event>, stats_repository: Repository) {
+    while let Some(event) = receiver.recv().await {
+        handle_event(event, &stats_repository).await;
+    }
+}

src/packages/http_tracker_core/statistics/event/mod.rs

@@ -0,0 +1,21 @@
+pub mod handler;
+pub mod listener;
+pub mod sender;
+
+/// An statistics event. It is used to collect tracker metrics.
+///
+/// - `Tcp` prefix means the event was triggered by the HTTP tracker
+/// - `Udp` prefix means the event was triggered by the UDP tracker
+/// - `4` or `6` prefixes means the IP version used by the peer
+/// - Finally the event suffix is the type of request: `announce`, `scrape` or `connection`
+///
+/// > NOTE: HTTP trackers do not use `connection` requests.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Event {
+    // code-review: consider one single event for request type with data: Event::Announce { scheme: HTTPorUDP, ip_version: V4orV6 }
+    // Attributes are enums too.
+    Tcp4Announce,
+    Tcp4Scrape,
+    Tcp6Announce,
+    Tcp6Scrape,
+}

src/packages/http_tracker_core/statistics/event/sender.rs

@@ -0,0 +1,29 @@
+use futures::future::BoxFuture;
+use futures::FutureExt;
+#[cfg(test)]
+use mockall::{automock, predicate::str};
+use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::SendError;
+
+use super::Event;
+
+/// A trait to allow sending statistics events
+#[cfg_attr(test, automock)]
+pub trait Sender: Sync + Send {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>>;
+}
+
+/// An [`statistics::EventSender`](crate::packages::http_tracker_core::statistics::event::sender::Sender) implementation.
+///
+/// It uses a channel sender to send the statistic events. The channel is created by a
+/// [`statistics::Keeper`](crate::packages::http_tracker_core::statistics::keeper::Keeper)
+#[allow(clippy::module_name_repetitions)]
+pub struct ChannelSender {
+    pub(crate) sender: mpsc::Sender<Event>,
+}
+
+impl Sender for ChannelSender {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>> {
+        async move { Some(self.sender.send(event).await) }.boxed()
+    }
+}

src/packages/http_tracker_core/statistics/keeper.rs

@@ -0,0 +1,77 @@
+use tokio::sync::mpsc;
+
+use super::event::listener::dispatch_events;
+use super::event::sender::{ChannelSender, Sender};
+use super::event::Event;
+use super::repository::Repository;
+
+const CHANNEL_BUFFER_SIZE: usize = 65_535;
+
+/// The service responsible for keeping tracker metrics (listening to statistics events and handle them).
+///
+/// It actively listen to new statistics events. When it receives a new event
+/// it accordingly increases the counters.
+pub struct Keeper {
+    pub repository: Repository,
+}
+
+impl Default for Keeper {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Keeper {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            repository: Repository::new(),
+        }
+    }
+
+    #[must_use]
+    pub fn new_active_instance() -> (Box<dyn Sender>, Repository) {
+        let mut stats_tracker = Self::new();
+
+        let stats_event_sender = stats_tracker.run_event_listener();
+
+        (stats_event_sender, stats_tracker.repository)
+    }
+
+    pub fn run_event_listener(&mut self) -> Box<dyn Sender> {
+        let (sender, receiver) = mpsc::channel::<Event>(CHANNEL_BUFFER_SIZE);
+
+        let stats_repository = self.repository.clone();
+
+        tokio::spawn(async move { dispatch_events(receiver, stats_repository).await });
+
+        Box::new(ChannelSender { sender })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::packages::http_tracker_core::statistics::event::Event;
+    use crate::packages::http_tracker_core::statistics::keeper::Keeper;
+    use crate::packages::http_tracker_core::statistics::metrics::Metrics;
+
+    #[tokio::test]
+    async fn should_contain_the_tracker_statistics() {
+        let stats_tracker = Keeper::new();
+
+        let stats = stats_tracker.repository.get_stats().await;
+
+        assert_eq!(stats.tcp4_announces_handled, Metrics::default().tcp4_announces_handled);
+    }
+
+    #[tokio::test]
+    async fn should_create_an_event_sender_to_send_statistical_events() {
+        let mut stats_tracker = Keeper::new();
+
+        let event_sender = stats_tracker.run_event_listener();
+
+        let result = event_sender.send_event(Event::Tcp4Announce).await;
+
+        assert!(result.is_some());
+    }
+}

src/packages/http_tracker_core/statistics/metrics.rs

@@ -0,0 +1,30 @@
+/// Metrics collected by the tracker.
+///
+/// - Number of connections handled
+/// - Number of `announce` requests handled
+/// - Number of `scrape` request handled
+///
+/// These metrics are collected for each connection type: UDP and HTTP
+/// and also for each IP version used by the peers: IPv4 and IPv6.
+#[derive(Debug, PartialEq, Default)]
+pub struct Metrics {
+    /// Total number of TCP (HTTP tracker) connections from IPv4 peers.
+    /// Since the HTTP tracker spec does not require a handshake, this metric
+    /// increases for every HTTP request.
+    pub tcp4_connections_handled: u64,
+
+    /// Total number of TCP (HTTP tracker) `announce` requests from IPv4 peers.
+    pub tcp4_announces_handled: u64,
+
+    /// Total number of TCP (HTTP tracker) `scrape` requests from IPv4 peers.
+    pub tcp4_scrapes_handled: u64,
+
+    /// Total number of TCP (HTTP tracker) connections from IPv6 peers.
+    pub tcp6_connections_handled: u64,
+
+    /// Total number of TCP (HTTP tracker) `announce` requests from IPv6 peers.
+    pub tcp6_announces_handled: u64,
+
+    /// Total number of TCP (HTTP tracker) `scrape` requests from IPv6 peers.
+    pub tcp6_scrapes_handled: u64,
+}

src/packages/http_tracker_core/statistics/mod.rs

@@ -0,0 +1,6 @@
+pub mod event;
+pub mod keeper;
+pub mod metrics;
+pub mod repository;
+pub mod services;
+pub mod setup;

src/packages/http_tracker_core/statistics/repository.rs

@@ -0,0 +1,66 @@
+use std::sync::Arc;
+
+use tokio::sync::{RwLock, RwLockReadGuard};
+
+use super::metrics::Metrics;
+
+/// A repository for the tracker 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 {
+        Self {
+            stats: Arc::new(RwLock::new(Metrics::default())),
+        }
+    }
+
+    pub async fn get_stats(&self) -> RwLockReadGuard<'_, Metrics> {
+        self.stats.read().await
+    }
+
+    pub async fn increase_tcp4_announces(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp4_announces_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_tcp4_connections(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp4_connections_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_tcp4_scrapes(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp4_scrapes_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_tcp6_announces(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp6_announces_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_tcp6_connections(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp6_connections_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_tcp6_scrapes(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.tcp6_scrapes_handled += 1;
+        drop(stats_lock);
+    }
+}

src/packages/http_tracker_core/statistics/services.rs

@@ -0,0 +1,104 @@
+//! Statistics services.
+//!
+//! It includes:
+//!
+//! - A [`factory`](crate::packages::http_tracker_core::statistics::setup::factory) function to build the structs needed to collect the tracker metrics.
+//! - A [`get_metrics`] service to get the tracker [`metrics`](crate::packages::http_tracker_core::statistics::metrics::Metrics).
+//!
+//! Tracker metrics are collected using a Publisher-Subscribe pattern.
+//!
+//! The factory function builds two structs:
+//!
+//! - An statistics event [`Sender`](crate::packages::http_tracker_core::statistics::event::sender::Sender)
+//! - An statistics [`Repository`]
+//!
+//! ```text
+//! let (stats_event_sender, stats_repository) = factory(tracker_usage_statistics);
+//! ```
+//!
+//! The statistics repository is responsible for storing the metrics in memory.
+//! The statistics event sender allows sending events related to metrics.
+//! There is an event listener that is receiving all the events and processing them with an event handler.
+//! Then, the event handler updates the metrics depending on the received event.
+use std::sync::Arc;
+
+use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
+use packages::http_tracker_core::statistics::metrics::Metrics;
+use packages::http_tracker_core::statistics::repository::Repository;
+use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+
+use crate::packages;
+
+/// All the metrics collected by the tracker.
+#[derive(Debug, PartialEq)]
+pub struct TrackerMetrics {
+    /// Domain level metrics.
+    ///
+    /// General metrics for all torrents (number of seeders, leechers, etcetera)
+    pub torrents_metrics: TorrentsMetrics,
+
+    /// Application level metrics. Usage statistics/metrics.
+    ///
+    /// Metrics about how the tracker is been used (number of  number of http scrape requests, etcetera)
+    pub protocol_metrics: Metrics,
+}
+
+/// It returns all the [`TrackerMetrics`]
+pub async fn get_metrics(
+    in_memory_torrent_repository: Arc<InMemoryTorrentRepository>,
+    stats_repository: Arc<Repository>,
+) -> TrackerMetrics {
+    let torrents_metrics = in_memory_torrent_repository.get_torrents_metrics();
+    let stats = stats_repository.get_stats().await;
+
+    TrackerMetrics {
+        torrents_metrics,
+        protocol_metrics: Metrics {
+            // TCPv4
+            tcp4_connections_handled: stats.tcp4_connections_handled,
+            tcp4_announces_handled: stats.tcp4_announces_handled,
+            tcp4_scrapes_handled: stats.tcp4_scrapes_handled,
+            // TCPv6
+            tcp6_connections_handled: stats.tcp6_connections_handled,
+            tcp6_announces_handled: stats.tcp6_announces_handled,
+            tcp6_scrapes_handled: stats.tcp6_scrapes_handled,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
+    use bittorrent_tracker_core::{self};
+    use torrust_tracker_configuration::Configuration;
+    use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+    use torrust_tracker_test_helpers::configuration;
+
+    use crate::packages::http_tracker_core::statistics;
+    use crate::packages::http_tracker_core::statistics::services::{get_metrics, TrackerMetrics};
+
+    pub fn tracker_configuration() -> Configuration {
+        configuration::ephemeral()
+    }
+
+    #[tokio::test]
+    async fn the_statistics_service_should_return_the_tracker_metrics() {
+        let config = tracker_configuration();
+
+        let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
+        let (_stats_event_sender, stats_repository) = statistics::setup::factory(config.core.tracker_usage_statistics);
+        let stats_repository = Arc::new(stats_repository);
+
+        let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), stats_repository.clone()).await;
+
+        assert_eq!(
+            tracker_metrics,
+            TrackerMetrics {
+                torrents_metrics: TorrentsMetrics::default(),
+                protocol_metrics: statistics::metrics::Metrics::default(),
+            }
+        );
+    }
+}

src/packages/http_tracker_core/statistics/setup.rs

@@ -0,0 +1,54 @@
+//! Setup for the tracker statistics.
+//!
+//! The [`factory`] function builds the structs needed for handling the tracker metrics.
+use crate::packages::http_tracker_core::statistics;
+
+/// It builds the structs needed for handling the tracker metrics.
+///
+/// It returns:
+///
+/// - An statistics event [`Sender`](crate::packages::http_tracker_core::statistics::event::sender::Sender) that allows you to send events related to statistics.
+/// - An statistics [`Repository`](crate::packages::http_tracker_core::statistics::repository::Repository) which is an in-memory repository for the tracker metrics.
+///
+/// When the input argument `tracker_usage_statistics`is false the setup does not run the event listeners, consequently the statistics
+/// events are sent are received but not dispatched to the handler.
+#[must_use]
+pub fn factory(
+    tracker_usage_statistics: bool,
+) -> (
+    Option<Box<dyn statistics::event::sender::Sender>>,
+    statistics::repository::Repository,
+) {
+    let mut stats_event_sender = None;
+
+    let mut stats_tracker = statistics::keeper::Keeper::new();
+
+    if tracker_usage_statistics {
+        stats_event_sender = Some(stats_tracker.run_event_listener());
+    }
+
+    (stats_event_sender, stats_tracker.repository)
+}
+
+#[cfg(test)]
+mod test {
+    use super::factory;
+
+    #[tokio::test]
+    async fn should_not_send_any_event_when_statistics_are_disabled() {
+        let tracker_usage_statistics = false;
+
+        let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics);
+
+        assert!(stats_event_sender.is_none());
+    }
+
+    #[tokio::test]
+    async fn should_send_events_when_statistics_are_enabled() {
+        let tracker_usage_statistics = true;
+
+        let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics);
+
+        assert!(stats_event_sender.is_some());
+    }
+}

src/packages/mod.rs

@@ -1 +1,6 @@
+//! This module contains logic pending to be extracted into workspace packages.
+//!
+//! It will be moved to the directory `packages`.
+pub mod http_tracker_core;
 pub mod statistics;
+pub mod udp_tracker_core;

src/packages/udp_tracker_core/mod.rs

@@ -0,0 +1 @@
+pub mod statistics;

src/packages/udp_tracker_core/statistics/event/handler.rs

@@ -0,0 +1,154 @@
+use crate::packages::udp_tracker_core::statistics::event::{Event, UdpResponseKind};
+use crate::packages::udp_tracker_core::statistics::repository::Repository;
+
+pub async fn handle_event(event: Event, stats_repository: &Repository) {
+    match event {
+        // UDP
+        Event::UdpRequestAborted => {
+            stats_repository.increase_udp_requests_aborted().await;
+        }
+        Event::UdpRequestBanned => {
+            stats_repository.increase_udp_requests_banned().await;
+        }
+
+        // UDP4
+        Event::Udp4Request => {
+            stats_repository.increase_udp4_requests().await;
+        }
+        Event::Udp4Connect => {
+            stats_repository.increase_udp4_connections().await;
+        }
+        Event::Udp4Announce => {
+            stats_repository.increase_udp4_announces().await;
+        }
+        Event::Udp4Scrape => {
+            stats_repository.increase_udp4_scrapes().await;
+        }
+        Event::Udp4Response {
+            kind,
+            req_processing_time,
+        } => {
+            stats_repository.increase_udp4_responses().await;
+
+            match kind {
+                UdpResponseKind::Connect => {
+                    stats_repository
+                        .recalculate_udp_avg_connect_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Announce => {
+                    stats_repository
+                        .recalculate_udp_avg_announce_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Scrape => {
+                    stats_repository
+                        .recalculate_udp_avg_scrape_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Error => {}
+            }
+        }
+        Event::Udp4Error => {
+            stats_repository.increase_udp4_errors().await;
+        }
+
+        // UDP6
+        Event::Udp6Request => {
+            stats_repository.increase_udp6_requests().await;
+        }
+        Event::Udp6Connect => {
+            stats_repository.increase_udp6_connections().await;
+        }
+        Event::Udp6Announce => {
+            stats_repository.increase_udp6_announces().await;
+        }
+        Event::Udp6Scrape => {
+            stats_repository.increase_udp6_scrapes().await;
+        }
+        Event::Udp6Response {
+            kind: _,
+            req_processing_time: _,
+        } => {
+            stats_repository.increase_udp6_responses().await;
+        }
+        Event::Udp6Error => {
+            stats_repository.increase_udp6_errors().await;
+        }
+    }
+
+    tracing::debug!("stats: {:?}", stats_repository.get_stats().await);
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::packages::udp_tracker_core::statistics::event::handler::handle_event;
+    use crate::packages::udp_tracker_core::statistics::event::Event;
+    use crate::packages::udp_tracker_core::statistics::repository::Repository;
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_connections_counter_when_it_receives_a_udp4_connect_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp4Connect, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_connections_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_announces_counter_when_it_receives_a_udp4_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp4Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_announces_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_scrapes_counter_when_it_receives_a_udp4_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp4Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_scrapes_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp6_connections_counter_when_it_receives_a_udp6_connect_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp6Connect, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_connections_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp6_announces_counter_when_it_receives_a_udp6_announce_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp6Announce, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_announces_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp6_scrapes_counter_when_it_receives_a_udp6_scrape_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp6Scrape, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_scrapes_handled, 1);
+    }
+}

src/packages/udp_tracker_core/statistics/event/listener.rs

@@ -0,0 +1,11 @@
+use tokio::sync::mpsc;
+
+use super::handler::handle_event;
+use super::Event;
+use crate::packages::udp_tracker_core::statistics::repository::Repository;
+
+pub async fn dispatch_events(mut receiver: mpsc::Receiver<Event>, stats_repository: Repository) {
+    while let Some(event) = receiver.recv().await {
+        handle_event(event, &stats_repository).await;
+    }
+}

src/packages/udp_tracker_core/statistics/event/mod.rs

@@ -0,0 +1,47 @@
+use std::time::Duration;
+
+pub mod handler;
+pub mod listener;
+pub mod sender;
+
+/// An statistics event. It is used to collect tracker metrics.
+///
+/// - `Tcp` prefix means the event was triggered by the HTTP tracker
+/// - `Udp` prefix means the event was triggered by the UDP tracker
+/// - `4` or `6` prefixes means the IP version used by the peer
+/// - Finally the event suffix is the type of request: `announce`, `scrape` or `connection`
+///
+/// > NOTE: HTTP trackers do not use `connection` requests.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Event {
+    // code-review: consider one single event for request type with data: Event::Announce { scheme: HTTPorUDP, ip_version: V4orV6 }
+    // Attributes are enums too.
+    UdpRequestAborted,
+    UdpRequestBanned,
+    Udp4Request,
+    Udp4Connect,
+    Udp4Announce,
+    Udp4Scrape,
+    Udp4Response {
+        kind: UdpResponseKind,
+        req_processing_time: Duration,
+    },
+    Udp4Error,
+    Udp6Request,
+    Udp6Connect,
+    Udp6Announce,
+    Udp6Scrape,
+    Udp6Response {
+        kind: UdpResponseKind,
+        req_processing_time: Duration,
+    },
+    Udp6Error,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum UdpResponseKind {
+    Connect,
+    Announce,
+    Scrape,
+    Error,
+}

src/packages/udp_tracker_core/statistics/event/sender.rs

@@ -0,0 +1,29 @@
+use futures::future::BoxFuture;
+use futures::FutureExt;
+#[cfg(test)]
+use mockall::{automock, predicate::str};
+use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::SendError;
+
+use super::Event;
+
+/// A trait to allow sending statistics events
+#[cfg_attr(test, automock)]
+pub trait Sender: Sync + Send {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>>;
+}
+
+/// An [`statistics::EventSender`](crate::packages::udp_tracker_core::statistics::event::sender::Sender) implementation.
+///
+/// It uses a channel sender to send the statistic events. The channel is created by a
+/// [`statistics::Keeper`](crate::packages::udp_tracker_core::statistics::keeper::Keeper)
+#[allow(clippy::module_name_repetitions)]
+pub struct ChannelSender {
+    pub(crate) sender: mpsc::Sender<Event>,
+}
+
+impl Sender for ChannelSender {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>> {
+        async move { Some(self.sender.send(event).await) }.boxed()
+    }
+}

src/packages/udp_tracker_core/statistics/keeper.rs

@@ -0,0 +1,77 @@
+use tokio::sync::mpsc;
+
+use super::event::listener::dispatch_events;
+use super::event::sender::{ChannelSender, Sender};
+use super::event::Event;
+use super::repository::Repository;
+
+const CHANNEL_BUFFER_SIZE: usize = 65_535;
+
+/// The service responsible for keeping tracker metrics (listening to statistics events and handle them).
+///
+/// It actively listen to new statistics events. When it receives a new event
+/// it accordingly increases the counters.
+pub struct Keeper {
+    pub repository: Repository,
+}
+
+impl Default for Keeper {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Keeper {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            repository: Repository::new(),
+        }
+    }
+
+    #[must_use]
+    pub fn new_active_instance() -> (Box<dyn Sender>, Repository) {
+        let mut stats_tracker = Self::new();
+
+        let stats_event_sender = stats_tracker.run_event_listener();
+
+        (stats_event_sender, stats_tracker.repository)
+    }
+
+    pub fn run_event_listener(&mut self) -> Box<dyn Sender> {
+        let (sender, receiver) = mpsc::channel::<Event>(CHANNEL_BUFFER_SIZE);
+
+        let stats_repository = self.repository.clone();
+
+        tokio::spawn(async move { dispatch_events(receiver, stats_repository).await });
+
+        Box::new(ChannelSender { sender })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::packages::udp_tracker_core::statistics::event::Event;
+    use crate::packages::udp_tracker_core::statistics::keeper::Keeper;
+    use crate::packages::udp_tracker_core::statistics::metrics::Metrics;
+
+    #[tokio::test]
+    async fn should_contain_the_tracker_statistics() {
+        let stats_tracker = Keeper::new();
+
+        let stats = stats_tracker.repository.get_stats().await;
+
+        assert_eq!(stats.udp4_announces_handled, Metrics::default().udp4_announces_handled);
+    }
+
+    #[tokio::test]
+    async fn should_create_an_event_sender_to_send_statistical_events() {
+        let mut stats_tracker = Keeper::new();
+
+        let event_sender = stats_tracker.run_event_listener();
+
+        let result = event_sender.send_event(Event::Udp4Connect).await;
+
+        assert!(result.is_some());
+    }
+}

src/packages/udp_tracker_core/statistics/metrics.rs

@@ -0,0 +1,67 @@
+/// Metrics collected by the tracker.
+///
+/// - Number of connections handled
+/// - Number of `announce` requests handled
+/// - Number of `scrape` request handled
+///
+/// These metrics are collected for each connection type: UDP and HTTP
+/// and also for each IP version used by the peers: IPv4 and IPv6.
+#[derive(Debug, PartialEq, Default)]
+pub struct Metrics {
+    // UDP
+    /// Total number of UDP (UDP tracker) requests aborted.
+    pub udp_requests_aborted: u64,
+
+    /// Total number of UDP (UDP tracker) requests banned.
+    pub udp_requests_banned: u64,
+
+    /// Total number of banned IPs.
+    pub udp_banned_ips_total: u64,
+
+    /// Average rounded time spent processing UDP connect requests.
+    pub udp_avg_connect_processing_time_ns: u64,
+
+    /// Average rounded time spent processing UDP announce requests.
+    pub udp_avg_announce_processing_time_ns: u64,
+
+    /// Average rounded time spent processing UDP scrape requests.
+    pub udp_avg_scrape_processing_time_ns: u64,
+
+    // UDPv4
+    /// Total number of UDP (UDP tracker) requests from IPv4 peers.
+    pub udp4_requests: u64,
+
+    /// Total number of UDP (UDP tracker) connections from IPv4 peers.
+    pub udp4_connections_handled: u64,
+
+    /// Total number of UDP (UDP tracker) `announce` requests from IPv4 peers.
+    pub udp4_announces_handled: u64,
+
+    /// Total number of UDP (UDP tracker) `scrape` requests from IPv4 peers.
+    pub udp4_scrapes_handled: u64,
+
+    /// Total number of UDP (UDP tracker) responses from IPv4 peers.
+    pub udp4_responses: u64,
+
+    /// Total number of UDP (UDP tracker) `error` requests from IPv4 peers.
+    pub udp4_errors_handled: u64,
+
+    // UDPv6
+    /// Total number of UDP (UDP tracker) requests from IPv6 peers.
+    pub udp6_requests: u64,
+
+    /// Total number of UDP (UDP tracker) `connection` requests from IPv6 peers.
+    pub udp6_connections_handled: u64,
+
+    /// Total number of UDP (UDP tracker) `announce` requests from IPv6 peers.
+    pub udp6_announces_handled: u64,
+
+    /// Total number of UDP (UDP tracker) `scrape` requests from IPv6 peers.
+    pub udp6_scrapes_handled: u64,
+
+    /// Total number of UDP (UDP tracker) responses from IPv6 peers.
+    pub udp6_responses: u64,
+
+    /// Total number of UDP (UDP tracker) `error` requests from IPv6 peers.
+    pub udp6_errors_handled: u64,
+}

src/packages/udp_tracker_core/statistics/mod.rs

@@ -0,0 +1,6 @@
+pub mod event;
+pub mod keeper;
+pub mod metrics;
+pub mod repository;
+pub mod services;
+pub mod setup;

src/packages/udp_tracker_core/statistics/repository.rs

@@ -0,0 +1,173 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::{RwLock, RwLockReadGuard};
+
+use super::metrics::Metrics;
+
+/// A repository for the tracker 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 {
+        Self {
+            stats: Arc::new(RwLock::new(Metrics::default())),
+        }
+    }
+
+    pub async fn get_stats(&self) -> RwLockReadGuard<'_, Metrics> {
+        self.stats.read().await
+    }
+
+    pub async fn increase_udp_requests_aborted(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp_requests_aborted += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp_requests_banned(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp_requests_banned += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_requests(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_requests += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_connections(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_connections_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_announces(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_announces_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_scrapes(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_scrapes_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_responses(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_responses += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp4_errors(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp4_errors_handled += 1;
+        drop(stats_lock);
+    }
+
+    #[allow(clippy::cast_precision_loss)]
+    #[allow(clippy::cast_possible_truncation)]
+    #[allow(clippy::cast_sign_loss)]
+    pub async fn recalculate_udp_avg_connect_processing_time_ns(&self, req_processing_time: Duration) {
+        let mut stats_lock = self.stats.write().await;
+
+        let req_processing_time = req_processing_time.as_nanos() as f64;
+        let udp_connections_handled = (stats_lock.udp4_connections_handled + stats_lock.udp6_connections_handled) as f64;
+
+        let previous_avg = stats_lock.udp_avg_connect_processing_time_ns;
+
+        // Moving average: https://en.wikipedia.org/wiki/Moving_average
+        let new_avg = previous_avg as f64 + (req_processing_time - previous_avg as f64) / udp_connections_handled;
+
+        stats_lock.udp_avg_connect_processing_time_ns = new_avg.ceil() as u64;
+
+        drop(stats_lock);
+    }
+
+    #[allow(clippy::cast_precision_loss)]
+    #[allow(clippy::cast_possible_truncation)]
+    #[allow(clippy::cast_sign_loss)]
+    pub async fn recalculate_udp_avg_announce_processing_time_ns(&self, req_processing_time: Duration) {
+        let mut stats_lock = self.stats.write().await;
+
+        let req_processing_time = req_processing_time.as_nanos() as f64;
+
+        let udp_announces_handled = (stats_lock.udp4_announces_handled + stats_lock.udp6_announces_handled) as f64;
+
+        let previous_avg = stats_lock.udp_avg_announce_processing_time_ns;
+
+        // Moving average: https://en.wikipedia.org/wiki/Moving_average
+        let new_avg = previous_avg as f64 + (req_processing_time - previous_avg as f64) / udp_announces_handled;
+
+        stats_lock.udp_avg_announce_processing_time_ns = new_avg.ceil() as u64;
+
+        drop(stats_lock);
+    }
+
+    #[allow(clippy::cast_precision_loss)]
+    #[allow(clippy::cast_possible_truncation)]
+    #[allow(clippy::cast_sign_loss)]
+    pub async fn recalculate_udp_avg_scrape_processing_time_ns(&self, req_processing_time: Duration) {
+        let mut stats_lock = self.stats.write().await;
+
+        let req_processing_time = req_processing_time.as_nanos() as f64;
+        let udp_scrapes_handled = (stats_lock.udp4_scrapes_handled + stats_lock.udp6_scrapes_handled) as f64;
+
+        let previous_avg = stats_lock.udp_avg_scrape_processing_time_ns;
+
+        // Moving average: https://en.wikipedia.org/wiki/Moving_average
+        let new_avg = previous_avg as f64 + (req_processing_time - previous_avg as f64) / udp_scrapes_handled;
+
+        stats_lock.udp_avg_scrape_processing_time_ns = new_avg.ceil() as u64;
+
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_requests(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_requests += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_connections(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_connections_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_announces(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_announces_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_scrapes(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_scrapes_handled += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_responses(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_responses += 1;
+        drop(stats_lock);
+    }
+
+    pub async fn increase_udp6_errors(&self) {
+        let mut stats_lock = self.stats.write().await;
+        stats_lock.udp6_errors_handled += 1;
+        drop(stats_lock);
+    }
+}

src/packages/udp_tracker_core/statistics/services.rs

@@ -0,0 +1,146 @@
+//! Statistics services.
+//!
+//! It includes:
+//!
+//! - A [`factory`](crate::packages::udp_tracker_core::statistics::setup::factory) function to build the structs needed to collect the tracker metrics.
+//! - A [`get_metrics`] service to get the tracker [`metrics`](crate::packages::udp_tracker_core::statistics::metrics::Metrics).
+//!
+//! Tracker metrics are collected using a Publisher-Subscribe pattern.
+//!
+//! The factory function builds two structs:
+//!
+//! - An statistics event [`Sender`](crate::packages::udp_tracker_core::statistics::event::sender::Sender)
+//! - An statistics [`Repository`]
+//!
+//! ```text
+//! let (stats_event_sender, stats_repository) = factory(tracker_usage_statistics);
+//! ```
+//!
+//! The statistics repository is responsible for storing the metrics in memory.
+//! The statistics event sender allows sending events related to metrics.
+//! There is an event listener that is receiving all the events and processing them with an event handler.
+//! Then, the event handler updates the metrics depending on the received event.
+//!
+//! For example, if you send the event [`Event::Udp4Connect`](crate::packages::udp_tracker_core::statistics::event::Event::Udp4Connect):
+//!
+//! ```text
+//! let result = event_sender.send_event(Event::Udp4Connect).await;
+//! ```
+//!
+//! Eventually the counter for UDP connections from IPv4 peers will be increased.
+//!
+//! ```rust,no_run
+//! pub struct Metrics {
+//!     // ...
+//!     pub udp4_connections_handled: u64,  // This will be incremented
+//!     // ...
+//! }
+//! ```
+use std::sync::Arc;
+
+use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
+use packages::udp_tracker_core::statistics::metrics::Metrics;
+use packages::udp_tracker_core::statistics::repository::Repository;
+use tokio::sync::RwLock;
+use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+
+use crate::packages;
+use crate::servers::udp::server::banning::BanService;
+
+/// All the metrics collected by the tracker.
+#[derive(Debug, PartialEq)]
+pub struct TrackerMetrics {
+    /// Domain level metrics.
+    ///
+    /// General metrics for all torrents (number of seeders, leechers, etcetera)
+    pub torrents_metrics: TorrentsMetrics,
+
+    /// Application level metrics. Usage statistics/metrics.
+    ///
+    /// Metrics about how the tracker is been used (number of udp announce requests, etcetera)
+    pub protocol_metrics: Metrics,
+}
+
+/// It returns all the [`TrackerMetrics`]
+pub async fn get_metrics(
+    in_memory_torrent_repository: Arc<InMemoryTorrentRepository>,
+    ban_service: Arc<RwLock<BanService>>,
+    stats_repository: Arc<Repository>,
+) -> TrackerMetrics {
+    let torrents_metrics = in_memory_torrent_repository.get_torrents_metrics();
+    let stats = stats_repository.get_stats().await;
+    let udp_banned_ips_total = ban_service.read().await.get_banned_ips_total();
+
+    TrackerMetrics {
+        torrents_metrics,
+        protocol_metrics: Metrics {
+            // UDP
+            udp_requests_aborted: stats.udp_requests_aborted,
+            udp_requests_banned: stats.udp_requests_banned,
+            udp_banned_ips_total: udp_banned_ips_total as u64,
+            udp_avg_connect_processing_time_ns: stats.udp_avg_connect_processing_time_ns,
+            udp_avg_announce_processing_time_ns: stats.udp_avg_announce_processing_time_ns,
+            udp_avg_scrape_processing_time_ns: stats.udp_avg_scrape_processing_time_ns,
+            // UDPv4
+            udp4_requests: stats.udp4_requests,
+            udp4_connections_handled: stats.udp4_connections_handled,
+            udp4_announces_handled: stats.udp4_announces_handled,
+            udp4_scrapes_handled: stats.udp4_scrapes_handled,
+            udp4_responses: stats.udp4_responses,
+            udp4_errors_handled: stats.udp4_errors_handled,
+            // UDPv6
+            udp6_requests: stats.udp6_requests,
+            udp6_connections_handled: stats.udp6_connections_handled,
+            udp6_announces_handled: stats.udp6_announces_handled,
+            udp6_scrapes_handled: stats.udp6_scrapes_handled,
+            udp6_responses: stats.udp6_responses,
+            udp6_errors_handled: stats.udp6_errors_handled,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
+    use bittorrent_tracker_core::{self};
+    use tokio::sync::RwLock;
+    use torrust_tracker_configuration::Configuration;
+    use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+    use torrust_tracker_test_helpers::configuration;
+
+    use crate::packages::udp_tracker_core::statistics;
+    use crate::packages::udp_tracker_core::statistics::services::{get_metrics, TrackerMetrics};
+    use crate::servers::udp::server::banning::BanService;
+    use crate::servers::udp::server::launcher::MAX_CONNECTION_ID_ERRORS_PER_IP;
+
+    pub fn tracker_configuration() -> Configuration {
+        configuration::ephemeral()
+    }
+
+    #[tokio::test]
+    async fn the_statistics_service_should_return_the_tracker_metrics() {
+        let config = tracker_configuration();
+
+        let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
+        let (_stats_event_sender, stats_repository) = statistics::setup::factory(config.core.tracker_usage_statistics);
+        let stats_repository = Arc::new(stats_repository);
+        let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP)));
+
+        let tracker_metrics = get_metrics(
+            in_memory_torrent_repository.clone(),
+            ban_service.clone(),
+            stats_repository.clone(),
+        )
+        .await;
+
+        assert_eq!(
+            tracker_metrics,
+            TrackerMetrics {
+                torrents_metrics: TorrentsMetrics::default(),
+                protocol_metrics: statistics::metrics::Metrics::default(),
+            }
+        );
+    }
+}

src/packages/udp_tracker_core/statistics/setup.rs

@@ -0,0 +1,54 @@
+//! Setup for the tracker statistics.
+//!
+//! The [`factory`] function builds the structs needed for handling the tracker metrics.
+use crate::packages::udp_tracker_core::statistics;
+
+/// It builds the structs needed for handling the tracker metrics.
+///
+/// It returns:
+///
+/// - An statistics event [`Sender`](crate::packages::udp_tracker_core::statistics::event::sender::Sender) that allows you to send events related to statistics.
+/// - An statistics [`Repository`](crate::packages::udp_tracker_core::statistics::repository::Repository) which is an in-memory repository for the tracker metrics.
+///
+/// When the input argument `tracker_usage_statistics`is false the setup does not run the event listeners, consequently the statistics
+/// events are sent are received but not dispatched to the handler.
+#[must_use]
+pub fn factory(
+    tracker_usage_statistics: bool,
+) -> (
+    Option<Box<dyn statistics::event::sender::Sender>>,
+    statistics::repository::Repository,
+) {
+    let mut stats_event_sender = None;
+
+    let mut stats_tracker = statistics::keeper::Keeper::new();
+
+    if tracker_usage_statistics {
+        stats_event_sender = Some(stats_tracker.run_event_listener());
+    }
+
+    (stats_event_sender, stats_tracker.repository)
+}
+
+#[cfg(test)]
+mod test {
+    use super::factory;
+
+    #[tokio::test]
+    async fn should_not_send_any_event_when_statistics_are_disabled() {
+        let tracker_usage_statistics = false;
+
+        let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics);
+
+        assert!(stats_event_sender.is_none());
+    }
+
+    #[tokio::test]
+    async fn should_send_events_when_statistics_are_enabled() {
+        let tracker_usage_statistics = true;
+
+        let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics);
+
+        assert!(stats_event_sender.is_some());
+    }
+}