Merge #1186: Overhaul core Tracker: extract stats

2a0bc47 refactor: [#1184] remove stats functionality from core tracker (Jose Celano)
c415430 refactor: [#1184] inject stats event sender and repository into the core tracker (Jose Celano)

Pull request description:

  Overhaul core Tracker: extract stats

ACKs for top commit:
  josecelano:
    ACK 2a0bc47

Tree-SHA512: b72acb6f25d5ea642d9889fa7da3e6602c7362c1b6381af94bac848eda5b4f1a1b1fd364660fa93f703a48bc874b883db2d14ce30e4c54352f754c72132fcac3

Author: josecelano

src/app.rs

@@ -29,6 +29,8 @@ use torrust_tracker_configuration::Configuration;
 use tracing::instrument;
 
 use crate::bootstrap::jobs::{health_check_api, http_tracker, torrent_cleanup, tracker_apis, udp_tracker};
+use crate::core::statistics::event::sender::Sender;
+use crate::core::statistics::repository::Repository;
 use crate::servers::registar::Registar;
 use crate::servers::udp::server::banning::BanService;
 use crate::{core, servers};
@@ -39,11 +41,13 @@ use crate::{core, servers};
 ///
 /// - Can't retrieve tracker keys from database.
 /// - Can't load whitelist from database.
-#[instrument(skip(config, tracker, ban_service))]
+#[instrument(skip(config, tracker, ban_service, stats_event_sender, stats_repository))]
 pub async fn start(
     config: &Configuration,
     tracker: Arc<core::Tracker>,
     ban_service: Arc<RwLock<BanService>>,
+    stats_event_sender: Arc<Option<Box<dyn Sender>>>,
+    stats_repository: Arc<Repository>,
 ) -> Vec<JoinHandle<()>> {
     if config.http_api.is_none()
         && (config.udp_trackers.is_none() || config.udp_trackers.as_ref().map_or(true, std::vec::Vec::is_empty))
@@ -83,7 +87,14 @@ pub async fn start(
                 );
             } else {
                 jobs.push(
-                    udp_tracker::start_job(udp_tracker_config, tracker.clone(), ban_service.clone(), registar.give_form()).await,
+                    udp_tracker::start_job(
+                        udp_tracker_config,
+                        tracker.clone(),
+                        stats_event_sender.clone(),
+                        ban_service.clone(),
+                        registar.give_form(),
+                    )
+                    .await,
                 );
             }
         }
@@ -97,6 +108,7 @@ pub async fn start(
             if let Some(job) = http_tracker::start_job(
                 http_tracker_config,
                 tracker.clone(),
+                stats_event_sender.clone(),
                 registar.give_form(),
                 servers::http::Version::V1,
             )
@@ -115,6 +127,8 @@ pub async fn start(
             http_api_config,
             tracker.clone(),
             ban_service.clone(),
+            stats_event_sender.clone(),
+            stats_repository.clone(),
             registar.give_form(),
             servers::apis::Version::V1,
         )

src/bootstrap/app.rs

@@ -22,7 +22,9 @@ use tracing::instrument;
 use super::config::initialize_configuration;
 use crate::bootstrap;
 use crate::core::databases::Database;
-use crate::core::services::{initialize_database, initialize_whitelist, tracker_factory};
+use crate::core::services::{initialize_database, initialize_whitelist, statistics, tracker_factory};
+use crate::core::statistics::event::sender::Sender;
+use crate::core::statistics::repository::Repository;
 use crate::core::whitelist::WhiteListManager;
 use crate::core::Tracker;
 use crate::servers::udp::server::banning::BanService;
@@ -36,8 +38,15 @@ use crate::shared::crypto::keys::{self, Keeper as _};
 ///
 /// Setup can file if the configuration is invalid.
 #[must_use]
+#[allow(clippy::type_complexity)]
 #[instrument(skip())]
-pub fn setup() -> (Configuration, Arc<Tracker>, Arc<RwLock<BanService>>) {
+pub fn setup() -> (
+    Configuration,
+    Arc<Tracker>,
+    Arc<RwLock<BanService>>,
+    Arc<Option<Box<dyn Sender>>>,
+    Arc<Repository>,
+) {
     #[cfg(not(test))]
     check_seed();
 
@@ -47,13 +56,19 @@ pub fn setup() -> (Configuration, Arc<Tracker>, Arc<RwLock<BanService>>) {
         panic!("Configuration error: {e}");
     }
 
-    let tracker = initialize_with_configuration(&configuration);
+    // Initialize services
+
+    let (stats_event_sender, stats_repository) = statistics::setup::factory(configuration.core.tracker_usage_statistics);
+    let stats_event_sender = Arc::new(stats_event_sender);
+    let stats_repository = Arc::new(stats_repository);
 
     let udp_ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP)));
 
+    let tracker = initialize_with_configuration(&configuration);
+
     tracing::info!("Configuration:\n{}", configuration.clone().mask_secrets().to_json());
 
-    (configuration, tracker, udp_ban_service)
+    (configuration, tracker, udp_ban_service, stats_event_sender, stats_repository)
 }
 
 /// checks if the seed is the instance seed in production.
