refactor: [torrust#1598] make recalculate udp avg announce 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

@@ -26,29 +26,13 @@ pub async fn handle_event(
                 (LabelValue::new("ok"), UdpRequestKind::Connect.into())
             }
             UdpRequestKind::Announce { announce_request } => {
-                let new_avg = stats_repository
-                    .recalculate_udp_avg_announce_processing_time_ns(req_processing_time)
-                    .await;
-
-                tracing::debug!(
-                    "Updating average processing time metric for announce 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_announce_processing_time_ns(req_processing_time, &label_set, now)
+                    .await;
+
                 (LabelValue::new("ok"), UdpRequestKind::Announce { announce_request }.into())
             }
             UdpRequestKind::Scrape => {

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

@@ -77,35 +77,30 @@ impl Metrics {
             udp_connections_handled
         );
 
-        self.update_udp_avg_connect_processing_time_ns(new_avg, label_set, now);
+        self.update_udp_avg_processing_time_ns(new_avg, label_set, now);
 
         new_avg
     }
 
-    fn update_udp_avg_connect_processing_time_ns(&mut self, new_avg: f64, label_set: &LabelSet, now: DurationSinceUnixEpoch) {
-        tracing::debug!("Updating average processing time metric for connect requests: {} ns", new_avg);
-
-        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_announce_processing_time_ns(&self, req_processing_time: Duration) -> f64 {
+    pub fn recalculate_udp_avg_announce_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_announces_handled = (self.udp4_announces_handled() + self.udp6_announces_handled()) as f64;
 
         let previous_avg = self.udp_avg_announce_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_announces_handled;
+        let new_avg = if udp_announces_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_announces_handled
+        };
 
         tracing::debug!(
             "Recalculated UDP average announce processing time: {} ns (previous: {} ns, req_processing_time: {} ns, udp_announces_handled: {})",
@@ -115,9 +110,29 @@ impl Metrics {
             udp_announces_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),
+        }
+    }
+
     #[allow(clippy::cast_precision_loss)]
     pub fn recalculate_udp_avg_scrape_processing_time_ns(&self, req_processing_time: Duration) -> f64 {
         let req_processing_time = req_processing_time.as_nanos() as f64;

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

@@ -88,10 +88,15 @@ impl Repository {
         new_avg
     }
 
-    pub async fn recalculate_udp_avg_announce_processing_time_ns(&self, req_processing_time: Duration) -> f64 {
-        let stats_lock = self.stats.write().await;
+    pub async fn recalculate_udp_avg_announce_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_announce_processing_time_ns(req_processing_time);
+        let new_avg = stats_lock.recalculate_udp_avg_announce_processing_time_ns(req_processing_time, label_set, now);
 
         drop(stats_lock);
 
@@ -390,7 +395,9 @@ mod tests {
 
         // Calculate new average with processing time of 1500ns
         let processing_time = Duration::from_nanos(1500);
-        let new_avg = repo.recalculate_udp_avg_announce_processing_time_ns(processing_time).await;
+        let new_avg = repo
+            .recalculate_udp_avg_announce_processing_time_ns(processing_time, &announce_labels, now)
+            .await;
 
         // Moving average: previous_avg + (new_value - previous_avg) / total_announces
         // 500 + (1500 - 500) / 5 = 500 + 200 = 700
@@ -453,16 +460,18 @@ mod tests {
             .recalculate_udp_avg_connect_processing_time_ns(processing_time, &connect_labels, now)
             .await;
 
-        let _announce_labels = LabelSet::from([("request_kind", "announce")]);
-        let announce_avg = repo.recalculate_udp_avg_announce_processing_time_ns(processing_time).await;
+        let announce_labels = LabelSet::from([("request_kind", "announce")]);
+        let announce_avg = repo
+            .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;
 
         // 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.is_infinite() || announce_avg.is_nan());
+        assert!((announce_avg - 1000.0).abs() < f64::EPSILON);
         assert!(scrape_avg.is_infinite() || scrape_avg.is_nan());
     }