Merge #1513: Overhaul stats: Emit events from the torrent-repository package and collect metrics

d154b2a refactor: [#1358] clean Swarm type (Jose Celano)
b3b0b71 refactor: [#1358] Swarm, cleaning upsert_peer method (Jose Celano)
47d1eab refactor: [#1358] Swarm tests to use new mock helpers (Jose Celano)
b13797e test: [#1358] add tests for events in torrent-repository pkg (Jose Celano)
f71211f test: [#1358] add tests to torrust_tracker_torrent_repository::swarms::Swarms (Jose Celano)
3d7e6ff test: [#1358] add tests to torrust_tracker_torrent_repository::swarm::Swarm (Jose Celano)
0e38707 fix: [#1358] revert Hash impl for Swarm (Jose Celano)
c9a893c refactor: [#1358] rename metrics for clarity (Jose Celano)
8ee258e refactor: [#1358] use the new field info-hash as ID for the Swarm (Hash,PartialEq) (Jose Celano)
dfba00c feat: [#1358] allow disabling the event sender in the torrent-repository pkg (Jose Celano)
60c00e8 feat: [#1358] add info-hash to all torrent-repository events (Jose Celano)
c706a1b refactor: [#1358] move logs (Jose Celano)
daba8a0 feat: [#1358] new metric in torrent-repository: total number of downloads (Jose Celano)
01a9970 feat: [#1358] new metric in torrent-repository: total number of peers (Jose Celano)
269d273 refactor: [#1358] rename metric (Jose Celano)
ba2033b fix: [#1358] trigger PeerRemoved event when peer is removed due to inactivity (Jose Celano)
d47483f feat: [#1358] new metric in torrent-repository: total number of torrents (Jose Celano)
29a2dfd dev: change default config (Jose Celano)
1eb545c feat: [#1358] remove persistent metric from torrent-repository pkg (Jose Celano)
6d95d1a refactor: [#1358] inject event sender in Swarm type (Jose Celano)
2c479a1 refactor: [#1358] inject event sender in Swarms type (Jose Celano)
68b930d feat: [#1495] expose new torrent-repositoru metrics via the REST API (Jose Celano)
41f4022 feat: [#1358] inject Swarms into InMemoryTorrentRepository in testing code (Jose Celano)
95766bb feat: [#1358] inject Swarms into InMemoryTorrentRepository in production code (Jose Celano)
f986bda feat: [#1358] add the  and run the event listener when the tracker starts (Jose Celano)
2522ad4 feat: [#1358] basic scaffolding for events in torrent-repository pkg (Jose Celano)

Pull request description:

  Emit events from the `torrent-repository package` and collect metrics. In this PR, I'm only planning to include four metrics. The ones we have in the current `stats` endpoint:

  ```json
  {
    "torrents": 387044,
    "seeders": 183887,
    "completed": 1260584,
    "leechers": 249136,
  }
  ```

  However, I will rename them to:

  ```
  {
    "torrent_repository_per_session_all_instances_all_torrents_total": 387044,
    "torrent_repository_per_session_all_instances_all_seeders_total": 183887,
    "torrent_repository_per_session_all_instances_all_torrents_downloads_total": 1260584,
    "torrent_repository_per_session_all_instances_all_leechers_total": 249136,
  }
  ```

  **UPDATE:** I will not use those long names but "labels". It's only to explain what they represent.

  - per session: since the main tracker process started.
  - all instances: all tracker servers (UDP or HTTP) running on different ports.
  - all torrents: all swarms.

  Becuase the current names are too ambiguous. See #1502 (comment).

  ### How to test

  1. Run the tracker.
  2. Use the tracker client to make an announce request.

    ```
    cargo run -p torrust-tracker-client --bin udp_tracker_client announce udp://127.0.0.1:6969 443c7602b4fde83d1154d6d9da48808418b181b6 | jq
    ```
  3. Use the REST API to get the metric value.

    ```
    curl -s "http://localhost:1212/api/v1/metrics?token=MyAccessToken&format=prometheus" | grep torrent_repository_peers_total
    ```

  ### Subtasks

  - [x] Add basic event scaffolding in the `torrent-respository` package.
  - [x] Run the event listener for the torrent-repository package when the tracker starts.
    - [x] Add the `TorrentRepositoryContainer` and run the listener when the tracker starts.
    - [x] In production code: use the `Swarms` service provided by the `TorrentRepositoryContainer` in the `TrackerCoreContainer` (otherwise they will be unrelated).
    - [x] In testing code: use the `Swarms` service provided by the `TorrentRepositoryContainer` in the `TrackerCoreContainer` (otherwise they will be unrelated).
  - [x] Expose the new metrics via the `metrics` API endpoint (even if it will have no value yet).
  - [x] Inject the event sender in `Swarms` and `Swarm` type to be able to send events from them.
    - [x] In `Swarms` type.
    - [x] In `Swarm` type.
  - [x] #1516
  - [x] Star triggering events and process them to update the metrics in the metrics collection. One event and the corresponding affected metrics at the time.
    - [x] `TorrentAdded` -> Increase torrents counter (all instances, all torrents)
    - [x] `Torrentremoved` -> Decrease torrents counter (all instances, all torrents)
    - [x] `PeerAdded` -> Increase peers counter (total, seeders, leechers)
    - [x] `PeerRemoved` -> Decrease peers counter (total, seeders, leechers)
    - [x] `PeerUdpated` -> Decrease peers counter (total, seeders, leechers)
    - [x] `PeerDownloadCompleted` -> Increase torrent downloads counter (all instances, all torrents)
  - [x] Add tests to check that events have been sent.
  - [x] Add tests to check that metrics are updated after receiving the event.

ACKs for top commit:
  josecelano:
    ACK d154b2a

Tree-SHA512: d4946847dae38b51bc6027b4de25d7e2ad5b1042294913008f503c590c81e586ed083ab980613d647a0f867601f7bd482f4c004628c01cdfffc9d8b810bb670e

Author: josecelano

Cargo.lock

@@ -55,6 +55,7 @@ torrust-rest-tracker-api-core = { version = "3.0.0-develop", path = "packages/re
 torrust-server-lib = { version = "3.0.0-develop", path = "packages/server-lib" }
 torrust-tracker-clock = { version = "3.0.0-develop", path = "packages/clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "packages/configuration" }
+torrust-tracker-torrent-repository = { version = "3.0.0-develop", path = "packages/torrent-repository" }
 torrust-udp-tracker-server = { version = "3.0.0-develop", path = "packages/udp-tracker-server" }
 tracing = "0"
 tracing-subscriber = { version = "0", features = ["json"] }

Cargo.toml

@@ -33,6 +33,7 @@ torrust-server-lib = { version = "3.0.0-develop", path = "../server-lib" }
 torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
 torrust-tracker-primitives = { version = "3.0.0-develop", path = "../primitives" }
+torrust-tracker-torrent-repository = { version = "3.0.0-develop", path = "../torrent-repository" }
 tower = { version = "0", features = ["timeout"] }
 tower-http = { version = "0", features = ["compression-full", "cors", "propagate-header", "request-id", "trace"] }
 tracing = "0"

packages/axum-http-tracker-server/Cargo.toml

@@ -10,6 +10,7 @@ use torrust_axum_server::tsl::make_rust_tls;
 use torrust_server_lib::registar::Registar;
 use torrust_tracker_configuration::{logging, Configuration};
 use torrust_tracker_primitives::peer;
+use torrust_tracker_torrent_repository::container::TorrentRepositoryContainer;
 
 use crate::server::{HttpServer, Launcher, Running, Stopped};
 
@@ -24,12 +25,12 @@ pub struct Environment<S> {
 
 impl<S> Environment<S> {
     /// Add a torrent to the tracker
-    pub fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
-        let _number_of_downloads_increased = self
-            .container
+    pub async fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) -> bool {
+        self.container
             .tracker_core_container
             .in_memory_torrent_repository
-            .upsert_peer(info_hash, peer, None);
+            .upsert_peer(info_hash, peer, None)
+            .await
     }
 }
 
@@ -143,7 +144,15 @@ impl EnvContainer {
             .expect("missing HTTP tracker configuration");
         let http_tracker_config = Arc::new(http_tracker_config[0].clone());
 
-        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
+        let torrent_repository_container = Arc::new(TorrentRepositoryContainer::initialize(
+            configuration.core.tracker_usage_statistics.into(),
+        ));
+
+        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize_from(
+            &core_config,
+            &torrent_repository_container,
+        ));
+
         let http_tracker_container =
             HttpTrackerCoreContainer::initialize_from_tracker_core(&tracker_core_container, &http_tracker_config);
 

packages/axum-http-tracker-server/src/environment.rs

@@ -260,6 +260,7 @@ mod tests {
     use torrust_server_lib::registar::Registar;
     use torrust_tracker_configuration::{logging, Configuration};
     use torrust_tracker_test_helpers::configuration::ephemeral_public;
+    use torrust_tracker_torrent_repository::container::TorrentRepositoryContainer;
 
     use crate::server::{HttpServer, Launcher};
 
@@ -279,7 +280,7 @@ mod tests {
         let http_core_broadcaster = Broadcaster::default();
         let http_stats_repository = Arc::new(Repository::new());
         let http_stats_event_bus = Arc::new(EventBus::new(
-            configuration.core.tracker_usage_statistics,
+            configuration.core.tracker_usage_statistics.into(),
             http_core_broadcaster.clone(),
         ));
 
@@ -289,7 +290,14 @@ mod tests {
             let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
         }
 
-        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
+        let torrent_repository_container = Arc::new(TorrentRepositoryContainer::initialize(
+            configuration.core.tracker_usage_statistics.into(),
+        ));
+
+        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize_from(
+            &core_config,
+            &torrent_repository_container,
+        ));
 
         let announce_service = Arc::new(AnnounceService::new(
             tracker_core_container.core_config.clone(),

packages/axum-http-tracker-server/src/server.rs

@@ -168,7 +168,7 @@ mod tests {
         let http_core_broadcaster = Broadcaster::default();
         let http_stats_repository = Arc::new(Repository::new());
         let http_stats_event_bus = Arc::new(EventBus::new(
-            config.core.tracker_usage_statistics,
+            config.core.tracker_usage_statistics.into(),
             http_core_broadcaster.clone(),
         ));
 

packages/axum-http-tracker-server/src/v1/handlers/announce.rs

@@ -139,7 +139,7 @@ mod tests {
         let http_core_broadcaster = Broadcaster::default();
         let http_stats_repository = Arc::new(Repository::new());
         let http_stats_event_bus = Arc::new(EventBus::new(
-            config.core.tracker_usage_statistics,
+            config.core.tracker_usage_statistics.into(),
             http_core_broadcaster.clone(),
         ));
 

packages/axum-http-tracker-server/src/v1/handlers/scrape.rs

@@ -474,7 +474,7 @@ mod for_all_config_modes {
             let previously_announced_peer = PeerBuilder::default().with_peer_id(&PeerId(*b"-qB00000000000000001")).build();
 
             // Add the Peer 1
-            env.add_torrent_peer(&info_hash, &previously_announced_peer);
+            env.add_torrent_peer(&info_hash, &previously_announced_peer).await;
 
             // Announce the new Peer 2. This new peer is non included on the response peer list
             let response = Client::new(*env.bind_address())
@@ -517,7 +517,7 @@ mod for_all_config_modes {
                 .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                 .with_peer_addr(&SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0x69, 0x69, 0x69, 0x69)), 8080))
                 .build();
-            env.add_torrent_peer(&info_hash, &peer_using_ipv4);
+            env.add_torrent_peer(&info_hash, &peer_using_ipv4).await;
 
             // Announce a peer using IPV6
             let peer_using_ipv6 = PeerBuilder::default()
@@ -527,7 +527,7 @@ mod for_all_config_modes {
                     8080,
                 ))
                 .build();
-            env.add_torrent_peer(&info_hash, &peer_using_ipv6);
+            env.add_torrent_peer(&info_hash, &peer_using_ipv6).await;
 
             // Announce the new Peer.
             let response = Client::new(*env.bind_address())
@@ -625,7 +625,7 @@ mod for_all_config_modes {
             let previously_announced_peer = PeerBuilder::default().with_peer_id(&PeerId(*b"-qB00000000000000001")).build();
 
             // Add the Peer 1
-            env.add_torrent_peer(&info_hash, &previously_announced_peer);
+            env.add_torrent_peer(&info_hash, &previously_announced_peer).await;
 
             // Announce the new Peer 2 accepting compact responses
             let response = Client::new(*env.bind_address())
@@ -666,7 +666,7 @@ mod for_all_config_modes {
             let previously_announced_peer = PeerBuilder::default().with_peer_id(&PeerId(*b"-qB00000000000000001")).build();
 
             // Add the Peer 1
-            env.add_torrent_peer(&info_hash, &previously_announced_peer);
+            env.add_torrent_peer(&info_hash, &previously_announced_peer).await;
 
             // Announce the new Peer 2 without passing the "compact" param
             // By default it should respond with the compact peer list
@@ -787,7 +787,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(peer_addr.ip(), client_ip);
@@ -829,7 +830,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(
@@ -878,7 +880,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(
@@ -925,7 +928,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(peer_addr.ip(), IpAddr::from_str("150.172.238.178").unwrap());
@@ -1010,7 +1014,8 @@ mod for_all_config_modes {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             let response = Client::new(*env.bind_address())
                 .scrape(
@@ -1050,7 +1055,8 @@ mod for_all_config_modes {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_no_bytes_pending_to_download()
                     .build(),
-            );
+            )
+            .await;
 
             let response = Client::new(*env.bind_address())
                 .scrape(
@@ -1282,7 +1288,8 @@ mod configured_as_whitelisted {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             let response = Client::new(*env.bind_address())
                 .scrape(
@@ -1318,7 +1325,8 @@ mod configured_as_whitelisted {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             env.container
                 .tracker_core_container
@@ -1494,7 +1502,8 @@ mod configured_as_private {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             let response = Client::new(*env.bind_address())
                 .scrape(
@@ -1525,7 +1534,8 @@ mod configured_as_private {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             let expiring_key = env
                 .container
@@ -1576,7 +1586,8 @@ mod configured_as_private {
                     .with_peer_id(&PeerId(*b"-qB00000000000000001"))
                     .with_bytes_pending_to_download(1)
                     .build(),
-            );
+            )
+            .await;
 
             let false_key: Key = "YZSl4lMZupRuOpSRC3krIKR5BPB14nrJ".parse().unwrap();
 

packages/axum-http-tracker-server/tests/server/v1/contract.rs

@@ -39,6 +39,7 @@ torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
 torrust-tracker-metrics = { version = "3.0.0-develop", path = "../metrics" }
 torrust-tracker-primitives = { version = "3.0.0-develop", path = "../primitives" }
+torrust-tracker-torrent-repository = { version = "3.0.0-develop", path = "../torrent-repository" }
 torrust-udp-tracker-server = { version = "3.0.0-develop", path = "../udp-tracker-server" }
 tower = { version = "0", features = ["timeout"] }
 tower-http = { version = "0", features = ["compression-full", "cors", "propagate-header", "request-id", "trace"] }

packages/axum-rest-tracker-api-server/Cargo.toml

@@ -12,6 +12,7 @@ use torrust_rest_tracker_api_core::container::TrackerHttpApiCoreContainer;
 use torrust_server_lib::registar::Registar;
 use torrust_tracker_configuration::{logging, Configuration};
 use torrust_tracker_primitives::peer;
+use torrust_tracker_torrent_repository::container::TorrentRepositoryContainer;
 use torrust_udp_tracker_server::container::UdpTrackerServerContainer;
 
 use crate::server::{ApiServer, Launcher, Running, Stopped};
@@ -32,12 +33,12 @@ where
     S: std::fmt::Debug + std::fmt::Display,
 {
     /// Add a torrent to the tracker
-    pub fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
-        let _number_of_downloads_increased = self
-            .container
+    pub async fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) -> bool {
+        self.container
             .tracker_core_container
             .in_memory_torrent_repository
-            .upsert_peer(info_hash, peer, None);
+            .upsert_peer(info_hash, peer, None)
+            .await
     }
 }
 
@@ -172,14 +173,25 @@ impl EnvContainer {
                 .clone(),
         );
 
-        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
+        let torrent_repository_container = Arc::new(TorrentRepositoryContainer::initialize(
+            core_config.tracker_usage_statistics.into(),
+        ));
+
+        let tracker_core_container = Arc::new(TrackerCoreContainer::initialize_from(
+            &core_config,
+            &torrent_repository_container,
+        ));
+
         let http_tracker_core_container =
             HttpTrackerCoreContainer::initialize_from_tracker_core(&tracker_core_container, &http_tracker_config);
+
         let udp_tracker_core_container =
             UdpTrackerCoreContainer::initialize_from_tracker_core(&tracker_core_container, &udp_tracker_config);
+
         let udp_tracker_server_container = UdpTrackerServerContainer::initialize(&core_config);
 
         let tracker_http_api_core_container = TrackerHttpApiCoreContainer::initialize_from(
+            &torrent_repository_container,
             &tracker_core_container,
             &http_tracker_core_container,
             &udp_tracker_core_container,