Merge #1381: Overhaul stats events: merge UDP core events with a different IP version

8603f8b refactor: [#1380] refactor: [#1373] merge UDP stats events with different IP version (Jose Celano)
2be682e refactor: [#1380] refactor: [#1371] add connection context to UDP core events (Jose Celano)

Pull request description:

  Change events from this:

  `bittorrent_udp_tracker_core::statistics::event::Event`:

  ```rust
  pub enum Event {
      Udp4Connect,
      Udp4Announce,
      Udp4Scrape,
      Udp6Connect,
      Udp6Announce,
      Udp6Scrape,
  }
  ```

  To this:

  ```rust
  pub enum Event {
      UdpConnect { context: ConnectionContext },
      UdpAnnounce { context: ConnectionContext },
      UdpScrape { context: ConnectionContext },
  }

  pub struct ConnectionContext {
      client_socket_addr: SocketAddr,
      server_socket_addr: SocketAddr,
  }
  ```

  This have also be done for [HTTP core events](#1374).

  ### Sub-tasks

  - [x] Add `ConnectionContext` to events.
  - [x] Merge events with the same request type (`connect`, `announce` and `scrape`).

ACKs for top commit:
  josecelano:
    ACK 8603f8b

Tree-SHA512: dde61535ec1f6c768ae2ad4d911e4ecece9be94c2546fd7b06a0d8d45d122caa6fcbf370cfa5de61a0d10424927cef17f7faf0a351252ab28ccb17e38058269e

Author: josecelano

packages/udp-tracker-core/benches/helpers/sync.rs

@@ -1,3 +1,4 @@
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -8,13 +9,16 @@ use crate::helpers::utils::{sample_ipv4_remote_addr, sample_issue_time};
 
 #[allow(clippy::unused_async)]
 pub async fn connect_once(samples: u64) -> Duration {
+    let client_socket_addr = sample_ipv4_remote_addr();
+    let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
     let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false);
     let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender);
     let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender));
     let start = Instant::now();
 
     for _ in 0..samples {
-        let _response = connect_service.handle_connect(sample_ipv4_remote_addr(), sample_issue_time());
+        let _response = connect_service.handle_connect(client_socket_addr, server_socket_addr, sample_issue_time());
     }
 
     start.elapsed()

packages/udp-tracker-core/src/services/announce.rs

@@ -7,7 +7,7 @@
 //!
 //! It also sends an [`udp_tracker_core::statistics::event::Event`]
 //! because events are specific for the HTTP tracker.
-use std::net::{IpAddr, SocketAddr};
+use std::net::SocketAddr;
 use std::ops::Range;
 use std::sync::Arc;
 
@@ -21,6 +21,7 @@ use torrust_tracker_primitives::core::AnnounceData;
 
 use crate::connection_cookie::{check, gen_remote_fingerprint, ConnectionCookieError};
 use crate::statistics;
+use crate::statistics::event::ConnectionContext;
 
 /// The `AnnounceService` is responsible for handling the `announce` requests.
 ///
