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..5240ed74 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,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)