diff --git a/Cargo.lock b/Cargo.lock index cf6f60cf3..8e8ccd7b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5303,6 +5303,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tracing", "url", "vt_path", "vt_str", diff --git a/crates/vt/src/session/cache/archive.rs b/crates/vt/src/session/cache/archive.rs index accf96796..480647299 100644 --- a/crates/vt/src/session/cache/archive.rs +++ b/crates/vt/src/session/cache/archive.rs @@ -73,6 +73,12 @@ pub fn create_output_archive( Ok(()) } +/// The size of the file at `path`, such as an archive, or `None` if it can't +/// be read. Only for logging. +pub fn file_size(path: &AbsolutePath) -> Option { + std::fs::metadata(path.as_path()).ok().map(|metadata| metadata.len()) +} + /// Extract a tar.zst archive, restoring files, folders and symbolic links /// relative to workspace root. /// diff --git a/crates/vt/src/session/cache/mod.rs b/crates/vt/src/session/cache/mod.rs index 179d94dda..e76ef7aad 100644 --- a/crates/vt/src/session/cache/mod.rs +++ b/crates/vt/src/session/cache/mod.rs @@ -11,7 +11,7 @@ use std::{ fs::File, io::Write, sync::{Arc, OnceLock}, - time::Duration, + time::{Duration, Instant}, }; // Re-export display functions for convenience @@ -31,6 +31,7 @@ use vt_plan::{ cache_metadata::{CacheMetadata, ExecutionCacheKey, SpawnFingerprint}, remote_cache::{RemoteCacheAccess, ResolvedRemoteCacheConfig}, }; +use vt_remote_cache::Fetched; use vt_str::Str; use wincode::{ SchemaRead, SchemaReadOwned, SchemaWrite, @@ -367,7 +368,8 @@ impl ExecutionCache { // Part of the key, so the same as when the entry was created. let input_negative_globs = PathGlobSet::new(&cache_metadata.input_config.negative_globs)?; - let local_miss = match self + let start = Instant::now(); + let local_lookup = self .try_hit_local( cache_metadata, &cache_key, @@ -375,8 +377,17 @@ impl ExecutionCache { &input_negative_globs, workspace_root, ) - .await? - { + .await?; + tracing::debug!( + outcome = match &local_lookup { + Ok(_) => "hit", + Err(CacheMiss::NotFound) => "not_found", + Err(_) => "mismatch", + }, + elapsed = ?start.elapsed(), + "looked up the local cache" + ); + let local_miss = match local_lookup { Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Local })), Err(miss) => miss, }; @@ -393,7 +404,8 @@ impl ExecutionCache { ) => remote_config, None => return Ok(Err(local_miss)), }; - let remote_miss = match self + let start = Instant::now(); + let remote_lookup = self .try_hit_remote( remote_config, cache_metadata, @@ -404,8 +416,18 @@ impl ExecutionCache { cache_dir, cancel_token, ) - .await? - { + .await?; + tracing::debug!( + outcome = match &remote_lookup { + Ok(_) => "hit", + Err(CacheMiss::NotFound) => "not_found", + Err(CacheMiss::FingerprintMismatch(_)) => "mismatch", + Err(CacheMiss::RemoteReadFailed(_)) => "read_failed", + }, + elapsed = ?start.elapsed(), + "looked up the remote cache" + ); + let remote_miss = match remote_lookup { Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Remote })), Err(miss) => miss, }; @@ -473,10 +495,24 @@ impl ExecutionCache { cache_dir: &AbsolutePath, cancel_token: &CancellationToken, ) -> anyhow::Result> { + let start = Instant::now(); let fetched = self .remote_clients .fetch(remote_config, cache_key, &cache_metadata.execution_cache_key, cancel_token) .await; + let (outcome, entries) = match &fetched { + Ok(Some(Fetched::Exact { entries })) => ("exact", entries.len()), + Ok(Some(Fetched::Fallback { .. })) => ("fallback", 0), + Ok(None) => ("not_found", 0), + Err(_) => ("failed", 0), + }; + tracing::debug!( + outcome, + entries, + error = fetched.as_ref().err().map(tracing::field::debug), + elapsed = ?start.elapsed(), + "fetched from the remote cache" + ); let validate = |cache_value: &CacheEntryValue| { cache_value.validate( &cache_metadata.unfiltered_envs, @@ -555,7 +591,9 @@ impl ExecutionCache { let cache_key = CacheEntryKey::from_metadata(cache_metadata); + let start = Instant::now(); self.record(&cache_key, execution_cache_key, &cache_value, cache_dir).await?; + tracing::debug!(elapsed = ?start.elapsed(), "recorded the cache entry locally"); let upload_error = Arc::new(OnceLock::new()); let remote_config = match &cache_metadata.remote_cache { @@ -611,6 +649,7 @@ impl ExecutionCache { workspace_root: &AbsolutePath, cache_dir: &AbsolutePath, ) -> anyhow::Result<()> { + let start = Instant::now(); if let Some(archive_name) = &hit.value.output_archive { let archive_path = cache_dir.join(archive_name.as_str()); if let Err(err) = archive::extract_output_archive(workspace_root, &archive_path) { @@ -620,6 +659,11 @@ impl ExecutionCache { } return Err(err.context("failed to extract the output archive")); } + tracing::debug!( + bytes = archive::file_size(&archive_path), + elapsed = ?start.elapsed(), + "extracted the output archive" + ); } if hit.source == CacheHitSource::Remote { @@ -633,6 +677,7 @@ impl ExecutionCache { tracing::warn!(?err, "failed to record a remote cache hit locally"); } } + tracing::debug!(source = ?hit.source, elapsed = ?start.elapsed(), "restored the outputs"); Ok(()) } } diff --git a/crates/vt/src/session/cache/remote.rs b/crates/vt/src/session/cache/remote.rs index 7afdac876..fa39efe1d 100644 --- a/crates/vt/src/session/cache/remote.rs +++ b/crates/vt/src/session/cache/remote.rs @@ -17,12 +17,14 @@ use std::{ fs::File, io::{self, Read as _, Write as _}, sync::{Arc, Mutex, OnceLock, PoisonError}, + time::Instant, }; use bytes::Bytes; use rustc_hash::FxHashMap; use tokio::sync::mpsc; use tokio_util::{sync::CancellationToken, task::TaskTracker}; +use tracing::Instrument as _; use vt_path::AbsolutePath; use vt_plan::{ cache_metadata::ExecutionCacheKey, @@ -142,8 +144,18 @@ pub(super) fn resolve( None => return Err(CacheMiss::NotFound), }; let mut newest_miss = None; - for entry in entries { - match check_entry(entry, &mut validate) { + for (index, entry) in entries.into_iter().enumerate() { + let start = Instant::now(); + let value_bytes = entry.value.len(); + let checked = check_entry(entry, &mut validate); + tracing::debug!( + index, + value_bytes, + valid = checked.is_ok(), + elapsed = ?start.elapsed(), + "checked a remote cache entry" + ); + match checked { Ok(restore) => return Ok(restore), Err(miss) => { newest_miss.get_or_insert(miss); @@ -238,6 +250,7 @@ impl RemoteClients { cache_dir: &AbsolutePath, cancel_token: &CancellationToken, ) -> Result { + let start = Instant::now(); let client = self.client(remote_config).map_err(ReadError::Download)?; let archive_name = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4()); let archive_path = cache_dir.join(archive_name.as_str()); @@ -251,6 +264,12 @@ impl RemoteClients { // Best-effort cleanup: the file may not have been created. let _ = std::fs::remove_file(temp_path.as_path()); } + tracing::debug!( + ok = result.is_ok(), + bytes = result.as_ref().ok().and_then(|&()| archive::file_size(&archive_path)), + elapsed = ?start.elapsed(), + "downloaded an output archive" + ); result.map(|()| archive_name) } @@ -268,11 +287,18 @@ impl RemoteClients { cache_value: &CacheEntryValue, cache_dir: &AbsolutePath, ) -> Result> + Send + use<>, UploadError> { + let start = Instant::now(); let client = self.client(remote_config)?; let key = encode_key(cache_key)?; let secondary_key = encode_key(execution_cache_key)?; let value = encode_value(cache_value)?; let archive = cache_value.output_archive.as_ref().map(|name| cache_dir.join(name.as_str())); + tracing::debug!( + value_bytes = value.len(), + archive = archive.is_some(), + elapsed = ?start.elapsed(), + "prepared a remote cache upload" + ); Ok(async move { client.store(&key, &secondary_key, &value, archive.as_deref()).await?; Ok(()) @@ -311,16 +337,28 @@ impl RemoteUploads { error: Arc>, ) { let cancel = self.cancel.clone(); - self.tracker.spawn(async move { - // A cancelled upload sets its error itself, instead of being - // aborted, so every cancelled upload has one. - let result = - cancel.run_until_cancelled(upload).await.unwrap_or(Err(UploadError::Interrupted)); - if let Err(err) = result { - tracing::debug!(?err, "remote cache upload failed"); - let _ = error.set(err); + // The upload outlives its task, so it carries the task's span. + self.tracker.spawn( + async move { + let start = Instant::now(); + // A cancelled upload sets its error itself, instead of being + // aborted, so every cancelled upload has one. + let result = cancel + .run_until_cancelled(upload) + .await + .unwrap_or(Err(UploadError::Interrupted)); + match result { + Ok(()) => { + tracing::debug!(elapsed = ?start.elapsed(), "remote cache upload finished"); + } + Err(err) => { + tracing::debug!(?err, elapsed = ?start.elapsed(), "remote cache upload failed"); + let _ = error.set(err); + } + } } - }); + .in_current_span(), + ); } /// The number of uploads still running. @@ -331,6 +369,8 @@ impl RemoteUploads { /// Wait for all uploads to finish. If `interrupt_token` is cancelled /// first, or already was, cancel them and wait for them to stop. pub(super) async fn wait(&self, interrupt_token: &CancellationToken) { + let start = Instant::now(); + let pending = self.tracker.len(); self.tracker.close(); tokio::select! { biased; @@ -340,6 +380,14 @@ impl RemoteUploads { self.tracker.wait().await; } } + if pending > 0 { + tracing::debug!( + pending, + interrupted = interrupt_token.is_cancelled(), + elapsed = ?start.elapsed(), + "waited for remote cache uploads" + ); + } self.tracker.reopen(); } } diff --git a/crates/vt/src/session/execute/cache_update.rs b/crates/vt/src/session/execute/cache_update.rs index f44791a34..aeb1ac3c3 100644 --- a/crates/vt/src/session/execute/cache_update.rs +++ b/crates/vt/src/session/execute/cache_update.rs @@ -1,7 +1,11 @@ //! Post-run cache update: decide whether a finished spawn may be cached and, //! if so, store its fingerprint, captured output, and output archive. -use std::{collections::BTreeMap, sync::Arc, time::Duration}; +use std::{ + collections::BTreeMap, + sync::Arc, + time::{Duration, Instant}, +}; use rustc_hash::FxHashSet; use tokio_util::sync::CancellationToken; @@ -460,7 +464,15 @@ fn collect_and_archive_outputs( let archive_name: Str = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4()); let archive_path = cache_dir.join(archive_name.as_str()); + let start = Instant::now(); archive::create_output_archive(workspace_root, &sorted_files, &output_folders, &archive_path)?; + tracing::debug!( + files = sorted_files.len(), + folders = output_folders.len(), + bytes = archive::file_size(&archive_path), + elapsed = ?start.elapsed(), + "archived the outputs" + ); Ok(Some(archive_name)) } diff --git a/crates/vt/src/session/execute/fingerprint.rs b/crates/vt/src/session/execute/fingerprint.rs index 35c0cdb6f..baec46628 100644 --- a/crates/vt/src/session/execute/fingerprint.rs +++ b/crates/vt/src/session/execute/fingerprint.rs @@ -9,6 +9,7 @@ use std::{ fs::File, io::{self, BufRead}, sync::Arc, + time::Instant, }; use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; @@ -167,6 +168,10 @@ impl PostRunFingerprint { tracked_envs: BTreeMap>, tracked_env_queries: BTreeMap>, ) -> anyhow::Result { + let start = Instant::now(); + // Rayon's threads don't run in this thread's span, so the per-input + // events name it as their parent, which keeps them under the task. + let span = tracing::Span::current(); let inferred_inputs = inferred_path_reads .par_iter() .filter(|(path, _)| { @@ -175,6 +180,11 @@ impl PostRunFingerprint { .map(|(relative_path, path_read)| { let full_path = Arc::::from(base_dir.join(relative_path)); let mut fingerprint = fingerprint_path(&full_path, *path_read)?; + tracing::trace!( + parent: &span, + path = relative_path.as_str(), + "fingerprinted an inferred input" + ); if let PathFingerprint::Folder(Some(listing)) = &mut fingerprint { if !rebuildable.is_empty() { let DirListing { entries, ignored_entries } = listing; @@ -198,6 +208,12 @@ impl PostRunFingerprint { Ok((relative_path.clone(), InferredInput { fingerprint, may_be_missing })) }) .collect::>>()?; + tracing::debug!( + reads = inferred_path_reads.len(), + inputs = inferred_inputs.len(), + elapsed = ?start.elapsed(), + "created the post-run fingerprint" + ); Ok(Self { inferred_inputs, tracked_envs, tracked_env_queries }) } @@ -220,6 +236,9 @@ impl PostRunFingerprint { input_negative_globs: &PathGlobSet<'_>, unfiltered_envs: &FxHashMap>, Arc>, ) -> anyhow::Result> { + let start = Instant::now(); + // As in `create`, the per-input events name this span as their parent. + let span = tracing::Span::current(); let input_mismatch = self.inferred_inputs.par_iter().find_map_any(|(input_relative_path, input)| { let InferredInput { fingerprint: path_fingerprint, may_be_missing } = input; @@ -243,9 +262,15 @@ impl PostRunFingerprint { current.ignored_entries.clone_from(&stored.ignored_entries); leave_out_excluded_entries(current, input_relative_path, input_negative_globs); } - if path_fingerprint == ¤t_path_fingerprint - || (*may_be_missing && current_path_fingerprint == PathFingerprint::NotFound) - { + let unchanged = path_fingerprint == ¤t_path_fingerprint + || (*may_be_missing && current_path_fingerprint == PathFingerprint::NotFound); + tracing::trace!( + parent: &span, + path = input_relative_path.as_str(), + unchanged, + "validated an inferred input" + ); + if unchanged { None } else { let (kind, entry_name) = @@ -263,6 +288,12 @@ impl PostRunFingerprint { Some(Ok(PostRunMismatch::Input { kind, path })) } }); + tracing::debug!( + inputs = self.inferred_inputs.len(), + valid = input_mismatch.is_none(), + elapsed = ?start.elapsed(), + "validated the inferred inputs" + ); if let Some(result) = input_mismatch { return result.map(Some); } diff --git a/crates/vt/src/session/execute/mod.rs b/crates/vt/src/session/execute/mod.rs index 3d2e4ab47..2d3b91b42 100644 --- a/crates/vt/src/session/execute/mod.rs +++ b/crates/vt/src/session/execute/mod.rs @@ -459,6 +459,7 @@ async fn run( ) .await .map_err(|err| Report::failed(ExecutionError::Spawn(err)))?; + tracing::debug!(fspy = fspy_enabled, elapsed = ?start.elapsed(), "spawned the task process"); // 6. Drain the pipes and wait for exit, concurrently with the IPC driver. // The driver must be polled during pipe drain — otherwise a tool doing @@ -561,6 +562,7 @@ async fn lookup_cache<'a>( // Compute globbed inputs from positive globs at execution time. // Globs are already workspace-root-relative (resolved at task graph stage). + let start = Instant::now(); let globbed_inputs = compute_globbed_inputs( workspace_root, &cache_metadata.input_config.positive_globs, @@ -569,6 +571,11 @@ async fn lookup_cache<'a>( .map_err(|err| { Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err }) })?; + tracing::debug!( + files = globbed_inputs.len(), + elapsed = ?start.elapsed(), + "hashed the globbed inputs" + ); match cache .try_hit(cache_metadata, &globbed_inputs, workspace_root, cache_dir, cancel_token) @@ -634,6 +641,7 @@ async fn run_child( stop_accepting: Option<&StopAccepting>, fast_fail_token: CancellationToken, ) -> Result { + let start = Instant::now(); let pipe_result: Result<(), ExecutionError> = if let Some(sinks) = sinks { let stdout = child.stdout.take().expect("SpawnStdio::Piped yields a stdout pipe"); let stderr = child.stderr.take().expect("SpawnStdio::Piped yields a stderr pipe"); @@ -663,5 +671,13 @@ async fn run_child( if let Some(stop_accepting) = stop_accepting { stop_accepting.signal(); } + if let Ok(outcome) = &wait_result { + tracing::debug!( + success = outcome.exit_status.success(), + exit_code = outcome.exit_status.code(), + elapsed = ?start.elapsed(), + "the task process exited" + ); + } wait_result } diff --git a/crates/vt/src/session/execute/scheduler.rs b/crates/vt/src/session/execute/scheduler.rs index fb605a4a4..9dcc7b588 100644 --- a/crates/vt/src/session/execute/scheduler.rs +++ b/crates/vt/src/session/execute/scheduler.rs @@ -2,7 +2,7 @@ //! (dependency order, concurrency limits, fast-fail), and hands each leaf to //! [`execute_spawn`] which owns *how* a single spawn runs. -use std::{cell::RefCell, ffi::OsStr, io::Write as _, num::NonZeroUsize, sync::Arc}; +use std::{cell::RefCell, ffi::OsStr, io::Write as _, num::NonZeroUsize, sync::Arc, time::Instant}; use futures_util::{FutureExt, StreamExt, future::LocalBoxFuture, stream::FuturesUnordered}; use petgraph::Direction; @@ -83,7 +83,11 @@ impl ExecutionContext<'_> { /// (killing in-flight child processes). Ctrl-C cancels the `cancel_token`. /// Either cancellation causes this method to close the semaphore, drain /// remaining futures, and return. - #[tracing::instrument(level = "debug", skip_all)] + #[tracing::instrument( + level = "debug", + skip_all, + fields(tasks = graph.graph.node_count(), concurrency_limit = graph.concurrency_limit) + )] async fn execute_expanded_graph(&self, graph: &ExecutionGraph) { if graph.graph.node_count() == 0 { return; @@ -144,9 +148,15 @@ impl ExecutionContext<'_> { ) -> LocalBoxFuture<'a, ExecutionNodeIndex> { let sem = semaphore.clone(); async move { + let start = Instant::now(); if let Ok(_permit) = sem.acquire_owned().await && !self.cancelled() { + tracing::debug!( + task = %graph.graph[node_ix].task_display, + waited = ?start.elapsed(), + "acquired a concurrency permit" + ); self.execute_node(graph, node_ix).await; } node_ix @@ -183,11 +193,21 @@ impl ExecutionContext<'_> { /// from the graph reporter and delegates to the appropriate execution /// method. On failure (non-zero exit or infrastructure error), cancels the /// `fast_fail_token`. - #[tracing::instrument(level = "debug", skip_all)] - async fn execute_leaf(&self, display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind) { + #[tracing::instrument( + level = "debug", + skip_all, + fields(task = %item_display.task_display, command = %item_display.command) + )] + async fn execute_leaf( + &self, + item_display: &ExecutionItemDisplay, + leaf_kind: &LeafExecutionKind, + ) { + let start = Instant::now(); // Borrow the reporter briefly to create the leaf reporter, then drop // the RefCell guard before any `.await` point. - let mut leaf_reporter = self.reporter.borrow_mut().new_leaf_execution(display, leaf_kind); + let mut leaf_reporter = + self.reporter.borrow_mut().new_leaf_execution(item_display, leaf_kind); let failed = match leaf_kind { LeafExecutionKind::InProcess(in_process_execution) => { @@ -226,6 +246,7 @@ impl ExecutionContext<'_> { } } }; + tracing::debug!(failed, elapsed = ?start.elapsed(), "finished the task item"); if failed { self.fast_fail_token.cancel(); } diff --git a/crates/vt_remote_cache/Cargo.toml b/crates/vt_remote_cache/Cargo.toml index 13d7e8b26..607665be6 100644 --- a/crates/vt_remote_cache/Cargo.toml +++ b/crates/vt_remote_cache/Cargo.toml @@ -24,6 +24,7 @@ serde_bytes = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["fs", "sync"] } +tracing = { workspace = true } url = { workspace = true } vt_path = { workspace = true } vt_str = { workspace = true } diff --git a/crates/vt_remote_cache/src/auth/github_oidc.rs b/crates/vt_remote_cache/src/auth/github_oidc.rs index b705ee1b1..243c62b8c 100644 --- a/crates/vt_remote_cache/src/auth/github_oidc.rs +++ b/crates/vt_remote_cache/src/auth/github_oidc.rs @@ -3,7 +3,7 @@ use std::{ fmt, sync::Arc, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; @@ -164,6 +164,7 @@ impl TokenRequest { } async fn send(&self, http: &reqwest::Client) -> Result { + let start = Instant::now(); let network_error = |err: reqwest::Error| TokenError::Network(err.without_url()); let response = http .get(self.url.clone()) @@ -174,6 +175,12 @@ impl TokenRequest { .await .map_err(network_error)?; let status = response.status(); + // Neither the request URL nor the tokens are logged. + tracing::debug!( + status = status.as_u16(), + elapsed = ?start.elapsed(), + "GitHub Actions OIDC token responded" + ); if status != StatusCode::OK { return Err(TokenError::Status(status)); } diff --git a/crates/vt_remote_cache/src/lib.rs b/crates/vt_remote_cache/src/lib.rs index 72f402ee8..5987090f8 100644 --- a/crates/vt_remote_cache/src/lib.rs +++ b/crates/vt_remote_cache/src/lib.rs @@ -5,7 +5,10 @@ pub mod auth; #[cfg(test)] mod test_server; -use std::{sync::Arc, time::Duration}; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; use bytes::Bytes; use reqwest::{ @@ -192,7 +195,9 @@ impl Client { /// Returns an error if the credentials can't be obtained, the request /// fails, the server responds with a status other than 200 or 404, or the /// body of a 200 response isn't an exact or fallback match. + #[tracing::instrument(level = "debug", skip_all)] pub async fn fetch(&self, key: &[u8], secondary_key: &[u8]) -> Result, Error> { + let start = Instant::now(); let headers = self.auth_headers(Operation::Fetch).await?; let body = encode_cbor(&FetchRequest { key, secondary_key }); let response = self @@ -204,6 +209,7 @@ impl Client { .send() .await .map_err(network_error)?; + log_response(&response, start); if response.status() == StatusCode::NOT_FOUND { // Read the body so the connection can be reused. It doesn't matter // if that fails, because the status alone is the answer. @@ -211,6 +217,7 @@ impl Client { return Ok(None); } let body = check_status(response).await?.bytes().await.map_err(network_error)?; + tracing::debug!(bytes = body.len(), elapsed = ?start.elapsed(), "remote cache fetch received"); decode_fetched(&body).map(Some) } @@ -222,11 +229,14 @@ impl Client { /// /// Returns an error if the credentials can't be obtained, the request /// fails, or the server responds with a status other than 200. + #[tracing::instrument(level = "debug", skip_all)] pub async fn download(&self, blob_id: &str) -> Result { + let start = Instant::now(); let mut url = self.blob_url.clone(); url.path_segments_mut().map_err(|()| Error::InvalidEndpoint(None))?.push(blob_id); let headers = self.auth_headers(Operation::Download).await?; let response = self.http.get(url).headers(headers).send().await.map_err(network_error)?; + log_response(&response, start); Ok(Download { response: check_status(response).await? }) } @@ -239,6 +249,7 @@ impl Client { /// Returns an error if the credentials can't be obtained, the blob file /// can't be opened, the request fails, or the server responds with a /// status other than 200. + #[tracing::instrument(level = "debug", skip_all)] pub async fn store( &self, key: &[u8], @@ -246,12 +257,24 @@ impl Client { value: &[u8], blob: Option<&AbsolutePath>, ) -> Result<(), Error> { + let start = Instant::now(); let headers = self.auth_headers(Operation::Store).await?; + let auth = start.elapsed(); let metadata = StoreMetadata { key, secondary_key, value }; let mut form = Form::new().part("metadata", metadata_part(&metadata)); - if let Some(blob) = blob { - form = form.part("blob", blob_part(blob).await?); - } + let blob_bytes = if let Some(blob) = blob { + let (part, length) = blob_part(blob).await?; + form = form.part("blob", part); + Some(length) + } else { + None + }; + tracing::debug!( + value_bytes = value.len(), + blob_bytes, + ?auth, + "sending a remote cache store" + ); let response = self .http .post(self.store_url.clone()) @@ -260,6 +283,7 @@ impl Client { .send() .await .map_err(network_error)?; + log_response(&response, start); // The response's blob ID isn't needed. Read the body anyway, so the // connection can be reused. check_status(response).await?.bytes().await.map_err(network_error)?; @@ -267,6 +291,18 @@ impl Client { } } +/// Log the status and length of `response`, and the time since `start`, when +/// the operation started. The request URL is left out, since the endpoint may +/// contain credentials. +fn log_response(response: &Response, start: Instant) { + tracing::debug!( + status = response.status().as_u16(), + content_length = response.content_length(), + elapsed = ?start.elapsed(), + "remote cache responded" + ); +} + /// Only HTTP 200 counts as success. For other statuses, the error includes the /// message in the response body. async fn check_status(response: Response) -> Result { @@ -322,12 +358,14 @@ fn metadata_part(metadata: &StoreMetadata<'_>) -> Part { Part::bytes(encode_cbor(metadata)).mime_str("application/cbor").expect("valid MIME type") } -async fn blob_part(path: &AbsolutePath) -> Result { +/// The blob part streamed from the file at `path`, and its length. +async fn blob_part(path: &AbsolutePath) -> Result<(Part, u64), Error> { let file = tokio::fs::File::open(path).await.map_err(Error::ReadBlob)?; let length = file.metadata().await.map_err(Error::ReadBlob)?.len(); - Ok(Part::stream_with_length(file, length) + let part = Part::stream_with_length(file, length) .mime_str("application/octet-stream") - .expect("valid MIME type")) + .expect("valid MIME type"); + Ok((part, length)) } #[cfg(test)]