@@ -112,6 +127,7 @@ pub fn initialize_tracker(config: &Configuration) -> Tracker {
     tracker_factory(config, &database, &whitelist_manager)
 }
 
+#[allow(clippy::type_complexity)]
 #[must_use]
 pub fn initialize_tracker_dependencies(config: &Configuration) -> (Arc<Box<dyn Database>>, Arc<WhiteListManager>) {
     let database = initialize_database(config);

src/bootstrap/jobs/http_tracker.rs

@@ -19,7 +19,8 @@ use torrust_tracker_configuration::HttpTracker;
 use tracing::instrument;
 
 use super::make_rust_tls;
-use crate::core;
+use crate::core::statistics::event::sender::Sender;
+use crate::core::{self, statistics};
 use crate::servers::http::server::{HttpServer, Launcher};
 use crate::servers::http::Version;
 use crate::servers::registar::ServiceRegistrationForm;
@@ -33,10 +34,11 @@ use crate::servers::registar::ServiceRegistrationForm;
 ///
 /// It would panic if the `config::HttpTracker` struct would contain inappropriate values.
 ///
-#[instrument(skip(config, tracker, form))]
+#[instrument(skip(config, tracker, stats_event_sender, form))]
 pub async fn start_job(
     config: &HttpTracker,
     tracker: Arc<core::Tracker>,
+    stats_event_sender: Arc<Option<Box<dyn Sender>>>,
     form: ServiceRegistrationForm,
     version: Version,
 ) -> Option<JoinHandle<()>> {
@@ -47,20 +49,21 @@ pub async fn start_job(
         .map(|tls| tls.expect("it should have a valid http tracker tls configuration"));
 
     match version {
-        Version::V1 => Some(start_v1(socket, tls, tracker.clone(), form).await),
+        Version::V1 => Some(start_v1(socket, tls, tracker.clone(), stats_event_sender.clone(), form).await),
     }
 }
 
 #[allow(clippy::async_yields_async)]
-#[instrument(skip(socket, tls, tracker, form))]
+#[instrument(skip(socket, tls, tracker, stats_event_sender, form))]
 async fn start_v1(
     socket: SocketAddr,
     tls: Option<RustlsConfig>,
     tracker: Arc<core::Tracker>,
+    stats_event_sender: Arc<Option<Box<dyn statistics::event::sender::Sender>>>,
     form: ServiceRegistrationForm,
 ) -> JoinHandle<()> {
     let server = HttpServer::new(Launcher::new(socket, tls))
-        .start(tracker, form)
+        .start(tracker, stats_event_sender, form)
         .await
         .expect("it should be able to start to the http tracker");
 
@@ -85,6 +88,7 @@ mod tests {
 
     use crate::bootstrap::app::initialize_with_configuration;
     use crate::bootstrap::jobs::http_tracker::start_job;
+    use crate::core::services::statistics;
     use crate::servers::http::Version;
     use crate::servers::registar::Registar;
 
@@ -93,10 +97,12 @@ mod tests {
         let cfg = Arc::new(ephemeral_public());
         let http_tracker = cfg.http_trackers.clone().expect("missing HTTP tracker configuration");
         let config = &http_tracker[0];
+        let (stats_event_sender, _stats_repository) = statistics::setup::factory(cfg.core.tracker_usage_statistics);
+        let stats_event_sender = Arc::new(stats_event_sender);
         let tracker = initialize_with_configuration(&cfg);
         let version = Version::V1;
 
-        start_job(config, tracker, Registar::default().give_form(), version)
+        start_job(config, tracker, stats_event_sender, Registar::default().give_form(), version)
             .await
             .expect("it should be able to join to the http tracker start-job");
     }

src/bootstrap/jobs/tracker_apis.rs

@@ -31,6 +31,8 @@ use tracing::instrument;
 
 use super::make_rust_tls;
 use crate::core;
+use crate::core::statistics::event::sender::Sender;
+use crate::core::statistics::repository::Repository;
 use crate::servers::apis::server::{ApiServer, Launcher};
 use crate::servers::apis::Version;
 use crate::servers::registar::ServiceRegistrationForm;
@@ -56,11 +58,13 @@ pub struct ApiServerJobStarted();
 /// It would panic if unable to send the  `ApiServerJobStarted` notice.
 ///
 ///
-#[instrument(skip(config, tracker, ban_service, form))]
+#[instrument(skip(config, tracker, ban_service, stats_event_sender, stats_repository, form))]
 pub async fn start_job(
     config: &HttpApi,
     tracker: Arc<core::Tracker>,
     ban_service: Arc<RwLock<BanService>>,
+    stats_event_sender: Arc<Option<Box<dyn Sender>>>,
+    stats_repository: Arc<Repository>,
     form: ServiceRegistrationForm,
     version: Version,
 ) -> Option<JoinHandle<()>> {
@@ -73,22 +77,53 @@ pub async fn start_job(
     let access_tokens = Arc::new(config.access_tokens.clone());
 
     match version {
-        Version::V1 => Some(start_v1(bind_to, tls, tracker.clone(), ban_service.clone(), form, access_tokens).await),
+        Version::V1 => Some(
+            start_v1(
+                bind_to,
+                tls,
+                tracker.clone(),
+                ban_service.clone(),
+                stats_event_sender.clone(),
+                stats_repository.clone(),
+                form,
+                access_tokens,
+            )
+            .await,
+        ),
     }
 }
 
 #[allow(clippy::async_yields_async)]
-#[instrument(skip(socket, tls, tracker, ban_service, form, access_tokens))]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip(
+    socket,
+    tls,
+    tracker,
+    ban_service,
+    stats_event_sender,
+    stats_repository,
+    form,
+    access_tokens
+))]
 async fn start_v1(
     socket: SocketAddr,
     tls: Option<RustlsConfig>,
     tracker: Arc<core::Tracker>,
     ban_service: Arc<RwLock<BanService>>,
+    stats_event_sender: Arc<Option<Box<dyn Sender>>>,
+    stats_repository: Arc<Repository>,
     form: ServiceRegistrationForm,
     access_tokens: Arc<AccessTokens>,
 ) -> JoinHandle<()> {
     let server = ApiServer::new(Launcher::new(socket, tls))
-        .start(tracker, ban_service, form, access_tokens)
+        .start(
+            tracker,
+            stats_event_sender,
+            stats_repository,
+            ban_service,
+            form,
+            access_tokens,
+        )
         .await
         .expect("it should be able to start to the tracker api");
 
@@ -107,6 +142,7 @@ mod tests {
 
     use crate::bootstrap::app::initialize_with_configuration;
     use crate::bootstrap::jobs::tracker_apis::start_job;
+    use crate::core::services::statistics;
     use crate::servers::apis::Version;
     use crate::servers::registar::Registar;
     use crate::servers::udp::server::banning::BanService;
@@ -116,12 +152,26 @@ mod tests {
     async fn it_should_start_http_tracker() {
         let cfg = Arc::new(ephemeral_public());
         let config = &cfg.http_api.clone().unwrap();
-        let tracker = initialize_with_configuration(&cfg);
+
         let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP)));
+        let (stats_event_sender, stats_repository) = statistics::setup::factory(cfg.core.tracker_usage_statistics);
+        let stats_event_sender = Arc::new(stats_event_sender);
+        let stats_repository = Arc::new(stats_repository);
+
+        let tracker = initialize_with_configuration(&cfg);
+
         let version = Version::V1;
 
-        start_job(config, tracker, ban_service, Registar::default().give_form(), version)
-            .await
-            .expect("it should be able to join to the tracker api start-job");
+        start_job(
+            config,
+            tracker,
+            ban_service,
+            stats_event_sender,
+            stats_repository,
+            Registar::default().give_form(),
+            version,
+        )
+        .await
+        .expect("it should be able to join to the tracker api start-job");
     }
 }

src/bootstrap/jobs/udp_tracker.rs

@@ -14,6 +14,7 @@ use torrust_tracker_configuration::UdpTracker;
 use tracing::instrument;
 
 use crate::core;
+use crate::core::statistics::event::sender::Sender;
 use crate::servers::registar::ServiceRegistrationForm;
 use crate::servers::udp::server::banning::BanService;
 use crate::servers::udp::server::spawner::Spawner;
@@ -31,18 +32,19 @@ use crate::servers::udp::UDP_TRACKER_LOG_TARGET;
 /// It will panic if the task did not finish successfully.
 #[must_use]
 #[allow(clippy::async_yields_async)]
-#[instrument(skip(config, tracker, ban_service, form))]
+#[instrument(skip(config, tracker, stats_event_sender, ban_service, form))]
 pub async fn start_job(
     config: &UdpTracker,
     tracker: Arc<core::Tracker>,
+    stats_event_sender: Arc<Option<Box<dyn Sender>>>,
     ban_service: Arc<RwLock<BanService>>,
     form: ServiceRegistrationForm,
 ) -> JoinHandle<()> {
     let bind_to = config.bind_address;
     let cookie_lifetime = config.cookie_lifetime;
 
     let server = Server::new(Spawner::new(bind_to))
-        .start(tracker, ban_service, form, cookie_lifetime)
+        .start(tracker, stats_event_sender, ban_service, form, cookie_lifetime)
         .await
         .expect("it should be able to start the udp tracker");
 

src/console/profiling.rs

@@ -179,9 +179,9 @@ pub async fn run() {
         return;
     };
 
-    let (config, tracker, ban_service) = bootstrap::app::setup();
+    let (config, tracker, ban_service, stats_event_sender, stats_repository) = bootstrap::app::setup();
 
-    let jobs = app::start(&config, tracker, ban_service).await;
+    let jobs = app::start(&config, tracker, ban_service, stats_event_sender, stats_repository).await;
 
     // Run the tracker for a fixed duration
     let run_duration = sleep(Duration::from_secs(duration_secs));