@@ -57,17 +58,18 @@ impl AnnounceService {
     ///   whitelist.
     pub async fn handle_announce(
         &self,
-        remote_addr: SocketAddr,
+        client_socket_addr: SocketAddr,
+        server_socket_addr: SocketAddr,
         request: &AnnounceRequest,
         cookie_valid_range: Range<f64>,
     ) -> Result<AnnounceData, UdpAnnounceError> {
-        Self::authenticate(remote_addr, request, cookie_valid_range)?;
+        Self::authenticate(client_socket_addr, request, cookie_valid_range)?;
 
         let info_hash = request.info_hash.into();
 
         self.authorize(&info_hash).await?;
 
-        let remote_client_ip = remote_addr.ip();
+        let remote_client_ip = client_socket_addr.ip();
 
         let mut peer = peer_builder::from_request(request, &remote_client_ip);
 
@@ -78,7 +80,7 @@ impl AnnounceService {
             .announce(&info_hash, &mut peer, &remote_client_ip, &peers_wanted)
             .await?;
 
-        self.send_stats_event(remote_client_ip).await;
+        self.send_stats_event(client_socket_addr, server_socket_addr).await;
 
         Ok(announce_data)
     }
@@ -99,14 +101,13 @@ impl AnnounceService {
         self.whitelist_authorization.authorize(info_hash).await
     }
 
-    async fn send_stats_event(&self, peer_ip: IpAddr) {
+    async fn send_stats_event(&self, client_socket_addr: SocketAddr, server_socket_addr: SocketAddr) {
         if let Some(udp_stats_event_sender) = self.opt_udp_core_stats_event_sender.as_deref() {
-            let event = match peer_ip {
-                IpAddr::V4(_) => statistics::event::Event::Udp4Announce,
-                IpAddr::V6(_) => statistics::event::Event::Udp6Announce,
-            };
-
-            udp_stats_event_sender.send_event(event).await;
+            udp_stats_event_sender
+                .send_event(statistics::event::Event::UdpAnnounce {
+                    context: ConnectionContext::new(client_socket_addr, server_socket_addr),
+                })
+                .await;
         }
     }
 }

packages/udp-tracker-core/src/services/connect.rs

@@ -8,6 +8,7 @@ use aquatic_udp_protocol::ConnectionId;
 
 use crate::connection_cookie::{gen_remote_fingerprint, make};
 use crate::statistics;
+use crate::statistics::event::ConnectionContext;
 
 /// The `ConnectService` is responsible for handling the `connect` requests.
 ///
@@ -30,18 +31,21 @@ impl ConnectService {
     /// # Panics
     ///
     /// It will panic if there was an error making the connection cookie.
-    pub async fn handle_connect(&self, remote_addr: SocketAddr, cookie_issue_time: f64) -> ConnectionId {
-        let connection_id = make(gen_remote_fingerprint(&remote_addr), cookie_issue_time).expect("it should be a normal value");
+    pub async fn handle_connect(
+        &self,
+        client_socket_addr: SocketAddr,
+        server_socket_addr: SocketAddr,
+        cookie_issue_time: f64,
+    ) -> ConnectionId {
+        let connection_id =
+            make(gen_remote_fingerprint(&client_socket_addr), cookie_issue_time).expect("it should be a normal value");
 
         if let Some(udp_stats_event_sender) = self.opt_udp_core_stats_event_sender.as_deref() {
-            match remote_addr {
-                SocketAddr::V4(_) => {
-                    udp_stats_event_sender.send_event(statistics::event::Event::Udp4Connect).await;
-                }
-                SocketAddr::V6(_) => {
-                    udp_stats_event_sender.send_event(statistics::event::Event::Udp6Connect).await;
-                }
-            }
+            udp_stats_event_sender
+                .send_event(statistics::event::Event::UdpConnect {
+                    context: ConnectionContext::new(client_socket_addr, server_socket_addr),
+                })
+                .await;
         }
 
         connection_id
@@ -54,6 +58,7 @@ mod tests {
     mod connect_request {
 
         use std::future;
+        use std::net::{IpAddr, Ipv4Addr, SocketAddr};
         use std::sync::Arc;
 
         use mockall::predicate::eq;
@@ -65,16 +70,19 @@ mod tests {
             sample_ipv6_remote_addr_fingerprint, sample_issue_time, MockUdpCoreStatsEventSender,
         };
         use crate::statistics;
+        use crate::statistics::event::ConnectionContext;
 
         #[tokio::test]
         async fn a_connect_response_should_contain_the_same_transaction_id_as_the_connect_request() {
+            let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
             let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false);
             let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender);
 
             let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender));
 
             let response = connect_service
-                .handle_connect(sample_ipv4_remote_addr(), sample_issue_time())
+                .handle_connect(sample_ipv4_remote_addr(), server_socket_addr, sample_issue_time())
                 .await;
 
             assert_eq!(
@@ -85,13 +93,15 @@ mod tests {
 
         #[tokio::test]
         async fn a_connect_response_should_contain_a_new_connection_id() {
+            let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
             let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false);
             let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender);
 
             let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender));
 
             let response = connect_service
-                .handle_connect(sample_ipv4_remote_addr(), sample_issue_time())
+                .handle_connect(sample_ipv4_remote_addr(), server_socket_addr, sample_issue_time())
                 .await;
 
             assert_eq!(
@@ -102,13 +112,16 @@ mod tests {
 
         #[tokio::test]
         async fn a_connect_response_should_contain_a_new_connection_id_ipv6() {
+            let client_socket_addr = sample_ipv6_remote_addr();
+            let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
             let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false);
             let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender);
 
             let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender));
 
             let response = connect_service
-                .handle_connect(sample_ipv6_remote_addr(), sample_issue_time())
+                .handle_connect(client_socket_addr, server_socket_addr, sample_issue_time())
                 .await;
 
             assert_eq!(
@@ -119,30 +132,38 @@ mod tests {
 
         #[tokio::test]
         async fn it_should_send_the_upd4_connect_event_when_a_client_tries_to_connect_using_a_ip4_socket_address() {
+            let client_socket_addr = sample_ipv4_socket_address();
+            let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
             let mut udp_stats_event_sender_mock = MockUdpCoreStatsEventSender::new();
             udp_stats_event_sender_mock
                 .expect_send_event()
-                .with(eq(statistics::event::Event::Udp4Connect))
+                .with(eq(statistics::event::Event::UdpConnect {
+                    context: ConnectionContext::new(client_socket_addr, server_socket_addr),
+                }))
                 .times(1)
                 .returning(|_| Box::pin(future::ready(Some(Ok(())))));
             let opt_udp_stats_event_sender: Arc<Option<Box<dyn statistics::event::sender::Sender>>> =
                 Arc::new(Some(Box::new(udp_stats_event_sender_mock)));
 
-            let client_socket_address = sample_ipv4_socket_address();
-
             let connect_service = Arc::new(ConnectService::new(opt_udp_stats_event_sender));
 
             connect_service
-                .handle_connect(client_socket_address, sample_issue_time())
+                .handle_connect(client_socket_addr, server_socket_addr, sample_issue_time())
                 .await;
         }
 
         #[tokio::test]
         async fn it_should_send_the_upd6_connect_event_when_a_client_tries_to_connect_using_a_ip6_socket_address() {
+            let client_socket_addr = sample_ipv6_remote_addr();
+            let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
+
             let mut udp_stats_event_sender_mock = MockUdpCoreStatsEventSender::new();
             udp_stats_event_sender_mock
                 .expect_send_event()
-                .with(eq(statistics::event::Event::Udp6Connect))
+                .with(eq(statistics::event::Event::UdpConnect {
+                    context: ConnectionContext::new(client_socket_addr, server_socket_addr),
+                }))
                 .times(1)
                 .returning(|_| Box::pin(future::ready(Some(Ok(())))));
             let opt_udp_stats_event_sender: Arc<Option<Box<dyn statistics::event::sender::Sender>>> =
@@ -151,7 +172,7 @@ mod tests {
             let connect_service = Arc::new(ConnectService::new(opt_udp_stats_event_sender));
 
             connect_service
-                .handle_connect(sample_ipv6_remote_addr(), sample_issue_time())
+                .handle_connect(client_socket_addr, server_socket_addr, sample_issue_time())
                 .await;
         }
     }

packages/udp-tracker-core/src/services/scrape.rs

@@ -19,6 +19,7 @@ use torrust_tracker_primitives::core::ScrapeData;
 
 use crate::connection_cookie::{check, gen_remote_fingerprint, ConnectionCookieError};
 use crate::statistics;
+use crate::statistics::event::ConnectionContext;
 
 /// The `ScrapeService` is responsible for handling the `scrape` requests.
 ///
@@ -49,18 +50,19 @@ impl ScrapeService {
     /// It will return an error if the tracker core scrape handler returns an error.
     pub async fn handle_scrape(
         &self,
-        remote_client_addr: SocketAddr,
+        client_socket_addr: SocketAddr,
+        server_socket_addr: SocketAddr,
         request: &ScrapeRequest,
         cookie_valid_range: Range<f64>,
     ) -> Result<ScrapeData, UdpScrapeError> {
-        Self::authenticate(remote_client_addr, request, cookie_valid_range)?;
+        Self::authenticate(client_socket_addr, request, cookie_valid_range)?;
 
         let scrape_data = self
             .scrape_handler
             .scrape(&Self::convert_from_aquatic(&request.info_hashes))
             .await?;
 
-        self.send_stats_event(remote_client_addr).await;
+        self.send_stats_event(client_socket_addr, server_socket_addr).await;
 
         Ok(scrape_data)
     }
@@ -81,13 +83,13 @@ impl ScrapeService {
         aquatic_infohashes.iter().map(|&x| x.into()).collect()
     }
 
-    async fn send_stats_event(&self, remote_addr: SocketAddr) {
+    async fn send_stats_event(&self, client_socket_addr: SocketAddr, server_socket_addr: SocketAddr) {
         if let Some(udp_stats_event_sender) = self.opt_udp_stats_event_sender.as_deref() {
-            let event = match remote_addr {
-                SocketAddr::V4(_) => statistics::event::Event::Udp4Scrape,
-                SocketAddr::V6(_) => statistics::event::Event::Udp6Scrape,
-            };
-            udp_stats_event_sender.send_event(event).await;
+            udp_stats_event_sender
+                .send_event(statistics::event::Event::UdpScrape {
+                    context: ConnectionContext::new(client_socket_addr, server_socket_addr),
+                })
+                .await;
         }
     }
 }