Merge #1557: Add a new labeled metric to track requests with wrong connection ID per client's software.

7e616d7 feat: [#1556] add a new metric to count connection ID errors per clietn software (Jose Celano)
d9f4c13 refactor: [#1556] extract functions (Jose Celano)

Pull request description:

  Add a new labeled metric to track requests with wrong connection ID per client's software.

ACKs for top commit:
  josecelano:
    ACK 7e616d7

Tree-SHA512: ed8c2be5845a179692b18b0cc84cbaccf1b7f8178f1754835ee472041eab8c35d366fa9e8193ae4020aba35c1272f0551003febf80e37c8d15597d72f4ca8ca4

Author: josecelano

cSpell.json

@@ -127,6 +127,7 @@
         "proto",
         "Quickstart",
         "Radeon",
+        "Rakshasa",
         "Rasterbar",
         "realpath",
         "reannounce",

packages/udp-tracker-server/src/event.rs

@@ -2,6 +2,7 @@ use std::fmt;
 use std::net::SocketAddr;
 use std::time::Duration;
 
+use aquatic_udp_protocol::AnnounceRequest;
 use bittorrent_tracker_core::error::{AnnounceError, ScrapeError};
 use bittorrent_udp_tracker_core::services::announce::UdpAnnounceError;
 use bittorrent_udp_tracker_core::services::scrape::UdpScrapeError;
@@ -42,15 +43,25 @@ pub enum Event {
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum UdpRequestKind {
     Connect,
-    Announce,
+    Announce { announce_request: AnnounceRequest },
     Scrape,
 }
 
+impl From<UdpRequestKind> for LabelValue {
+    fn from(kind: UdpRequestKind) -> Self {
+        match kind {
+            UdpRequestKind::Connect => LabelValue::new("connect"),
+            UdpRequestKind::Announce { .. } => LabelValue::new("announce"),
+            UdpRequestKind::Scrape => LabelValue::new("scrape"),
+        }
+    }
+}
+
 impl fmt::Display for UdpRequestKind {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let proto_str = match self {
             UdpRequestKind::Connect => "connect",
-            UdpRequestKind::Announce => "announce",
+            UdpRequestKind::Announce { .. } => "announce",
             UdpRequestKind::Scrape => "scrape",
         };
         write!(f, "{proto_str}")

packages/udp-tracker-server/src/handlers/announce.rs

@@ -44,15 +44,25 @@ pub async fn handle_announce(
         udp_server_stats_event_sender
             .send(Event::UdpRequestAccepted {
                 context: ConnectionContext::new(client_socket_addr, server_service_binding.clone()),
-                kind: UdpRequestKind::Announce,
+                kind: UdpRequestKind::Announce {
+                    announce_request: *request,
+                },
             })
             .await;
     }
 
     let announce_data = announce_service
         .handle_announce(client_socket_addr, server_service_binding, request, cookie_valid_range)
         .await
-        .map_err(|e| (e.into(), request.transaction_id, UdpRequestKind::Announce))?;
+        .map_err(|e| {
+            (
+                e.into(),
+                request.transaction_id,
+                UdpRequestKind::Announce {
+                    announce_request: *request,
+                },
+            )
+        })?;
 
     Ok(build_response(client_socket_addr, request, core_config, &announce_data))
 }
@@ -118,9 +128,9 @@ fn build_response(
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
 
-    mod announce_request {
+    pub mod announce_request {
 
         use std::net::Ipv4Addr;
         use std::num::NonZeroU16;
@@ -133,7 +143,7 @@ mod tests {
 
         use crate::handlers::tests::{sample_ipv4_remote_addr_fingerprint, sample_issue_time};
 
-        struct AnnounceRequestBuilder {
+        pub struct AnnounceRequestBuilder {
             request: AnnounceRequest,
         }
 
@@ -431,13 +441,14 @@ mod tests {
                 let client_socket_addr = sample_ipv4_socket_address();
                 let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969);
                 let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap();
+                let announce_request = AnnounceRequestBuilder::default().into();
 
                 let mut udp_server_stats_event_sender_mock = MockUdpServerStatsEventSender::new();
                 udp_server_stats_event_sender_mock
                     .expect_send()
                     .with(eq(Event::UdpRequestAccepted {
                         context: ConnectionContext::new(client_socket_addr, server_service_binding.clone()),
-                        kind: UdpRequestKind::Announce,
+                        kind: UdpRequestKind::Announce { announce_request },
                     }))
                     .times(1)
                     .returning(|_| Box::pin(future::ready(Some(Ok(1)))));
@@ -451,7 +462,7 @@ mod tests {
                     &core_udp_tracker_services.announce_service,
                     client_socket_addr,
                     server_service_binding,
-                    &AnnounceRequestBuilder::default().into(),
+                    &announce_request,
                     &core_tracker_services.core_config,
                     &udp_server_stats_event_sender,
                     sample_cookie_valid_range(),
@@ -795,12 +806,16 @@ mod tests {
                 let server_socket_addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 203, 0, 113, 196)), 6969);
                 let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap();
 
+                let announce_request = AnnounceRequestBuilder::default()
+                    .with_connection_id(make(gen_remote_fingerprint(&client_socket_addr), sample_issue_time()).unwrap())
+                    .into();
+
                 let mut udp_server_stats_event_sender_mock = MockUdpServerStatsEventSender::new();
                 udp_server_stats_event_sender_mock
                     .expect_send()
                     .with(eq(Event::UdpRequestAccepted {
                         context: ConnectionContext::new(client_socket_addr, server_service_binding.clone()),
-                        kind: UdpRequestKind::Announce,
+                        kind: UdpRequestKind::Announce { announce_request },
                     }))
                     .times(1)
                     .returning(|_| Box::pin(future::ready(Some(Ok(1)))));
@@ -810,10 +825,6 @@ mod tests {
                 let (core_tracker_services, core_udp_tracker_services, _server_udp_tracker_services) =
                     initialize_core_tracker_services_for_default_tracker_configuration();
 
-                let announce_request = AnnounceRequestBuilder::default()
-                    .with_connection_id(make(gen_remote_fingerprint(&client_socket_addr), sample_issue_time()).unwrap())
-                    .into();
-
                 handle_announce(
                     &core_udp_tracker_services.announce_service,
                     client_socket_addr,
@@ -887,6 +898,14 @@ mod tests {
                     let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
                     let db_downloads_metric_repository = Arc::new(DatabaseDownloadsMetricRepository::new(&database));
 
+                    let request = AnnounceRequestBuilder::default()
+                        .with_connection_id(make(gen_remote_fingerprint(&client_socket_addr), sample_issue_time()).unwrap())
+                        .with_info_hash(info_hash)
+                        .with_peer_id(peer_id)
+                        .with_ip_address(client_ip_v4)
+                        .with_port(client_port)
+                        .into();
+
                     let mut udp_core_stats_event_sender_mock = MockUdpCoreStatsEventSender::new();
                     udp_core_stats_event_sender_mock
                         .expect_send()
@@ -912,7 +931,9 @@ mod tests {
                         .expect_send()
                         .with(eq(Event::UdpRequestAccepted {
                             context: ConnectionContext::new(client_socket_addr, server_service_binding_clone.clone()),
-                            kind: UdpRequestKind::Announce,
+                            kind: UdpRequestKind::Announce {
+                                announce_request: request,
+                            },
                         }))
                         .times(1)
                         .returning(|_| Box::pin(future::ready(Some(Ok(1)))));
@@ -926,14 +947,6 @@ mod tests {
                         &db_downloads_metric_repository,
                     ));
 
-                    let request = AnnounceRequestBuilder::default()
-                        .with_connection_id(make(gen_remote_fingerprint(&client_socket_addr), sample_issue_time()).unwrap())
-                        .with_info_hash(info_hash)
-                        .with_peer_id(peer_id)
-                        .with_ip_address(client_ip_v4)
-                        .with_port(client_port)
-                        .into();
-
                     let core_config = Arc::new(config.core.clone());
 
                     let announce_service = Arc::new(AnnounceService::new(

packages/udp-tracker-server/src/handlers/mod.rs

@@ -177,7 +177,7 @@ pub async fn handle_request(
             )
             .await
             {
-                Ok(response) => Ok((response, UdpRequestKind::Announce)),
+                Ok(response) => Ok((response, UdpRequestKind::Announce { announce_request })),
                 Err(err) => Err(err),
             }
         }

packages/udp-tracker-server/src/server/processor.rs

@@ -87,16 +87,15 @@ impl Processor {
         };
 
         let udp_response_kind = match &response {
-            Response::Connect(_) => event::UdpResponseKind::Ok {
-                req_kind: event::UdpRequestKind::Connect,
-            },
-            Response::AnnounceIpv4(_) | Response::AnnounceIpv6(_) => event::UdpResponseKind::Ok {
-                req_kind: event::UdpRequestKind::Announce,
-            },
-            Response::Scrape(_) => event::UdpResponseKind::Ok {
-                req_kind: event::UdpRequestKind::Scrape,
-            },
             Response::Error(_e) => event::UdpResponseKind::Error { opt_req_kind: None },
+            _ => {
+                if let Some(req_kind) = opt_req_kind {
+                    event::UdpResponseKind::Ok { req_kind }
+                } else {
+                    // code-review: this case should never happen.
+                    event::UdpResponseKind::Error { opt_req_kind }
+                }
+            }
         };
 
         let mut writer = Cursor::new(Vec::with_capacity(200));

packages/udp-tracker-server/src/statistics/event/handler/error.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use aquatic_udp_protocol::PeerClient;
 use bittorrent_udp_tracker_core::services::banning::BanService;
 use tokio::sync::RwLock;
 use torrust_tracker_metrics::label::LabelSet;
@@ -8,45 +9,118 @@ use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
 use crate::event::{ConnectionContext, ErrorKind, UdpRequestKind};
 use crate::statistics::repository::Repository;
-use crate::statistics::UDP_TRACKER_SERVER_ERRORS_TOTAL;
+use crate::statistics::{UDP_TRACKER_SERVER_CONNECTION_ID_ERRORS_TOTAL, UDP_TRACKER_SERVER_ERRORS_TOTAL};
 
 pub async fn handle_event(
-    context: ConnectionContext,
-    kind: Option<UdpRequestKind>,
-    error: ErrorKind,
-    stats_repository: &Repository,
+    connection_context: ConnectionContext,
+    opt_udp_request_kind: Option<UdpRequestKind>,
+    error_kind: ErrorKind,
+    repository: &Repository,
     ban_service: &Arc<RwLock<BanService>>,
     now: DurationSinceUnixEpoch,
 ) {
-    // Increase the number of errors
-    // code-review: should we ban IP due to other errors too?
-    if let ErrorKind::ConnectionCookie(_msg) = error {
+    if let ErrorKind::ConnectionCookie(_msg) = error_kind.clone() {
         let mut ban_service = ban_service.write().await;
-        ban_service.increase_counter(&context.client_socket_addr().ip());
+        ban_service.increase_counter(&connection_context.client_socket_addr().ip());
     }
 
-    // Global fixed metrics
-    match context.client_socket_addr().ip() {
+    update_global_fixed_metrics(&connection_context, repository).await;
+
+    update_extendable_metrics(&connection_context, opt_udp_request_kind, error_kind, repository, now).await;
+}
+
+async fn update_global_fixed_metrics(connection_context: &ConnectionContext, repository: &Repository) {
+    match connection_context.client_socket_addr().ip() {
         std::net::IpAddr::V4(_) => {
-            stats_repository.increase_udp4_errors().await;
+            repository.increase_udp4_errors().await;
         }
         std::net::IpAddr::V6(_) => {
-            stats_repository.increase_udp6_errors().await;
+            repository.increase_udp6_errors().await;
         }
     }
+}
 
-    // Extendable metrics
-    let mut label_set = LabelSet::from(context);
-    if let Some(kind) = kind {
+async fn update_extendable_metrics(
+    connection_context: &ConnectionContext,
+    opt_udp_request_kind: Option<UdpRequestKind>,
+    error_kind: ErrorKind,
+    repository: &Repository,
+    now: DurationSinceUnixEpoch,
+) {
+    update_all_errors_counter(connection_context, opt_udp_request_kind.clone(), repository, now).await;
+    update_connection_id_errors_counter(opt_udp_request_kind, error_kind, repository, now).await;
+}
+
+async fn update_all_errors_counter(
+    connection_context: &ConnectionContext,
+    opt_udp_request_kind: Option<UdpRequestKind>,
+    repository: &Repository,
+    now: DurationSinceUnixEpoch,
+) {
+    let mut label_set = LabelSet::from(connection_context.clone());
+
+    if let Some(kind) = opt_udp_request_kind.clone() {
         label_set.upsert(label_name!("request_kind"), kind.to_string().into());
     }
-    match stats_repository
+
+    match repository
         .increase_counter(&metric_name!(UDP_TRACKER_SERVER_ERRORS_TOTAL), &label_set, now)
         .await
     {
         Ok(()) => {}
         Err(err) => tracing::error!("Failed to increase the counter: {}", err),
-    };
+    }
+}
+
+async fn update_connection_id_errors_counter(
+    opt_udp_request_kind: Option<UdpRequestKind>,
+    error_kind: ErrorKind,
+    repository: &Repository,
+    now: DurationSinceUnixEpoch,
+) {
+    if let ErrorKind::ConnectionCookie(_) = error_kind {
+        if let Some(UdpRequestKind::Announce { announce_request }) = opt_udp_request_kind {
+            let (client_software_name, client_software_version) = extract_name_and_version(&announce_request.peer_id.client());
+
+            let label_set = LabelSet::from([
+                (label_name!("client_software_name"), client_software_name.into()),
+                (label_name!("client_software_version"), client_software_version.into()),
+            ]);
+
+            match repository
+                .increase_counter(&metric_name!(UDP_TRACKER_SERVER_CONNECTION_ID_ERRORS_TOTAL), &label_set, now)
+                .await
+            {
+                Ok(()) => {}
+                Err(err) => tracing::error!("Failed to increase the counter: {}", err),
+            };
+        }
+    }
+}
+
+fn extract_name_and_version(peer_client: &PeerClient) -> (String, String) {
+    match peer_client {
+        PeerClient::BitTorrent(compact_string) => ("BitTorrent".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::Deluge(compact_string) => ("Deluge".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::LibTorrentRakshasa(compact_string) => ("lt (rakshasa)".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::LibTorrentRasterbar(compact_string) => ("lt (rasterbar)".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::QBitTorrent(compact_string) => ("QBitTorrent".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::Transmission(compact_string) => ("Transmission".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::UTorrent(compact_string) => ("µTorrent".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::UTorrentEmbedded(compact_string) => ("µTorrent Emb.".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::UTorrentMac(compact_string) => ("µTorrent Mac".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::UTorrentWeb(compact_string) => ("µTorrent Web".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::Vuze(compact_string) => ("Vuze".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::WebTorrent(compact_string) => ("WebTorrent".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::WebTorrentDesktop(compact_string) => ("WebTorrent Desktop".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::Mainline(compact_string) => ("Mainline".to_string(), compact_string.as_str().to_owned()),
+        PeerClient::OtherWithPrefixAndVersion { prefix, version } => {
+            (format!("Other ({})", prefix.as_str()), version.as_str().to_owned())
+        }
+        PeerClient::OtherWithPrefix(compact_string) => (format!("Other ({compact_string})"), String::new()),
+        PeerClient::Other => ("Other".to_string(), String::new()),
+        _ => ("Unknown".to_string(), String::new()),
+    }
 }
 
 #[cfg(test)]