feat: add per-piece download duration metric split by traffic type

dragonfly-client-metric/src/lib.rs

@@ -215,6 +215,21 @@ pub static DOWNLOAD_TRAFFIC: LazyLock<IntCounterVec> = LazyLock::new(|| {
     .expect("metric can be created")
 });
 
+/// Used to record the download piece duration.
+pub static DOWNLOAD_PIECE_DURATION: LazyLock<HistogramVec> = LazyLock::new(|| {
+    HistogramVec::new(
+        HistogramOpts::new(
+            "download_piece_duration_milliseconds",
+            "Histogram of the download piece duration.",
+        )
+        .namespace(dragonfly_client_config::SERVICE_NAME)
+        .subsystem(dragonfly_client_config::NAME)
+        .buckets(exponential_buckets(1.0, 2.0, 24).unwrap()),
+        &["type"],
+    )
+    .expect("metric can be created")
+});
+
 /// Used to count the upload traffic.
 pub static UPLOAD_TRAFFIC: LazyLock<IntCounterVec> = LazyLock::new(|| {
     IntCounterVec::new(
@@ -637,6 +652,10 @@ fn register_custom_metrics() {
         .register(Box::new(DOWNLOAD_TRAFFIC.clone()))
         .expect("metric can be registered");
 
+    REGISTRY
+        .register(Box::new(DOWNLOAD_PIECE_DURATION.clone()))
+        .expect("metric can be registered");
+
     REGISTRY
         .register(Box::new(UPLOAD_TRAFFIC.clone()))
         .expect("metric can be registered");
@@ -752,6 +771,7 @@ fn reset_custom_metrics() {
     CONCURRENT_DOWNLOAD_TASK_GAUGE.reset();
     CONCURRENT_UPLOAD_PIECE_GAUGE.reset();
     DOWNLOAD_TRAFFIC.reset();
+    DOWNLOAD_PIECE_DURATION.reset();
     UPLOAD_TRAFFIC.reset();
     DOWNLOAD_TASK_DURATION.reset();
     BACKEND_REQUEST_COUNT.reset();
@@ -1051,6 +1071,13 @@ pub fn collect_download_piece_traffic_metrics(typ: &TrafficType, length: u64) {
         .inc_by(length);
 }
 
+/// Collects the download piece duration metrics.
+pub fn collect_download_piece_duration_metrics(typ: &TrafficType, cost: Duration) {
+    DOWNLOAD_PIECE_DURATION
+        .with_label_values(&[typ.as_str_name()])
+        .observe(cost.as_millis() as f64);
+}
+
 /// Collects the upload piece started metrics.
 pub fn collect_upload_piece_started_metrics() {
     CONCURRENT_UPLOAD_PIECE_GAUGE.with_label_values(&[]).inc();
@@ -1690,6 +1717,17 @@ mod tests {
         assert!(traffic >= 2048);
     }
 
+    #[test]
+    fn test_collect_download_piece_duration_metrics() {
+        let traffic_type = TrafficType::RemotePeer;
+        collect_download_piece_duration_metrics(&traffic_type, Duration::from_millis(42));
+
+        let count = DOWNLOAD_PIECE_DURATION
+            .with_label_values(&[traffic_type.as_str_name()])
+            .get_sample_count();
+        assert!(count >= 1);
+    }
+
     #[test]
     fn test_collect_backend_request_metrics() {
         collect_backend_request_started_metrics("http", "GET");

dragonfly-client/src/resource/piece.rs

@@ -24,7 +24,8 @@ use dragonfly_client_config::dfdaemon::Config;
 use dragonfly_client_core::{error::BackendError, Error, Result};
 use dragonfly_client_metric::{
     collect_backend_request_failure_metrics, collect_backend_request_finished_metrics,
-    collect_backend_request_started_metrics, collect_download_piece_traffic_metrics,
+    collect_backend_request_started_metrics, collect_download_piece_duration_metrics,
+    collect_download_piece_traffic_metrics,
 };
 use dragonfly_client_storage::{io::RangeReader, metadata, Storage};
 use dragonfly_client_util::net::format_socket_addr;
@@ -335,8 +336,13 @@ impl Piece {
         Span::current().record("piece_id", piece_id);
         Span::current().record("piece_length", length);
 
+        // Record the start time.
+        let start_time = Instant::now();
+
         // Upload the piece content.
         let (_, reader) = self.storage.upload_piece(piece_id, task_id, range).await?;
+        collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed());
+
         Ok(reader)
     }
 
@@ -396,6 +402,8 @@ impl Piece {
             .acquire(length as usize)
             .await;
 
+        // Record the start time.
+        let start_time = Instant::now();
         let (mut stream, offset, digest) = match (
             self.config.download.protocol.as_str(),
             parent.download_ip,
@@ -460,6 +468,10 @@ impl Piece {
         {
             Ok(piece) => {
                 collect_download_piece_traffic_metrics(&TrafficType::RemotePeer, length);
+                collect_download_piece_duration_metrics(
+                    &TrafficType::RemotePeer,
+                    start_time.elapsed(),
+                );
 
                 scopeguard::ScopeGuard::into_inner(guard);
                 Ok(piece)
@@ -621,6 +633,10 @@ impl Piece {
         {
             Ok(piece) => {
                 collect_download_piece_traffic_metrics(&TrafficType::BackToSource, length);
+                collect_download_piece_duration_metrics(
+                    &TrafficType::BackToSource,
+                    start_time.elapsed(),
+                );
 
                 scopeguard::ScopeGuard::into_inner(guard);
                 Ok(piece)
@@ -686,11 +702,16 @@ impl Piece {
         Span::current().record("piece_id", piece_id);
         Span::current().record("piece_length", length);
 
+        // Record the start time.
+        let start_time = Instant::now();
+
         // Upload the piece content.
         let (_, reader) = self
             .storage
             .upload_persistent_piece(piece_id, task_id, range)
             .await?;
+        collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed());
+
         Ok(reader)
     }
 
@@ -746,6 +767,8 @@ impl Piece {
             };
         });
 
+        // Record the start time.
+        let start_time = Instant::now();
         let (mut stream, offset, digest) = match (
             self.config.download.protocol.as_str(),
             parent.download_ip,
@@ -809,6 +832,10 @@ impl Piece {
         {
             Ok(piece) => {
                 collect_download_piece_traffic_metrics(&TrafficType::RemotePeer, length);
+                collect_download_piece_duration_metrics(
+                    &TrafficType::RemotePeer,
+                    start_time.elapsed(),
+                );
 
                 scopeguard::ScopeGuard::into_inner(guard);
                 Ok(piece)
@@ -968,6 +995,10 @@ impl Piece {
         {
             Ok(piece) => {
                 collect_download_piece_traffic_metrics(&TrafficType::BackToSource, length);
+                collect_download_piece_duration_metrics(
+                    &TrafficType::BackToSource,
+                    start_time.elapsed(),
+                );
 
                 scopeguard::ScopeGuard::into_inner(guard);
                 Ok(piece)
@@ -1033,11 +1064,16 @@ impl Piece {
         Span::current().record("piece_id", piece_id);
         Span::current().record("piece_length", length);
 
+        // Record the start time.
+        let start_time = Instant::now();
+
         // Upload the piece content.
         let (_, reader) = self
             .storage
             .upload_persistent_cache_piece(piece_id, task_id, range)
             .await?;
+        collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed());
+
         Ok(reader)
     }
 
@@ -1093,6 +1129,8 @@ impl Piece {
             };
         });
 
+        // Record the start time.
+        let start_time = Instant::now();
         let (mut stream, offset, digest) = match (
             self.config.download.protocol.as_str(),
             parent.download_ip,
@@ -1156,6 +1194,10 @@ impl Piece {
         {
             Ok(piece) => {
                 collect_download_piece_traffic_metrics(&TrafficType::RemotePeer, length);
+                collect_download_piece_duration_metrics(
+                    &TrafficType::RemotePeer,
+                    start_time.elapsed(),
+                );
 
                 scopeguard::ScopeGuard::into_inner(guard);
                 Ok(piece)