refactor: [torrust#1598] make recalculate udp avg scrape processing time metric and update atomic

It also fixes a division by zero bug when the metrics is updated before
the counter for number of conenctions has been increased.

It only avoid the division by zero. I will propoerly fixed with
independent request counter for the moving average calculation.

Author: josecelano

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

@@ -4,7 +4,7 @@ use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
 use crate::event::{ConnectionContext, UdpRequestKind, UdpResponseKind};
 use crate::statistics::repository::Repository;
-use crate::statistics::{UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS, UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL};
+use crate::statistics::UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL;
 
 pub async fn handle_event(
     context: ConnectionContext,
@@ -36,33 +36,20 @@ pub async fn handle_event(
                 (LabelValue::new("ok"), UdpRequestKind::Announce { announce_request }.into())
             }
             UdpRequestKind::Scrape => {
-                let new_avg = stats_repository
-                    .recalculate_udp_avg_scrape_processing_time_ns(req_processing_time)
-                    .await;
-
-                tracing::debug!("Updating average processing time metric for scrape requests: {} ns", new_avg);
-
                 let mut label_set = LabelSet::from(context.clone());
                 label_set.upsert(label_name!("request_kind"), LabelValue::new(&req_kind.to_string()));
-                match stats_repository
-                    .set_gauge(
-                        &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS),
-                        &label_set,
-                        new_avg,
-                        now,
-                    )
-                    .await
-                {
-                    Ok(()) => {}
-                    Err(err) => tracing::error!("Failed to set gauge: {}", err),
-                }
+
+                let _new_avg = stats_repository
+                    .recalculate_udp_avg_scrape_processing_time_ns(req_processing_time, &label_set, now)
+                    .await;
+
                 (LabelValue::new("ok"), LabelValue::new(&UdpRequestKind::Scrape.to_string()))
             }
         },
         UdpResponseKind::Error { opt_req_kind: _ } => (LabelValue::new("error"), LabelValue::ignore()),
     };
 
-    // Extendable metrics
+    // Increase the number of responses sent
     let mut label_set = LabelSet::from(context);
     if result_label_value == LabelValue::new("ok") {
         label_set.upsert(label_name!("request_kind"), kind_label_value);

packages/udp-tracker-server/src/statistics/metrics.rs

@@ -115,34 +115,25 @@ impl Metrics {
         new_avg
     }
 
-    fn update_udp_avg_processing_time_ns(&mut self, new_avg: f64, label_set: &LabelSet, now: DurationSinceUnixEpoch) {
-        tracing::debug!(
-            "Updating average processing time metric to {} ns for label set {}",
-            new_avg,
-            label_set,
-        );
-
-        match self.set_gauge(
-            &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS),
-            label_set,
-            new_avg,
-            now,
-        ) {
-            Ok(()) => {}
-            Err(err) => tracing::error!("Failed to set gauge: {}", err),
-        }
-    }
-
     #[allow(clippy::cast_precision_loss)]
-    pub fn recalculate_udp_avg_scrape_processing_time_ns(&self, req_processing_time: Duration) -> f64 {
+    pub fn recalculate_udp_avg_scrape_processing_time_ns(
+        &mut self,
+        req_processing_time: Duration,
+        label_set: &LabelSet,
+        now: DurationSinceUnixEpoch,
+    ) -> f64 {
         let req_processing_time = req_processing_time.as_nanos() as f64;
 
         let udp_scrapes_handled = (self.udp4_scrapes_handled() + self.udp6_scrapes_handled()) as f64;
 
         let previous_avg = self.udp_avg_scrape_processing_time_ns();
 
-        // Moving average: https://en.wikipedia.org/wiki/Moving_average
-        let new_avg = previous_avg as f64 + (req_processing_time - previous_avg as f64) / udp_scrapes_handled;
+        let new_avg = if udp_scrapes_handled == 0.0 {
+            req_processing_time
+        } else {
+            // Moving average: https://en.wikipedia.org/wiki/Moving_average
+            previous_avg as f64 + (req_processing_time - previous_avg as f64) / udp_scrapes_handled
+        };
 
         tracing::debug!(
             "Recalculated UDP average scrape processing time: {} ns (previous: {} ns, req_processing_time: {} ns, udp_scrapes_handled: {})",
@@ -152,9 +143,29 @@ impl Metrics {
             udp_scrapes_handled
         );
 
+        self.update_udp_avg_processing_time_ns(new_avg, label_set, now);
+
         new_avg
     }
 
+    fn update_udp_avg_processing_time_ns(&mut self, new_avg: f64, label_set: &LabelSet, now: DurationSinceUnixEpoch) {
+        tracing::debug!(
+            "Updating average processing time metric to {} ns for label set {}",
+            new_avg,
+            label_set,
+        );
+
+        match self.set_gauge(
+            &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS),
+            label_set,
+            new_avg,
+            now,
+        ) {
+            Ok(()) => {}
+            Err(err) => tracing::error!("Failed to set gauge: {}", err),
+        }
+    }
+
     // UDP
     /// Total number of UDP (UDP tracker) requests aborted.
     #[must_use]

packages/udp-tracker-server/src/statistics/repository.rs

@@ -103,10 +103,15 @@ impl Repository {
         new_avg
     }
 
-    pub async fn recalculate_udp_avg_scrape_processing_time_ns(&self, req_processing_time: Duration) -> f64 {
-        let stats_lock = self.stats.write().await;
+    pub async fn recalculate_udp_avg_scrape_processing_time_ns(
+        &self,
+        req_processing_time: Duration,
+        label_set: &LabelSet,
+        now: DurationSinceUnixEpoch,
+    ) -> f64 {
+        let mut stats_lock = self.stats.write().await;
 
-        let new_avg = stats_lock.recalculate_udp_avg_scrape_processing_time_ns(req_processing_time);
+        let new_avg = stats_lock.recalculate_udp_avg_scrape_processing_time_ns(req_processing_time, label_set, now);
 
         drop(stats_lock);
 
@@ -436,7 +441,9 @@ mod tests {
 
         // Calculate new average with processing time of 1200ns
         let processing_time = Duration::from_nanos(1200);
-        let new_avg = repo.recalculate_udp_avg_scrape_processing_time_ns(processing_time).await;
+        let new_avg = repo
+            .recalculate_udp_avg_scrape_processing_time_ns(processing_time, &scrape_labels, now)
+            .await;
 
         // Moving average: previous_avg + (new_value - previous_avg) / total_scrapes
         // 800 + (1200 - 800) / 4 = 800 + 100 = 900
@@ -465,14 +472,16 @@ mod tests {
             .recalculate_udp_avg_announce_processing_time_ns(processing_time, &announce_labels, now)
             .await;
 
-        let _scrape_labels = LabelSet::from([("request_kind", "scrape")]);
-        let scrape_avg = repo.recalculate_udp_avg_scrape_processing_time_ns(processing_time).await;
+        let scrape_labels = LabelSet::from([("request_kind", "scrape")]);
+        let scrape_avg = repo
+            .recalculate_udp_avg_scrape_processing_time_ns(processing_time, &scrape_labels, now)
+            .await;
 
         // With 0 total connections, the formula becomes 0 + (1000 - 0) / 0
         // This should handle the division by zero case gracefully
         assert!((connect_avg - 1000.0).abs() < f64::EPSILON);
         assert!((announce_avg - 1000.0).abs() < f64::EPSILON);
-        assert!(scrape_avg.is_infinite() || scrape_avg.is_nan());
+        assert!((scrape_avg - 1000.0).abs() < f64::EPSILON);
     }
 
     #[tokio::test]