refactor: [#1551] extract event handler for each udp event

Author: josecelano

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

@@ -0,0 +1,95 @@
+use std::sync::Arc;
+
+use bittorrent_udp_tracker_core::services::banning::BanService;
+use tokio::sync::RwLock;
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::{label_name, metric_name};
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use crate::event::{ConnectionContext, ErrorKind, UdpRequestKind};
+use crate::statistics::repository::Repository;
+use crate::statistics::UDP_TRACKER_SERVER_ERRORS_TOTAL;
+
+pub async fn handle_event(
+    context: ConnectionContext,
+    kind: Option<UdpRequestKind>,
+    error: ErrorKind,
+    stats_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 {
+        let mut ban_service = ban_service.write().await;
+        ban_service.increase_counter(&context.client_socket_addr().ip());
+    }
+
+    // Global fixed metrics
+    match context.client_socket_addr().ip() {
+        std::net::IpAddr::V4(_) => {
+            stats_repository.increase_udp4_errors().await;
+        }
+        std::net::IpAddr::V6(_) => {
+            stats_repository.increase_udp6_errors().await;
+        }
+    }
+
+    // Extendable metrics
+    let mut label_set = LabelSet::from(context);
+    if let Some(kind) = kind {
+        label_set.upsert(label_name!("request_kind"), kind.to_string().into());
+    }
+    match stats_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),
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+    use std::sync::Arc;
+
+    use bittorrent_udp_tracker_core::services::banning::BanService;
+    use torrust_tracker_clock::clock::Time;
+    use torrust_tracker_primitives::service_binding::{Protocol, ServiceBinding};
+
+    use crate::event::{ConnectionContext, Event};
+    use crate::statistics::event::handler::error::ErrorKind;
+    use crate::statistics::event::handler::handle_event;
+    use crate::statistics::repository::Repository;
+    use crate::CurrentClock;
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_errors_counter_when_it_receives_a_udp4_error_event() {
+        let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
+
+        handle_event(
+            Event::UdpError {
+                context: ConnectionContext::new(
+                    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 195)), 8080),
+                    ServiceBinding::new(
+                        Protocol::UDP,
+                        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969),
+                    )
+                    .unwrap(),
+                ),
+                kind: None,
+                error: ErrorKind::RequestParse("Invalid request format".to_string()),
+            },
+            &stats_repository,
+            &ban_service,
+            CurrentClock::now(),
+        )
+        .await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_errors_handled, 1);
+    }
+}

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

@@ -0,0 +1,49 @@
+mod error;
+mod request_aborted;
+mod request_accepted;
+mod request_banned;
+mod request_received;
+mod response_sent;
+
+use std::sync::Arc;
+
+use bittorrent_udp_tracker_core::services::banning::BanService;
+use tokio::sync::RwLock;
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use crate::event::Event;
+use crate::statistics::repository::Repository;
+
+pub async fn handle_event(
+    event: Event,
+    stats_repository: &Repository,
+    ban_service: &Arc<RwLock<BanService>>,
+    now: DurationSinceUnixEpoch,
+) {
+    match event {
+        Event::UdpRequestAborted { context } => {
+            request_aborted::handle_event(context, stats_repository, now).await;
+        }
+        Event::UdpRequestBanned { context } => {
+            request_banned::handle_event(context, stats_repository, now).await;
+        }
+        Event::UdpRequestReceived { context } => {
+            request_received::handle_event(context, stats_repository, now).await;
+        }
+        Event::UdpRequestAccepted { context, kind } => {
+            request_accepted::handle_event(context, kind, stats_repository, now).await;
+        }
+        Event::UdpResponseSent {
+            context,
+            kind,
+            req_processing_time,
+        } => {
+            response_sent::handle_event(context, kind, req_processing_time, stats_repository, now).await;
+        }
+        Event::UdpError { context, kind, error } => {
+            error::handle_event(context, kind, error, stats_repository, ban_service, now).await;
+        }
+    }
+
+    tracing::debug!("stats: {:?}", stats_repository.get_stats().await);
+}

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

@@ -0,0 +1,92 @@
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::metric_name;
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use crate::event::ConnectionContext;
+use crate::statistics::repository::Repository;
+use crate::statistics::UDP_TRACKER_SERVER_REQUESTS_ABORTED_TOTAL;
+
+pub async fn handle_event(context: ConnectionContext, stats_repository: &Repository, now: DurationSinceUnixEpoch) {
+    // Global fixed metrics
+    stats_repository.increase_udp_requests_aborted().await;
+
+    // Extendable metrics
+    match stats_repository
+        .increase_counter(
+            &metric_name!(UDP_TRACKER_SERVER_REQUESTS_ABORTED_TOTAL),
+            &LabelSet::from(context),
+            now,
+        )
+        .await
+    {
+        Ok(()) => {}
+        Err(err) => tracing::error!("Failed to increase the counter: {}", err),
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+    use std::sync::Arc;
+
+    use bittorrent_udp_tracker_core::services::banning::BanService;
+    use torrust_tracker_clock::clock::Time;
+    use torrust_tracker_primitives::service_binding::{Protocol, ServiceBinding};
+
+    use crate::event::{ConnectionContext, Event};
+    use crate::statistics::event::handler::handle_event;
+    use crate::statistics::repository::Repository;
+    use crate::CurrentClock;
+
+    #[tokio::test]
+    async fn should_increase_the_number_of_aborted_requests_when_it_receives_a_udp_request_aborted_event() {
+        let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
+
+        handle_event(
+            Event::UdpRequestAborted {
+                context: ConnectionContext::new(
+                    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 195)), 8080),
+                    ServiceBinding::new(
+                        Protocol::UDP,
+                        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969),
+                    )
+                    .unwrap(),
+                ),
+            },
+            &stats_repository,
+            &ban_service,
+            CurrentClock::now(),
+        )
+        .await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp_requests_aborted, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp_abort_counter_when_it_receives_a_udp_abort_event() {
+        let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
+
+        handle_event(
+            Event::UdpRequestAborted {
+                context: ConnectionContext::new(
+                    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 195)), 8080),
+                    ServiceBinding::new(
+                        Protocol::UDP,
+                        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969),
+                    )
+                    .unwrap(),
+                ),
+            },
+            &stats_repository,
+            &ban_service,
+            CurrentClock::now(),
+        )
+        .await;
+        let stats = stats_repository.get_stats().await;
+        assert_eq!(stats.udp_requests_aborted, 1);
+    }
+}