From 72b4530ad1dcfdc8a7181b81fd1567b276152276 Mon Sep 17 00:00:00 2001 From: Sophie Phoon Date: Wed, 26 Aug 2026 15:05:26 +0800 Subject: [PATCH 1/2] feat: add per-piece download duration metric split by traffic type --- dragonfly-client-metric/src/lib.rs | 38 ++++++++++++++++++++++++ dragonfly-client/src/resource/piece.rs | 40 +++++++++++++++++++++++++- 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/dragonfly-client-metric/src/lib.rs b/dragonfly-client-metric/src/lib.rs index 27d0f2fb..faa2a935 100644 --- a/dragonfly-client-metric/src/lib.rs +++ b/dragonfly-client-metric/src/lib.rs @@ -215,6 +215,21 @@ pub static DOWNLOAD_TRAFFIC: LazyLock = LazyLock::new(|| { .expect("metric can be created") }); +/// Used to record the download piece duration. +pub static DOWNLOAD_PIECE_DURATION: LazyLock = 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 = 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"); diff --git a/dragonfly-client/src/resource/piece.rs b/dragonfly-client/src/resource/piece.rs index 8c39c44d..09c305d2 100644 --- a/dragonfly-client/src/resource/piece.rs +++ b/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,9 @@ 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, @@ -443,6 +452,9 @@ impl Piece { } }; + // Collect the download piece duration metrics. + collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); + // Record the finish of downloading piece. match self .storage @@ -606,6 +618,9 @@ impl Piece { start_time.elapsed(), ); + // Collect the download piece duration metrics. + collect_download_piece_duration_metrics(&TrafficType::BackToSource, start_time.elapsed()); + let mut stream = response.reader.into_inner(); match self .storage @@ -686,11 +701,17 @@ impl Piece { Span::current().record("piece_id", piece_id); Span::current().record("piece_length", length); + let start_time = Instant::now(); + // Upload the piece content. let (_, reader) = self .storage .upload_persistent_piece(piece_id, task_id, range) .await?; + + // Collect the download piece duration metrics. + collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed()); + Ok(reader) } @@ -746,6 +767,8 @@ impl Piece { }; }); + let start_time = Instant::now(); + let (mut stream, offset, digest) = match ( self.config.download.protocol.as_str(), parent.download_ip, @@ -793,6 +816,8 @@ impl Piece { } }; + collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); + // Record the finish of downloading piece. match self .storage @@ -950,6 +975,9 @@ impl Piece { start_time.elapsed(), ); + // Collect the download piece duration metrics. + collect_download_piece_duration_metrics(&TrafficType::BackToSource, start_time.elapsed()); + // Record the finish of downloading piece. Consumes the stream of // bytes chunks underlying the reader, so the chunks are written to // the storage without copying. @@ -1033,11 +1061,17 @@ impl Piece { Span::current().record("piece_id", piece_id); Span::current().record("piece_length", length); + let start_time = Instant::now(); + // Upload the piece content. let (_, reader) = self .storage .upload_persistent_cache_piece(piece_id, task_id, range) .await?; + + // Collect the download piece duration metrics. + collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed()); + Ok(reader) } @@ -1093,6 +1127,8 @@ impl Piece { }; }); + let start_time = Instant::now(); + let (mut stream, offset, digest) = match ( self.config.download.protocol.as_str(), parent.download_ip, @@ -1140,6 +1176,8 @@ impl Piece { } }; + collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); + // Record the finish of downloading piece. match self .storage From 57e1f7c14a60a372106be679a201350e681df9bd Mon Sep 17 00:00:00 2001 From: Gaius Date: Fri, 28 Aug 2026 10:55:28 +0800 Subject: [PATCH 2/2] fix(piece): record piece download duration only on successful storage write Previously, `collect_download_piece_duration_metrics` was called immediately after the network transfer completed, before the piece was written to storage. This caused duration metrics to exclude storage write time and to be recorded even on storage failures. Now the metric is collected only inside the success branch after `storage` confirms the piece is written. Signed-off-by: Gaius --- dragonfly-client/src/resource/piece.rs | 46 ++++++++++++++------------ 1 file changed, 25 insertions(+), 21 deletions(-) diff --git a/dragonfly-client/src/resource/piece.rs b/dragonfly-client/src/resource/piece.rs index 09c305d2..5240ed74 100644 --- a/dragonfly-client/src/resource/piece.rs +++ b/dragonfly-client/src/resource/piece.rs @@ -338,9 +338,9 @@ impl Piece { // 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) @@ -404,7 +404,6 @@ 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, @@ -452,9 +451,6 @@ impl Piece { } }; - // Collect the download piece duration metrics. - collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); - // Record the finish of downloading piece. match self .storage @@ -472,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) @@ -618,9 +618,6 @@ impl Piece { start_time.elapsed(), ); - // Collect the download piece duration metrics. - collect_download_piece_duration_metrics(&TrafficType::BackToSource, start_time.elapsed()); - let mut stream = response.reader.into_inner(); match self .storage @@ -636,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) @@ -701,6 +702,7 @@ 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. @@ -708,8 +710,6 @@ impl Piece { .storage .upload_persistent_piece(piece_id, task_id, range) .await?; - - // Collect the download piece duration metrics. collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed()); Ok(reader) @@ -767,8 +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, @@ -816,8 +816,6 @@ impl Piece { } }; - collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); - // Record the finish of downloading piece. match self .storage @@ -834,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) @@ -975,9 +977,6 @@ impl Piece { start_time.elapsed(), ); - // Collect the download piece duration metrics. - collect_download_piece_duration_metrics(&TrafficType::BackToSource, start_time.elapsed()); - // Record the finish of downloading piece. Consumes the stream of // bytes chunks underlying the reader, so the chunks are written to // the storage without copying. @@ -996,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) @@ -1061,6 +1064,7 @@ 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. @@ -1068,8 +1072,6 @@ impl Piece { .storage .upload_persistent_cache_piece(piece_id, task_id, range) .await?; - - // Collect the download piece duration metrics. collect_download_piece_duration_metrics(&TrafficType::LocalPeer, start_time.elapsed()); Ok(reader) @@ -1127,8 +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, @@ -1176,8 +1178,6 @@ impl Piece { } }; - collect_download_piece_duration_metrics(&TrafficType::RemotePeer, start_time.elapsed()); - // Record the finish of downloading piece. match self .storage @@ -1194,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)