diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index 6fddb93bf7..deb1c281b9 100644 --- a/nativelink-config/src/cas_server.rs +++ b/nativelink-config/src/cas_server.rs @@ -1299,6 +1299,15 @@ pub struct LocalWorkerConfig { /// `DeadlineExceeded` and may be retried by the scheduler, and a warning /// names the action every minute before that. Value in seconds. /// + /// A fetch that fails with a transient store error is fetched again + /// from this worker, with a pause doubling from 1 s to 30 s, for as + /// long as this leaves room for another attempt, or the action's own + /// timeout when that is shorter. The scheduler is not told of the + /// retries: it ends an executing action on its own at the action's + /// timeout plus `worker_timeout_s`, and at + /// `max_action_executing_timeout_s`, so keep this below those or the + /// action is requeued while the worker still retries. + /// /// Default: 10 minutes #[serde( default, diff --git a/nativelink-util/src/metrics.rs b/nativelink-util/src/metrics.rs index 1850793117..f44b024cd3 100644 --- a/nativelink-util/src/metrics.rs +++ b/nativelink-util/src/metrics.rs @@ -905,6 +905,11 @@ pub static WORKER_METRICS: LazyLock = LazyLock::new(|| { .with_description("Input hardlinks made again because the fast store evicted their source while the tree was staged, by kind: plain, executable, or exhausted when the retries ran out") .build(), + prepare_retries: meter + .u64_counter("worker.prepare.retries") + .with_description("Input fetches tried again on the same worker after a transient store failure: retried per attempt made again, then one of recovered, exhausted when the download deadline ran out, failed when a later attempt failed for good, or abandoned when a kill or the download timeout ended the fetch mid-retry") + .build(), + dispatch_requeues: meter .u64_counter("scheduler.dispatch.requeues") .with_description("Dispatches sent back to the queue untried, by reason: declined, channel_full, unacknowledged") @@ -954,6 +959,8 @@ pub struct WorkerMetrics { pub worker_disconnections: metrics::Counter, /// Input hardlinks made again after their source was evicted mid-staging, by kind. pub input_relinks: metrics::Counter, + /// Input fetches tried again on the worker after a transient store failure: `retried` per attempt made again, then one of `recovered`, `exhausted`, `failed` or `abandoned`. + pub prepare_retries: metrics::Counter, /// Dispatches requeued untried, by reason. pub dispatch_requeues: metrics::Counter, /// Keepalives received, cumulative. @@ -1017,6 +1024,17 @@ pub fn record_input_relink(kind: &'static str) { .add(1, &[KeyValue::new("kind", kind)]); } +/// Records an input fetch the worker tried again after a transient store +/// failure: `retried` per attempt made again, then one terminal outcome, +/// `recovered`, `exhausted` (the download deadline ran out), `failed` (a +/// later attempt failed for good) or `abandoned` (a kill or the download +/// timeout ended it mid-retry). +pub fn record_prepare_retry(outcome: &'static str) { + WORKER_METRICS + .prepare_retries + .add(1, &[KeyValue::new("outcome", outcome)]); +} + /// Records a keepalive from a worker. pub fn record_worker_keepalive() { WORKER_METRICS.worker_keepalives.add(1, &[]); diff --git a/nativelink-worker/Cargo.toml b/nativelink-worker/Cargo.toml index b68888bd8e..9d01b03a83 100644 --- a/nativelink-worker/Cargo.toml +++ b/nativelink-worker/Cargo.toml @@ -78,6 +78,9 @@ serial_test = { version = "4.0.0", features = [ "async", ], default-features = false } tempfile = { version = "3.15.0", default-features = false } +tokio = { version = "1.52.2", features = [ + "test-util", +], default-features = false } tonic-prost = { version = "0.14.6", default-features = false } tracing-test = { version = "0.2.5", default-features = false, features = [ "no-env-filter", diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs index 08082490d4..dece024ae5 100644 --- a/nativelink-worker/src/running_actions_manager.rs +++ b/nativelink-worker/src/running_actions_manager.rs @@ -68,7 +68,7 @@ use nativelink_util::action_messages::{ }; use nativelink_util::common::{DigestInfo, fs}; use nativelink_util::digest_hasher::{DigestHasher, DigestHasherFunc}; -use nativelink_util::metrics::record_input_relink; +use nativelink_util::metrics::{record_input_relink, record_prepare_retry}; use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime}; use nativelink_util::store_trait::{Store, StoreLike, UploadSizeInfo}; use nativelink_util::task::JoinHandleDropGuard; @@ -920,6 +920,100 @@ fn input_fetch_pause(attempt: u32, digest: &DigestInfo) -> Duration { Duration::from_millis(INPUT_FETCH_PAUSE_STEP_MS * attempt + jitter) } +/// The pause before a whole input tree is fetched again doubles from here. +const PREPARE_RETRY_PAUSE_MS: u64 = 1000; + +/// The pause before a whole input tree is fetched again never exceeds this. +const PREPARE_RETRY_PAUSE_CAP_MS: u64 = 30_000; + +/// Whether a failed input fetch is worth trying again from this worker. +/// These are the codes the store retrier itself treats as transient; a +/// throttled object store surfaces as `Internal` once the retrier gives up, +/// and an unclassified store failure as `Unknown`. The worker's own +/// filesystem lands on those two codes as well: a full disk, a blob at +/// its link limit or a read-only mount map to `Unknown`, a broken pipe to +/// `Internal`. A worker whose disk is the problem has to hand the action +/// back so it lands on a healthy one, not fetch into the same disk until +/// the deadline, so an operating system error is never retried here. Such +/// an error carries the OS error number in its message; a store's does not. +fn is_transient_fetch_error(err: &Error) -> bool { + match err.code { + Code::Unavailable + | Code::ResourceExhausted + | Code::Aborted + | Code::DeadlineExceeded + | Code::DataLoss => true, + Code::Internal | Code::Unknown => !is_os_error(err), + _ => false, + } +} + +/// Whether the error was raised by the operating system on this worker: +/// `std::io::Error` prints those as `... (os error N)`. +fn is_os_error(err: &Error) -> bool { + err.messages + .iter() + .any(|message| message.contains("(os error ")) +} + +/// Pause before fetching the input tree again: 1 s, 2 s, 4 s ... capped at +/// 30 s, shortened by up to a quarter with a jitter taken from the digest so +/// the workers that failed together do not all come back together. +fn prepare_retry_pause(attempt: u32, digest: &DigestInfo) -> Duration { + let doublings = attempt.saturating_sub(1).min(5); + let base_ms = (PREPARE_RETRY_PAUSE_MS << doublings).min(PREPARE_RETRY_PAUSE_CAP_MS); + let jitter_byte = (u64::from(digest.packed_hash()[0]) * u64::from(attempt)) % 256; + Duration::from_millis(base_ms - base_ms * jitter_byte / 1024) +} + +/// Counts the retries of one input fetch on `worker.prepare.retries` and +/// settles it on exactly one terminal outcome once a retry was made: +/// `recovered`, `exhausted`, `failed` when a later attempt failed for good, +/// or `abandoned` when the fetch was dropped mid-retry by a kill or the +/// download timeout. A fetch that succeeded or failed first time records +/// nothing, so the terminal outcomes add up to the fetches retried. +struct PrepareRetries { + record: R, + retries: u32, + settled: bool, +} + +impl PrepareRetries { + const fn new(record: R) -> Self { + Self { + record, + retries: 0, + settled: false, + } + } + + fn retried(&mut self) { + self.retries += 1; + (self.record)("retried"); + } + + fn settle(mut self, outcome: &'static str) { + self.settled = true; + if self.retries > 0 { + (self.record)(outcome); + } + } +} + +impl Drop for PrepareRetries { + fn drop(&mut self) { + if !self.settled && self.retries > 0 { + (self.record)("abandoned"); + } + } +} + +/// Records on `worker.prepare.retries`. A named function rather than the +/// helper itself, so the metrics reference generator sees the call. +fn record_prepare_outcome(outcome: &'static str) { + record_prepare_retry(outcome); +} + /// Fetches a blob into the fast store with `populate` and resolves it there /// with `resolve`, again when the store has taken it away in between. async fn fetch_and_resolve( @@ -2326,7 +2420,109 @@ impl RunningActionImpl { /// /// This function will aggressively download and spawn potentially thousands of futures. It is /// up to the stores to rate limit if needed. - async fn inner_prepare_action(self: Arc) -> Result, Error> { + /// Fetches the Command and the input tree into the work directory, + /// again after a transient store failure while `deadline` (none: + /// unbounded) leaves room for another attempt. A worker that already + /// holds part of the tree in its fast store finishes the job here rather + /// than handing the action back for another worker to fetch everything + /// against the same struggling store. The two fetches are retried + /// together: a Command read that fails during the same storm would + /// otherwise fail the action while the inputs were still being retried. + async fn fetch_with_retry( + &self, + filesystem_store: Pin<&FilesystemStore>, + deadline: Option, + ) -> Result { + let started = Instant::now(); + let mut attempt: u32 = 1; + let mut retries = PrepareRetries::new(record_prepare_outcome); + loop { + let command_fut = self.metrics().get_proto_command_from_store.wrap(async { + get_and_decode_digest::( + self.running_actions_manager.cas_store.as_ref(), + self.action_info.command_digest.into(), + ) + .await + .err_tip(|| "Converting command_digest to Command") + }); + let inputs_fut = + self.metrics() + .download_to_directory + .wrap(prepare_action_inputs_with_lease( + &self.running_actions_manager.directory_cache, + &self.running_actions_manager.cas_store, + filesystem_store, + &self.action_info.input_root_digest, + &self.work_directory, + self.input_lease.clone(), + )); + let err = match try_join(command_fut, inputs_fut).await { + Ok((command, ())) => { + if attempt > 1 { + info!( + operation_id = %self.operation_id, + attempt, + elapsed_s = started.elapsed().as_secs(), + "prepare_action: input fetch recovered" + ); + } + retries.settle("recovered"); + return Ok(command); + } + Err(err) if is_transient_fetch_error(&err) => err, + Err(err) => { + retries.settle("failed"); + return Err(err); + } + }; + let pause = prepare_retry_pause(attempt, &self.action_info.input_root_digest); + if deadline.is_some_and(|deadline| Instant::now() + pause >= deadline) { + retries.settle("exhausted"); + return Err(err.append(format!( + "Input fetch of operation {} failed {attempt} time(s) over {}s; the next retry would start after max_download_timeout", + self.operation_id, + started.elapsed().as_secs() + ))); + } + warn!( + operation_id = %self.operation_id, + attempt, + wait_ms = pause.as_millis(), + error = %err, + "prepare_action: input fetch failed, retrying" + ); + retries.retried(); + // Start the next attempt from an empty directory: the + // directories and links of the partial tree would otherwise + // fail it with AlreadyExists. + if let Err(err) = self.clear_work_directory().await { + retries.settle("failed"); + return Err(err); + } + tokio::time::sleep(pause).await; + attempt += 1; + } + } + + /// Empties the work directory between two fetch attempts. + async fn clear_work_directory(&self) -> Result<(), Error> { + if let Err(remove_err) = fs::remove_dir_all(&self.work_directory).await + && remove_err.code != Code::NotFound + { + return Err(remove_err.append(format!( + "Failed to clear work directory {} before fetching the inputs again", + self.work_directory + ))); + } + fs::create_dir(&self.work_directory) + .await + .err_tip(|| format!("Error recreating work directory {}", self.work_directory)) + } + + async fn inner_prepare_action( + self: Arc, + download_deadline: Option, + ) -> Result, Error> { // Before anything lands on disk: an action that reserved more than // is free is refused as backpressure, not run into ENOSPC halfway // through its inputs. Free means free after what the actions @@ -2371,41 +2567,18 @@ impl RunningActionImpl { (self.running_actions_manager.callbacks.now_fn)(); } let command = { - // Download and build out our input files/folders. Also fetch and decode our Command. - let command_fut = self.metrics().get_proto_command_from_store.wrap(async { - get_and_decode_digest::( - self.running_actions_manager.cas_store.as_ref(), - self.action_info.command_digest.into(), - ) - .await - .err_tip(|| "Converting command_digest to Command") - .map_err(tag_missing_input) - }); let filesystem_store_pin = Pin::new(self.running_actions_manager.filesystem_store.as_ref()); - let (command, ()) = try_join(command_fut, async { - fs::create_dir(&self.work_directory) - .await - .err_tip(|| format!("Error creating work directory {}", self.work_directory))?; - // Now the work directory has been created, we have to clean up. - self.did_cleanup.store(false, Ordering::Release); - // Download the input files/folder and place them into the temp directory. - // Use directory cache if available for better performance. - self.metrics() - .download_to_directory - .wrap(prepare_action_inputs_with_lease( - &self.running_actions_manager.directory_cache, - &self.running_actions_manager.cas_store, - filesystem_store_pin, - &self.action_info.input_root_digest, - &self.work_directory, - self.input_lease.clone(), - )) - .await - .map_err(tag_missing_input) - }) - .await?; - command + fs::create_dir(&self.work_directory) + .await + .err_tip(|| format!("Error creating work directory {}", self.work_directory))?; + // Now the work directory has been created, we have to clean up. + self.did_cleanup.store(false, Ordering::Release); + // Fetch and decode the Command and download the input tree into + // the work directory, retried together after a store failure. + self.fetch_with_retry(filesystem_store_pin, download_deadline) + .await + .map_err(tag_missing_input)? }; { // Create all directories needed for our output paths. This is required by the bazel spec. @@ -3824,12 +3997,25 @@ impl RunningAction for RunningActionImpl { let operation_id = self.operation_id.clone(); let metrics = self.metrics().clone(); let mut kill_rx = self.kill_token.subscribe(); + // The retry loop inside stops short of this; the timeout below is + // the backstop for an attempt that never answers. The retries are + // also bounded by the action's own timeout: the scheduler ends the + // action soon after that passes, whatever the worker is still + // fetching. A window too far out to represent leaves the retries + // unbounded, as the timeout is. + let action_timeout = self.action_info.timeout; + let retry_window = if action_timeout.is_zero() { + download_timeout + } else { + download_timeout.min(action_timeout) + }; + let download_deadline = Instant::now().checked_add(retry_window); // Boxed: the prepare state machine is large, and pinning it on the // stack next to the select below overflowed a test thread. let mut prepare_fut = Box::pin( metrics .prepare_action - .wrap(Self::inner_prepare_action(self)), + .wrap(Self::inner_prepare_action(self, download_deadline)), ); // A fetch that never answers would otherwise hold the slot for good // while the worker's keepalives say it is fine: name it every minute @@ -5290,4 +5476,122 @@ mod tests { input_fetch_pause(1, &test_digest(7)) ); } + + #[test] + fn prepare_retry_pause_doubles_and_stays_under_the_cap() { + let cap = Duration::from_millis(PREPARE_RETRY_PAUSE_CAP_MS); + for first_byte in [0u8, 1, 19, 127, 255] { + let digest = test_digest(first_byte); + for attempt in 1..=10u32 { + let base = Duration::from_millis( + (PREPARE_RETRY_PAUSE_MS << attempt.saturating_sub(1).min(5)) + .min(PREPARE_RETRY_PAUSE_CAP_MS), + ); + let pause = prepare_retry_pause(attempt, &digest); + assert!(pause <= base, "{pause:?} for attempt {attempt}"); + assert!(pause > base * 3 / 4, "{pause:?} for attempt {attempt}"); + assert!(pause <= cap, "{pause:?} for attempt {attempt}"); + } + } + assert_eq!( + prepare_retry_pause(1, &test_digest(0)), + Duration::from_secs(1) + ); + assert_eq!( + prepare_retry_pause(2, &test_digest(0)), + Duration::from_secs(2) + ); + assert_eq!( + prepare_retry_pause(3, &test_digest(0)), + Duration::from_secs(4) + ); + assert_eq!(prepare_retry_pause(9, &test_digest(0)), cap); + assert_ne!( + prepare_retry_pause(1, &test_digest(3)), + prepare_retry_pause(1, &test_digest(7)) + ); + } + + #[test] + fn transient_fetch_errors_are_the_retrier_codes() { + for code in [ + Code::Unavailable, + Code::ResourceExhausted, + Code::Aborted, + Code::Internal, + Code::Unknown, + Code::DeadlineExceeded, + Code::DataLoss, + ] { + let err = make_err!(code, "store answered 503 ServerBusy"); + assert!(is_transient_fetch_error(&err), "{code:?}"); + } + for code in [ + Code::NotFound, + Code::InvalidArgument, + Code::FailedPrecondition, + Code::PermissionDenied, + Code::Unauthenticated, + Code::Unimplemented, + Code::OutOfRange, + Code::AlreadyExists, + Code::Cancelled, + ] { + let err = make_err!(code, "store answered"); + assert!(!is_transient_fetch_error(&err), "{code:?}"); + } + } + + /// The worker's own disk failing maps to the same codes a store + /// failure does, but is never retried on this worker. The error + /// numbers are Linux's (ENOSPC, EPIPE), so that part runs on unix. + #[test] + fn an_operating_system_error_is_not_a_transient_fetch_error() { + #[cfg(target_family = "unix")] + { + let disk_full = Error::from(std::io::Error::from_raw_os_error(28)); + assert_eq!(disk_full.code, Code::Unknown); + assert!(!is_transient_fetch_error(&disk_full), "{disk_full:?}"); + let broken_pipe = + Error::from(std::io::Error::from_raw_os_error(32)).append("Writing an input"); + assert_eq!(broken_pipe.code, Code::Internal); + assert!(!is_transient_fetch_error(&broken_pipe), "{broken_pipe:?}"); + } + // What the standard library prints for any OS error, whatever the + // platform's number for it. + let os_error = Error::from(std::io::Error::other( + "No space left on device (os error 28)", + )); + assert_eq!(os_error.code, Code::Internal); + assert!(!is_transient_fetch_error(&os_error), "{os_error:?}"); + // A store that reset the connection is not this worker's disk. + let reset = Error::from(std::io::Error::from(std::io::ErrorKind::ConnectionReset)); + assert_eq!(reset.code, Code::Unavailable); + assert!(is_transient_fetch_error(&reset), "{reset:?}"); + } + + /// Every retried fetch settles on one terminal outcome, dropped ones + /// included; a fetch never retried records nothing. + #[test] + fn prepare_retries_settle_once() { + let recorded = std::cell::RefCell::new(Vec::new()); + let record = |outcome: &'static str| recorded.borrow_mut().push(outcome); + + let mut retries = PrepareRetries::new(record); + retries.retried(); + retries.retried(); + retries.settle("recovered"); + assert_eq!(*recorded.borrow(), vec!["retried", "retried", "recovered"]); + + recorded.borrow_mut().clear(); + PrepareRetries::new(record).settle("failed"); + assert!(recorded.borrow().is_empty(), "no retry, nothing to settle"); + + recorded.borrow_mut().clear(); + { + let mut retries = PrepareRetries::new(record); + retries.retried(); + } + assert_eq!(*recorded.borrow(), vec!["retried", "abandoned"]); + } } diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index 2aafc0daba..66c721b801 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -46,6 +46,8 @@ mod tests { use nativelink_config::stores::{ EvictionPolicy, FastSlowSpec, FilesystemSpec, MemorySpec, StoreDirection, StoreSpec, }; + #[cfg(target_family = "unix")] + use nativelink_error::make_err; use nativelink_error::{Code, Error, ResultExt, make_input_err}; use nativelink_macro::nativelink_test; #[cfg(target_family = "unix")] @@ -3952,6 +3954,122 @@ exit 0 #[cfg(target_family = "unix")] default_health_status_indicator!(HangingStore); + /// A store that refuses the first `failures` reads of one digest with a + /// chosen code and serves it after that: a slow tier whose own retrier + /// gave up, as the worker sees it. + #[cfg(target_family = "unix")] + #[derive(Debug)] + struct FlakyStore { + inner: Store, + fail_on: DigestInfo, + failures: u64, + code: Code, + reads: AtomicU64, + } + + #[cfg(target_family = "unix")] + impl FlakyStore { + fn new( + inner: Arc, + fail_on: DigestInfo, + failures: u64, + code: Code, + ) -> Arc { + Arc::new(Self { + inner: Store::new(inner), + fail_on, + failures, + code, + reads: AtomicU64::new(0), + }) + } + + /// Reads of the flaky digest so far, refused or served. + fn reads(&self) -> u64 { + self.reads.load(Ordering::SeqCst) + } + } + + #[cfg(target_family = "unix")] + impl MetricsComponent for FlakyStore { + fn publish( + &self, + _kind: MetricKind, + _field_metadata: MetricFieldData, + ) -> Result { + Ok(MetricPublishKnownKindData::Component) + } + } + + #[cfg(target_family = "unix")] + #[async_trait] + impl StoreDriver for FlakyStore { + async fn post_init(self: Arc) -> Result<(), Error> { + Ok(()) + } + async fn has_with_results( + self: Pin<&Self>, + keys: &[StoreKey<'_>], + results: &mut [Option], + ) -> Result<(), Error> { + self.inner + .as_store_driver_pin() + .has_with_results(keys, results) + .await + } + async fn update( + self: Pin<&Self>, + key: StoreKey<'_>, + reader: DropCloserReadHalf, + size_info: UploadSizeInfo, + ) -> Result { + self.inner + .as_store_driver_pin() + .update(key, reader, size_info) + .await + } + async fn get_part( + self: Pin<&Self>, + key: StoreKey<'_>, + writer: &mut DropCloserWriteHalf, + offset: u64, + length: Option, + ) -> Result<(), Error> { + if key == StoreKey::Digest(self.fail_on) { + let read = self.reads.fetch_add(1, Ordering::SeqCst) + 1; + if read <= self.failures { + return Err(make_err!(self.code, "flaky store refused read {read}")); + } + } + self.inner + .as_store_driver_pin() + .get_part(key, writer, offset, length) + .await + } + fn inner_store(&self, _key: Option) -> &dyn StoreDriver { + self + } + fn as_any<'a>(&'a self) -> &'a (dyn core::any::Any + Sync + Send + 'static) { + self + } + fn as_any_arc(self: Arc) -> Arc { + self + } + fn register_remove_callback( + self: Arc, + _callback: Arc, + ) -> Result<(), Error> { + Ok(()) + } + } + + #[cfg(target_family = "unix")] + default_health_status_indicator!(FlakyStore); + + /// The one input file of the actions `fetch_setup` builds. + #[cfg(target_family = "unix")] + const FETCH_SETUP_FILE_CONTENT: &[u8] = b"never arrives"; + /// A manager whose CAS never finishes reading one input of the action it /// returns the digest of; the store comes back too, so a test can lift /// that or make writes hang instead. @@ -3966,14 +4084,48 @@ exit 0 ), Box, > { - const FILE_CONTENT: &[u8] = b"never arrives"; + fetch_setup(max_download_timeout, HangingStore::new).await + } + + /// A manager whose CAS reads one input of the action it returns the + /// digest of through the store `make_slow` wraps around the memory store + /// holding everything; the wrapper comes back too. + #[cfg(target_family = "unix")] + async fn fetch_setup( + max_download_timeout: Duration, + make_slow: impl FnOnce(Arc, DigestInfo) -> Arc, + ) -> Result<(Arc, DigestInfo, Arc), Box> + { + fetch_setup_tree(max_download_timeout, None, false, make_slow).await + } + + /// The second file of a nested `fetch_setup_tree`, at the root. + #[cfg(target_family = "unix")] + const FETCH_SETUP_OTHER_CONTENT: &[u8] = b"other input\n"; + + /// `fetch_setup` with the action's own timeout set and, when `nested`, + /// the flaky file under a subdirectory beside a second file at the + /// root, so that a failed attempt leaves a partly built tree behind. + #[cfg(target_family = "unix")] + async fn fetch_setup_tree( + max_download_timeout: Duration, + action_timeout: Option, + nested: bool, + make_slow: impl FnOnce(Arc, DigestInfo) -> Arc, + ) -> Result<(Arc, DigestInfo, Arc), Box> + { let inner = MemoryStore::new(&MemorySpec::default()); - let file_digest = DigestInfo::new([9u8; 32], FILE_CONTENT.len() as u64); + let file_digest = DigestInfo::new([9u8; 32], FETCH_SETUP_FILE_CONTENT.len() as u64); inner .as_ref() - .update_oneshot(file_digest, FILE_CONTENT.into()) + .update_oneshot(file_digest, FETCH_SETUP_FILE_CONTENT.into()) + .await?; + let other_digest = DigestInfo::new([8u8; 32], FETCH_SETUP_OTHER_CONTENT.len() as u64); + inner + .as_ref() + .update_oneshot(other_digest, FETCH_SETUP_OTHER_CONTENT.into()) .await?; - let slow_store = HangingStore::new(inner.clone(), file_digest); + let slow_store = make_slow(inner.clone(), file_digest); let fast_config = FilesystemSpec { content_path: make_temp_path("content_path"), temp_path: make_temp_path("temp_path"), @@ -3992,15 +4144,41 @@ exit 0 Store::new(fast_store), Store::new(slow_store.clone()), ); - let input_root_digest = serialize_and_upload_message( - &Directory { + let flaky_file = FileNode { + name: "input.txt".to_string(), + digest: Some(file_digest.into()), + ..Default::default() + }; + let root = if nested { + let sub_digest = serialize_and_upload_message( + &Directory { + files: vec![flaky_file], + ..Default::default() + }, + inner.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + Directory { files: vec![FileNode { - name: "input.txt".to_string(), - digest: Some(file_digest.into()), + name: "other.txt".to_string(), + digest: Some(other_digest.into()), ..Default::default() }], + directories: vec![DirectoryNode { + name: "sub".to_string(), + digest: Some(sub_digest.into()), + }], ..Default::default() - }, + } + } else { + Directory { + files: vec![flaky_file], + ..Default::default() + } + }; + let input_root_digest = serialize_and_upload_message( + &root, inner.as_pin(), &mut DigestHasherFunc::Sha256.hasher(), ) @@ -4018,6 +4196,10 @@ exit 0 &Action { command_digest: Some(command_digest.into()), input_root_digest: Some(input_root_digest.into()), + timeout: action_timeout.map(|timeout| prost_types::Duration { + seconds: i64::try_from(timeout.as_secs()).unwrap_or(i64::MAX), + nanos: 0, + }), ..Default::default() }, inner.as_pin(), @@ -4150,6 +4332,253 @@ exit 0 Ok(()) } + /// An input read that fails with a transient code is fetched again from + /// this worker, after 1 s then 2 s, and the action goes on once the + /// store answers; the retries are the worker's own, the scheduler sees + /// one action executing within its ceilings. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn prepare_action_retries_a_transient_fetch_failure() + -> Result<(), Box> { + let (running_actions_manager, action_digest, store) = + fetch_setup(Duration::from_mins(10), |inner, digest| { + FlakyStore::new(inner, digest, 2, Code::Unavailable) + }) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let started = tokio::time::Instant::now(); + let action = action.prepare_action().await?; + let elapsed = started.elapsed(); + assert_eq!(store.reads(), 3, "two refused reads, then the one served"); + // Two backoffs of 1 s and 2 s, each shortened by up to a quarter. + assert!(elapsed > Duration::from_secs(2), "{elapsed:?}"); + assert!( + elapsed <= Duration::from_secs(3) + Duration::from_millis(10), + "{elapsed:?}" + ); + let staged = std::fs::read(format!("{}/input.txt", action.get_work_directory()))?; + assert_eq!(staged, FETCH_SETUP_FILE_CONTENT); + action.cleanup().await?; + tokio::time::timeout(Duration::from_secs(5), running_actions_manager.kill_all()) + .await + .expect("an entry was left behind"); + Ok(()) + } + + /// A first attempt that fails after part of the tree is built (the + /// subdirectory exists, the file in it does not) is cleared before the + /// next, which stages the whole tree. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn prepare_action_restages_a_partly_built_tree() -> Result<(), Box> + { + let (running_actions_manager, action_digest, store) = + fetch_setup_tree(Duration::from_mins(10), None, true, |inner, digest| { + FlakyStore::new(inner, digest, 1, Code::Unavailable) + }) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let action = action.prepare_action().await?; + assert_eq!(store.reads(), 2, "one refused read, then the one served"); + let work_directory = action.get_work_directory(); + assert_eq!( + std::fs::read(format!("{work_directory}/sub/input.txt"))?, + FETCH_SETUP_FILE_CONTENT + ); + assert_eq!( + std::fs::read(format!("{work_directory}/other.txt"))?, + FETCH_SETUP_OTHER_CONTENT + ); + action.cleanup().await?; + tokio::time::timeout(Duration::from_secs(5), running_actions_manager.kill_all()) + .await + .expect("an entry was left behind"); + Ok(()) + } + + /// A kill that lands during the pause before the next attempt ends the + /// prepare at once with `Aborted`, not at the end of the pause. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn a_kill_during_the_retry_pause_stops_prepare() -> Result<(), Box> + { + let (running_actions_manager, action_digest, store) = + fetch_setup(Duration::from_mins(10), |inner, digest| { + FlakyStore::new(inner, digest, u64::MAX, Code::Unavailable) + }) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let action_directory = action.get_work_directory().clone(); + let started = tokio::time::Instant::now(); + // Attempts at 0 s and ~1 s; the 2 s pause after the second runs + // until ~3 s, and the kill lands in the middle of it. + let killer = { + let manager = running_actions_manager.clone(); + let operation_id = operation_id.clone(); + async move { + tokio::time::sleep(Duration::from_millis(1500)).await; + manager.kill_operation(&operation_id).await + } + }; + let (kill_res, prepare_res) = tokio::join!(killer, action.clone().prepare_action()); + kill_res?; + let err = prepare_res.expect_err("a killed fetch must not complete"); + assert_eq!(err.code, Code::Aborted, "{err:?}"); + assert_eq!(store.reads(), 2, "{err:?}"); + let elapsed = started.elapsed(); + assert!( + elapsed < Duration::from_millis(2500), + "the kill ended the pause early: {elapsed:?}" + ); + action.cleanup().await?; + assert!( + !Path::new(&action_directory).exists(), + "the work directory is gone after cleanup" + ); + Ok(()) + } + + /// A missing input is not a store hiccup: it fails prepare at once, and + /// the store is asked exactly once. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn prepare_action_does_not_retry_a_missing_input() + -> Result<(), Box> { + let (running_actions_manager, action_digest, store) = + fetch_setup(Duration::from_mins(10), |inner, digest| { + FlakyStore::new(inner, digest, u64::MAX, Code::NotFound) + }) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let started = tokio::time::Instant::now(); + let err = action + .clone() + .prepare_action() + .await + .expect_err("a missing input must fail prepare"); + assert_eq!(err.code, Code::NotFound, "{err:?}"); + assert_eq!(store.reads(), 1, "no retry of a missing input"); + assert!( + started.elapsed() < Duration::from_secs(1), + "{:?}", + started.elapsed() + ); + action.cleanup().await?; + tokio::time::timeout(Duration::from_secs(5), running_actions_manager.kill_all()) + .await + .expect("an entry was left behind"); + Ok(()) + } + + /// A store that never recovers is retried only while the download + /// deadline leaves room for another attempt; then the last error comes + /// back, naming the attempts, before the deadline itself fires. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn prepare_action_stops_retrying_at_the_download_deadline() + -> Result<(), Box> { + let (running_actions_manager, action_digest, store) = + fetch_setup(Duration::from_secs(5), |inner, digest| { + FlakyStore::new(inner, digest, u64::MAX, Code::Unavailable) + }) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let action_directory = action.get_work_directory().clone(); + let started = tokio::time::Instant::now(); + let err = action + .clone() + .prepare_action() + .await + .expect_err("the store never recovers, prepare must fail"); + let elapsed = started.elapsed(); + // Attempts at 0 s, ~1 s and ~3 s; the 4 s pause after the third + // would end past the 5 s deadline. + assert_eq!(store.reads(), 3, "{err:?}"); + assert_ne!(err.code, Code::DeadlineExceeded, "{err:?}"); + assert!(err.to_string().contains("failed 3 time(s)"), "{err:?}"); + assert!(elapsed < Duration::from_secs(5), "{elapsed:?}"); + action.cleanup().await?; + assert!( + !Path::new(&action_directory).exists(), + "the work directory is gone after cleanup" + ); + tokio::time::timeout(Duration::from_secs(5), running_actions_manager.kill_all()) + .await + .expect("an entry was left behind"); + Ok(()) + } + + /// An action with its own timeout is not retried past it: the scheduler + /// ends the action soon after that passes, so a window of ten minutes + /// shrinks to the four seconds the action may take. + #[cfg(target_family = "unix")] + #[nativelink_test(start_paused = true)] + async fn prepare_action_stops_retrying_at_the_action_timeout() + -> Result<(), Box> { + let (running_actions_manager, action_digest, store) = fetch_setup_tree( + Duration::from_mins(10), + Some(Duration::from_secs(4)), + false, + |inner, digest| FlakyStore::new(inner, digest, u64::MAX, Code::Unavailable), + ) + .await?; + let operation_id = OperationId::default(); + let action = running_actions_manager + .create_and_add_action( + "test-worker".to_string(), + start_execute_for(action_digest, &operation_id.to_string()), + ) + .await?; + let started = tokio::time::Instant::now(); + let err = action + .clone() + .prepare_action() + .await + .expect_err("the store never recovers, prepare must fail"); + let elapsed = started.elapsed(); + // Attempts at 0 s, under 1 s and under 3 s (each pause is shortened + // by up to a quarter); the pause after the third is at least 3 s + // and would end past the 4 s the action is allowed. + assert_eq!(store.reads(), 3, "{err:?}"); + assert!(err.to_string().contains("failed 3 time(s)"), "{err:?}"); + assert!(elapsed < Duration::from_secs(4), "{elapsed:?}"); + action.cleanup().await?; + tokio::time::timeout(Duration::from_secs(5), running_actions_manager.kill_all()) + .await + .expect("an entry was left behind"); + Ok(()) + } + /// A kill that arrives while the results upload ends it at once with /// `Aborted`: a kill is acknowledged in every phase, not only while the /// command runs. diff --git a/web/apps/docs/content/docs/reference/metrics.mdx b/web/apps/docs/content/docs/reference/metrics.mdx index 5030b9dd61..4b5b19ea27 100644 --- a/web/apps/docs/content/docs/reference/metrics.mdx +++ b/web/apps/docs/content/docs/reference/metrics.mdx @@ -5,10 +5,10 @@ full: true --- {/* AUTOGENERATED. Do not edit by hand. - Source: nativelink-util/src/metrics.rs @ v1.7.6 (d06d9c9b4) + Source: nativelink-util/src/metrics.rs @ v1.7.6 (d0e6aa8e3) Regenerate from web/: bun --filter @nativelink/docs gen:metrics-reference */} -NativeLink **v1.7.6** declares 46 OpenTelemetry instruments. This page is +NativeLink **v1.7.6** declares 47 OpenTelemetry instruments. This page is generated from the declarations in [`nativelink-util/src/metrics.rs`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs) and from every `.add()` and `.record()` call site across the workspace, so the @@ -44,6 +44,7 @@ wrong. | [`worker.connections`](#worker-connections) | `nativelink_worker_connections` (unreferenced) | Counter | Conditionally | scheduler process only | | [`worker.disconnections`](#worker-disconnections) | `nativelink_worker_disconnections` (unreferenced) | Counter | Conditionally | scheduler process only | | [`worker.input.relinks`](#worker-input-relinks) | `nativelink_worker_input_relinks` (unreferenced) | Counter | Yes | whenever the code path runs | +| [`worker.prepare.retries`](#worker-prepare-retries) | `nativelink_worker_prepare_retries` (unreferenced) | Counter | Yes | whenever the code path runs | | [`scheduler.dispatch.requeues`](#scheduler-dispatch-requeues) | `nativelink_scheduler_dispatch_requeues` (unreferenced) | Counter | Conditionally | scheduler process only | | [`worker.keepalives`](#worker-keepalives) | `nativelink_worker_keepalives` (unreferenced) | Counter | Conditionally | scheduler process only | | [`worker.keepalive.gap`](#worker-keepalive-gap) | `nativelink_worker_keepalive_gap` (unreferenced) | Histogram | Conditionally | scheduler process only | @@ -438,7 +439,21 @@ Input hardlinks made again after their source was evicted mid-staging, by kind. | Prometheus series | `nativelink_worker_input_relinks` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | | Declared at | [`nativelink-util/src/metrics.rs:903`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L903) | -| Emitted from | [`nativelink-worker/src/running_actions_manager.rs:944`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L944), [`nativelink-worker/src/running_actions_manager.rs:1078`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1078), [`nativelink-worker/src/running_actions_manager.rs:1096`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1096) | +| Emitted from | [`nativelink-worker/src/running_actions_manager.rs:1038`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1038), [`nativelink-worker/src/running_actions_manager.rs:1172`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1172), [`nativelink-worker/src/running_actions_manager.rs:1190`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1190) | + +### `worker.prepare.retries` + +Input fetches tried again on the worker after a transient store failure: `retried` per attempt made again, then one of `recovered`, `exhausted`, `failed` or `abandoned`. + +| Property | Value | +| --- | --- | +| Instrument type | `Counter` | +| Field | `prepare_retries` | +| Unit | — | +| Prometheus series | `nativelink_worker_prepare_retries` (not referenced by any shipped rule or dashboard) | +| Emitted | Yes. whenever the code path runs | +| Declared at | [`nativelink-util/src/metrics.rs:908`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L908) | +| Emitted from | [`nativelink-worker/src/running_actions_manager.rs:1014`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-worker/src/running_actions_manager.rs#L1014) | ### `scheduler.dispatch.requeues` @@ -451,7 +466,7 @@ Dispatches requeued untried, by reason. | Unit | — | | Prometheus series | `nativelink_scheduler_dispatch_requeues` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:908`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L908) | +| Declared at | [`nativelink-util/src/metrics.rs:913`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L913) | | Emitted from | [`nativelink-scheduler/src/api_worker_scheduler.rs:1187`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1187), [`nativelink-scheduler/src/api_worker_scheduler.rs:1381`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1381), [`nativelink-scheduler/src/api_worker_scheduler.rs:1701`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1701), [`nativelink-scheduler/src/api_worker_scheduler.rs:1999`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1999), [`nativelink-scheduler/src/api_worker_scheduler.rs:2030`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L2030), [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:679`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L679) | ### `worker.keepalives` @@ -465,7 +480,7 @@ Keepalives received, cumulative. | Unit | `{keepalive}` | | Prometheus series | `nativelink_worker_keepalives` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:913`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L913) | +| Declared at | [`nativelink-util/src/metrics.rs:918`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L918) | | Emitted from | [`nativelink-scheduler/src/api_worker_scheduler.rs:422`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L422) | ### `worker.keepalive.gap` @@ -479,7 +494,7 @@ Seconds between consecutive messages from a worker. | Unit | `s` | | Prometheus series | `nativelink_worker_keepalive_gap` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:919`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L919) | +| Declared at | [`nativelink-util/src/metrics.rs:924`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L924) | | Emitted from | [`nativelink-scheduler/src/api_worker_scheduler.rs:356`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L356) | **Histogram buckets:** 12 explicit boundaries, from @@ -498,7 +513,7 @@ Connected workers currently paused or draining. | Unit | `{worker}` | | Prometheus series | `nativelink_worker_state_count` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:930`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L930) | +| Declared at | [`nativelink-util/src/metrics.rs:935`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L935) | | Emitted from | [`nativelink-scheduler/src/api_worker_scheduler.rs:405`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L405), [`nativelink-scheduler/src/api_worker_scheduler.rs:458`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L458), [`nativelink-scheduler/src/api_worker_scheduler.rs:461`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L461), [`nativelink-scheduler/src/api_worker_scheduler.rs:571`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L571), [`nativelink-scheduler/src/api_worker_scheduler.rs:1148`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1148), [`nativelink-scheduler/src/api_worker_scheduler.rs:1190`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1190), [`nativelink-scheduler/src/api_worker_scheduler.rs:1326`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/api_worker_scheduler.rs#L1326) | ## gRPC server metrics @@ -508,7 +523,7 @@ Global gRPC serving metrics. One duration histogram covers rate, errors and latency: the count gives rate, the status attribute separates errors, and the buckets give latency. -Declared as `RPC_METRICS` in [`nativelink-util/src/metrics.rs:1041`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1041), backed by the `RpcMetrics` struct. +Declared as `RPC_METRICS` in [`nativelink-util/src/metrics.rs:1059`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1059), backed by the `RpcMetrics` struct. Emitted by the `OtlpLayer` tower middleware on every gRPC server NativeLink starts, so a CAS, a scheduler and a worker API endpoint each report their own. @@ -523,7 +538,7 @@ Duration of inbound gRPC calls, by service, method and status. | Unit | `s` | | Prometheus series | `nativelink_rpc_server_duration` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1045`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1045) | +| Declared at | [`nativelink-util/src/metrics.rs:1063`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1063) | | Emitted from | [`nativelink-util/src/telemetry.rs:371`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/telemetry.rs#L371) | **Histogram buckets:** 15 explicit boundaries, from @@ -535,7 +550,7 @@ measurement. Global scheduler metrics. -Declared as `SCHEDULER_METRICS` in [`nativelink-util/src/metrics.rs:1094`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1094), backed by the `SchedulerOtlpMetrics` struct. +Declared as `SCHEDULER_METRICS` in [`nativelink-util/src/metrics.rs:1112`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1112), backed by the `SchedulerOtlpMetrics` struct. Emitted from the scheduler process once per matching pass, whether or not the pass assigned anything. @@ -551,7 +566,7 @@ while the queue is non-empty means matching is the bottleneck. | Unit | `s` | | Prometheus series | `nativelink_scheduler_matching_duration` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1098`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1098) | +| Declared at | [`nativelink-util/src/metrics.rs:1116`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1116) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:458`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L458) | **Histogram buckets:** 13 explicit boundaries, from @@ -570,7 +585,7 @@ Matching passes, by result. | Unit | `{pass}` | | Prometheus series | `nativelink_scheduler_matching_passes` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1107`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1107) | +| Declared at | [`nativelink-util/src/metrics.rs:1125`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1125) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:458`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L458) | ### `scheduler.unsatisfiable.queued` @@ -586,7 +601,7 @@ instance, judged against that instance's own workers. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_unsatisfiable_queued` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1113`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1113) | +| Declared at | [`nativelink-util/src/metrics.rs:1131`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1131) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1036`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1036) | ### `scheduler.unsatisfiable.failed` @@ -601,7 +616,7 @@ the properties that could not be satisfied. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_unsatisfiable_failed` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1121`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1121) | +| Declared at | [`nativelink-util/src/metrics.rs:1139`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1139) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:560`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L560) | ### `scheduler.queue.depth` @@ -615,7 +630,7 @@ Queued actions the last matching pass read. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_queue_depth` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1127`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1127) | +| Declared at | [`nativelink-util/src/metrics.rs:1145`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1145) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1037`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1037) | ### `scheduler.queue.age` @@ -629,7 +644,7 @@ Queued actions the last matching pass read, by age bucket. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_queue_age` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1133`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1133) | +| Declared at | [`nativelink-util/src/metrics.rs:1151`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1151) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1038`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1038) | ### `scheduler.queue.oldest_age` @@ -643,7 +658,7 @@ Age of the oldest action the last matching pass read, in seconds. | Unit | `s` | | Prometheus series | `nativelink_scheduler_queue_oldest_age` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1141`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1141) | +| Declared at | [`nativelink-util/src/metrics.rs:1159`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1159) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1038`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1038) | ### `scheduler.executing.age` @@ -657,7 +672,7 @@ Executing actions the last executing sweep read, by age bucket. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_executing_age` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1149`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1149) | +| Declared at | [`nativelink-util/src/metrics.rs:1167`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1167) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1307`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1307) | ### `scheduler.executing.oldest_age` @@ -672,7 +687,7 @@ executing, in seconds. | Unit | `s` | | Prometheus series | `nativelink_scheduler_executing_oldest_age` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1157`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1157) | +| Declared at | [`nativelink-util/src/metrics.rs:1175`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1175) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1307`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1307) | ### `scheduler.queue.retired` @@ -686,7 +701,7 @@ Queued actions retired for having no client. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_queue_retired` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1165`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1165) | +| Declared at | [`nativelink-util/src/metrics.rs:1183`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1183) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:783`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L783) | ### `scheduler.matching.parked_dispatched` @@ -701,7 +716,7 @@ opened before the pass ended, ahead of the newer actions it was reading. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_matching_parked_dispatched` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1171`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1171) | +| Declared at | [`nativelink-util/src/metrics.rs:1189`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1189) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1013`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1013) | ### `scheduler.awaited_action.orphans` @@ -715,7 +730,7 @@ Listed operations whose record was gone on read, by site. | Unit | `{operation}` | | Prometheus series | `nativelink_scheduler_awaited_action_orphans` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1179`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1179) | +| Declared at | [`nativelink-util/src/metrics.rs:1197`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1197) | | Emitted from | [`nativelink-scheduler/src/admin.rs:107`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/admin.rs#L107), [`nativelink-scheduler/src/simple_scheduler.rs:600`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L600), [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:614`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L614), [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:731`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L731), [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:1662`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L1662), [`nativelink-scheduler/src/simple_scheduler_state_manager.rs:1685`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler_state_manager.rs#L1685) | ### `scheduler.sweep.failures` @@ -729,7 +744,7 @@ Abandoned-queue sweeps that failed before finishing. | Unit | `{sweep}` | | Prometheus series | `nativelink_scheduler_sweep_failures` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1187`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1187) | +| Declared at | [`nativelink-util/src/metrics.rs:1205`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1205) | | Emitted from | [`nativelink-scheduler/src/simple_scheduler.rs:1259`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1259), [`nativelink-scheduler/src/simple_scheduler.rs:1318`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/simple_scheduler.rs#L1318) | ### `scheduler.hints.resolutions` @@ -743,14 +758,14 @@ Actions sized by the historical resource scheduler, by source. | Unit | `{action}` | | Prometheus series | `nativelink_scheduler_hints_resolutions` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. scheduler process only | -| Declared at | [`nativelink-util/src/metrics.rs:1193`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1193) | +| Declared at | [`nativelink-util/src/metrics.rs:1211`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1211) | | Emitted from | [`nativelink-scheduler/src/historical_resource_scheduler.rs:643`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/historical_resource_scheduler.rs#L643), [`nativelink-scheduler/src/historical_resource_scheduler.rs:647`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/historical_resource_scheduler.rs#L647), [`nativelink-scheduler/src/historical_resource_scheduler.rs:650`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-scheduler/src/historical_resource_scheduler.rs#L650) | ## STORE_METRICS Global object-store metrics. -Declared as `STORE_METRICS` in [`nativelink-util/src/metrics.rs:1409`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1409), backed by the `StoreMetrics` struct. +Declared as `STORE_METRICS` in [`nativelink-util/src/metrics.rs:1427`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1427), backed by the `StoreMetrics` struct. ### `store.object.throttles` @@ -763,14 +778,14 @@ Throttled responses, by provider and operation. | Unit | `{response}` | | Prometheus series | `nativelink_store_object_throttles` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1413`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1413) | -| Emitted from | [`nativelink-store/src/azure_blob_store.rs:243`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/azure_blob_store.rs#L243) | +| Declared at | [`nativelink-util/src/metrics.rs:1431`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1431) | +| Emitted from | [`nativelink-store/src/azure_blob_store.rs:268`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/azure_blob_store.rs#L268) | ## Store tier metrics Global tiered-store metrics. -Declared as `STORE_TIER_METRICS` in [`nativelink-util/src/metrics.rs:1437`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1437), backed by the `StoreTierMetrics` struct. +Declared as `STORE_TIER_METRICS` in [`nativelink-util/src/metrics.rs:1455`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1455), backed by the `StoreTierMetrics` struct. Emitted from the `fast_slow` store wrapper only. A deployment without a `fast_slow` store emits none of these. @@ -785,7 +800,7 @@ Reads served per tier, by result. | Unit | `{operation}` | | Prometheus series | `nativelink_store_tier_operations` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. only through a `fast_slow` store | -| Declared at | [`nativelink-util/src/metrics.rs:1441`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1441) | +| Declared at | [`nativelink-util/src/metrics.rs:1459`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1459) | | Emitted from | [`nativelink-store/src/fast_slow_store.rs:348`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L348), [`nativelink-store/src/fast_slow_store.rs:839`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L839), [`nativelink-store/src/fast_slow_store.rs:855`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L855), [`nativelink-store/src/fast_slow_store.rs:865`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L865), [`nativelink-store/src/fast_slow_store.rs:882`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L882), [`nativelink-store/src/fast_slow_store.rs:912`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L912), [`nativelink-store/src/fast_slow_store.rs:934`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L934) | ### `store.tier.io` @@ -799,14 +814,14 @@ Bytes moved per tier and direction. | Unit | `By` | | Prometheus series | `nativelink_store_tier_io` (not referenced by any shipped rule or dashboard) | | Emitted | Conditionally. only through a `fast_slow` store | -| Declared at | [`nativelink-util/src/metrics.rs:1447`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1447) | +| Declared at | [`nativelink-util/src/metrics.rs:1465`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1465) | | Emitted from | [`nativelink-store/src/fast_slow_store.rs:357`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L357), [`nativelink-store/src/fast_slow_store.rs:856`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L856), [`nativelink-store/src/fast_slow_store.rs:913`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L913), [`nativelink-store/src/fast_slow_store.rs:935`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/fast_slow_store.rs#L935) | ## Health check metrics Global health-check metrics. -Declared as `HEALTH_METRICS` in [`nativelink-util/src/metrics.rs:1488`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1488), backed by the `HealthMetrics` struct. +Declared as `HEALTH_METRICS` in [`nativelink-util/src/metrics.rs:1506`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1506), backed by the `HealthMetrics` struct. Emitted each time the health check endpoint runs, one observation per registered health indicator. @@ -821,14 +836,14 @@ Health check results, by namespace and status. | Unit | `{check}` | | Prometheus series | `nativelink_health_checks` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1492`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1492) | +| Declared at | [`nativelink-util/src/metrics.rs:1510`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1510) | | Emitted from | [`nativelink-util/src/health_utils.rs:248`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/health_utils.rs#L248) | ## Connection pool metrics Global connection-pool metrics. -Declared as `CONNECTION_METRICS` in [`nativelink-util/src/metrics.rs:1519`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1519), backed by the `ConnectionMetrics` struct. +Declared as `CONNECTION_METRICS` in [`nativelink-util/src/metrics.rs:1537`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1537), backed by the `ConnectionMetrics` struct. Emitted by the gRPC connection manager and the Redis store as they hand out and re-establish connections. A deployment that uses neither emits none of these. @@ -843,7 +858,7 @@ Free slots at the moment a connection was taken. | Unit | `{connection}` | | Prometheus series | `nativelink_connection_pool_available` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1523`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1523) | +| Declared at | [`nativelink-util/src/metrics.rs:1541`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1541) | | Emitted from | [`nativelink-store/src/redis_store.rs:587`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/redis_store.rs#L587), [`nativelink-util/src/connection_manager.rs:341`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L341), [`nativelink-util/src/connection_manager.rs:351`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L351) | **Histogram buckets:** 11 explicit boundaries, from @@ -862,7 +877,7 @@ Acquisitions, by pool and result. | Unit | `{acquisition}` | | Prometheus series | `nativelink_connection_pool_acquisitions` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1532`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1532) | +| Declared at | [`nativelink-util/src/metrics.rs:1550`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1550) | | Emitted from | [`nativelink-store/src/redis_store.rs:587`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/redis_store.rs#L587), [`nativelink-util/src/connection_manager.rs:341`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L341), [`nativelink-util/src/connection_manager.rs:351`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L351) | ### `connection.reconnects` @@ -876,7 +891,7 @@ Reconnects, by pool. | Unit | `{reconnect}` | | Prometheus series | `nativelink_connection_reconnects` (not referenced by any shipped rule or dashboard) | | Emitted | Yes. whenever the code path runs | -| Declared at | [`nativelink-util/src/metrics.rs:1538`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1538) | +| Declared at | [`nativelink-util/src/metrics.rs:1556`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1556) | | Emitted from | [`nativelink-store/src/redis_store.rs:319`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/redis_store.rs#L319) | ## Attributes