diff --git a/nativelink-scheduler/src/grpc_scheduler.rs b/nativelink-scheduler/src/grpc_scheduler.rs index bb6c74b9c7..8663a5f4fa 100644 --- a/nativelink-scheduler/src/grpc_scheduler.rs +++ b/nativelink-scheduler/src/grpc_scheduler.rs @@ -111,7 +111,8 @@ impl GrpcScheduler { Arc::new(|duration| Box::pin(sleep(duration))), jitter_fn.clone(), spec.retry.clone(), - ), + ) + .with_client("grpc_scheduler"), connection_manager: ConnectionManager::new( core::iter::once(endpoint), spec.connections_per_endpoint, diff --git a/nativelink-store/src/azure_blob_store.rs b/nativelink-store/src/azure_blob_store.rs index 65ca460e15..0de1fd508b 100644 --- a/nativelink-store/src/azure_blob_store.rs +++ b/nativelink-store/src/azure_blob_store.rs @@ -226,6 +226,7 @@ where jitter_fn, spec.common.retry.clone(), ) + .with_client("azure") .with_throttle_budget(throttle_retry_budget), consider_expired_after_s: i64::from(spec.common.consider_expired_after_s), max_retry_buffer_per_request: spec diff --git a/nativelink-store/src/common_s3_utils.rs b/nativelink-store/src/common_s3_utils.rs index 6003e0ffbc..fe86b53ee6 100644 --- a/nativelink-store/src/common_s3_utils.rs +++ b/nativelink-store/src/common_s3_utils.rs @@ -113,7 +113,8 @@ impl TlsClient { Arc::new(|duration| Box::pin(sleep(duration))), common.retry.make_jitter_fn(), common.retry.clone(), - ), + ) + .with_client("s3_http"), } } } diff --git a/nativelink-store/src/gcs_store.rs b/nativelink-store/src/gcs_store.rs index dd9731a5cc..a82945167b 100644 --- a/nativelink-store/src/gcs_store.rs +++ b/nativelink-store/src/gcs_store.rs @@ -136,7 +136,8 @@ where Arc::new(|duration| Box::pin(sleep(duration))), jitter_fn, spec.common.retry.clone(), - ), + ) + .with_client("gcs"), consider_expired_after_s: u64::from(spec.common.consider_expired_after_s), max_retry_buffer_size, max_chunk_size, diff --git a/nativelink-store/src/grpc_store.rs b/nativelink-store/src/grpc_store.rs index e6542cc827..47e4c508ab 100644 --- a/nativelink-store/src/grpc_store.rs +++ b/nativelink-store/src/grpc_store.rs @@ -311,7 +311,8 @@ impl GrpcStore { Arc::new(|duration| Box::pin(sleep(duration))), jitter_fn.clone(), spec.retry.clone(), - ), + ) + .with_client("grpc_store"), connection_manager: ConnectionManager::new( endpoints, spec.connections_per_endpoint, diff --git a/nativelink-store/src/ontap_s3_store.rs b/nativelink-store/src/ontap_s3_store.rs index be7868de68..401c01d17f 100644 --- a/nativelink-store/src/ontap_s3_store.rs +++ b/nativelink-store/src/ontap_s3_store.rs @@ -204,7 +204,8 @@ where Arc::new(|duration| Box::pin(sleep(duration))), jitter_fn, spec.common.retry.clone(), - ), + ) + .with_client("ontap_s3"), consider_expired_after_s: u64::from(spec.common.consider_expired_after_s), max_retry_buffer_per_request: spec .common diff --git a/nativelink-store/src/s3_store.rs b/nativelink-store/src/s3_store.rs index 18d9a57c04..32ede2e16e 100644 --- a/nativelink-store/src/s3_store.rs +++ b/nativelink-store/src/s3_store.rs @@ -201,7 +201,8 @@ where Arc::new(|duration| Box::pin(sleep(duration))), jitter_fn, spec.common.retry.clone(), - ), + ) + .with_client("s3"), consider_expired_after_s: i64::from(spec.common.consider_expired_after_s), max_retry_buffer_per_request: spec .common diff --git a/nativelink-util/src/connection_manager.rs b/nativelink-util/src/connection_manager.rs index 9dfad7bdcf..753e7cb956 100644 --- a/nativelink-util/src/connection_manager.rs +++ b/nativelink-util/src/connection_manager.rs @@ -169,7 +169,8 @@ impl ConnectionManager { Arc::new(|duration| Box::pin(tokio::time::sleep(duration))), jitter_fn, retry, - ), + ) + .with_client("connection"), }; background_spawn!("connection_manager_worker_spawn", async move { worker diff --git a/nativelink-util/src/metrics.rs b/nativelink-util/src/metrics.rs index 52e03ee300..cd6405b157 100644 --- a/nativelink-util/src/metrics.rs +++ b/nativelink-util/src/metrics.rs @@ -1676,3 +1676,43 @@ pub fn record_connection_reconnect(pool: &'static str) { .reconnects .add(1, &[KeyValue::new(CONNECTION_POOL, pool)]); } + +// Metric attribute keys for retried operations. +pub const RETRY_CLIENT: &str = "retry.client"; +pub const RETRY_OUTCOME: &str = "retry.outcome"; + +/// Global metrics for operations run under a [`crate::retry::Retrier`]. +pub static RETRY_METRICS: LazyLock = LazyLock::new(|| { + let meter = global::meter_with_scope(InstrumentationScope::builder("nativelink").build()); + + RetryMetrics { + outcomes: meter + .u64_counter("retry.outcomes") + .with_description( + "Retried operations by client and outcome: each retry (retried), a success after at least one retry (recovered), retries run out (exhausted), or an error that is never retried, NotFound aside (permanent)", + ) + .with_unit("{event}") + .build(), + } +}); + +/// Metrics for operations run under a [`crate::retry::Retrier`]. +#[derive(Debug)] +pub struct RetryMetrics { + /// Retry events, by client and outcome. + pub outcomes: metrics::Counter, +} + +/// Records one retry event. `client` names what is retrying (`azure`, `s3`, +/// `grpc_store`, ...), `outcome` is one of `retried`, `recovered`, +/// `exhausted` or `permanent`. An `exhausted` count rising is a backend that +/// stayed down, or throttled, for longer than the retry policy waits. +pub fn record_retry_outcome(client: &'static str, outcome: &'static str) { + RETRY_METRICS.outcomes.add( + 1, + &[ + KeyValue::new(RETRY_CLIENT, client), + KeyValue::new(RETRY_OUTCOME, outcome), + ], + ); +} diff --git a/nativelink-util/src/retry.rs b/nativelink-util/src/retry.rs index 352025c844..e1a67e5349 100644 --- a/nativelink-util/src/retry.rs +++ b/nativelink-util/src/retry.rs @@ -20,7 +20,9 @@ use futures::future::Future; use futures::stream::StreamExt; use nativelink_config::stores::{ErrorCode, Retry}; use nativelink_error::{Code, Error, make_err}; -use tracing::{error, info}; +use tracing::{error, info, warn}; + +use crate::metrics::record_retry_outcome; struct ExponentialBackoff { current: Duration, @@ -56,12 +58,44 @@ pub enum RetryResult { Err(Error), } +/// What became of an operation, or one attempt of it, under a [`Retrier`]. +/// Recorded as `retry.outcomes`, labeled with the retrier's client. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RetryOutcome { + /// An attempt failed with a retryable error and another follows. A + /// throttle waited out within the throttle budget counts, though it + /// spends no attempt. + Retried, + /// The operation succeeded after at least one retry. + Recovered, + /// The operation failed with a retryable error and the retries ran out. + Exhausted, + /// The operation failed with an error that is never retried. `NotFound` + /// is left out: on a store it is the ordinary miss. + Permanent, +} + +impl RetryOutcome { + const fn as_str(self) -> &'static str { + match self { + Self::Retried => "retried", + Self::Recovered => "recovered", + Self::Exhausted => "exhausted", + Self::Permanent => "permanent", + } + } +} + +type OutcomeFn = Arc; + /// Class used to retry a job with a sleep function in between each retry. #[derive(Clone)] pub struct Retrier { sleep_fn: SleepFn, jitter_fn: JitterFn, config: Retry, + client: &'static str, + outcome_fn: Option, /// How long one `retry` call may spend on `RetryAfter` waits before /// they start spending attempts. Zero means they always do. throttle_budget: Duration, @@ -71,6 +105,7 @@ impl core::fmt::Debug for Retrier { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { f.debug_struct("Retrier") .field("config", &self.config) + .field("client", &self.client) .finish_non_exhaustive() } } @@ -102,10 +137,28 @@ impl Retrier { sleep_fn, jitter_fn, config, + client: "unknown", + outcome_fn: None, throttle_budget: Duration::ZERO, } } + /// Names what this retrier retries for (`azure`, `s3`, `grpc_store`, + /// ...), the `retry.client` label on its `retry.outcomes`. + #[must_use] + pub const fn with_client(mut self, client: &'static str) -> Self { + self.client = client; + self + } + + /// Also hands every outcome to `outcome_fn`, after it is recorded as a + /// metric. For tests, which cannot read the meter. + #[must_use] + pub fn with_outcome_fn(mut self, outcome_fn: OutcomeFn) -> Self { + self.outcome_fn = Some(outcome_fn); + self + } + /// Lets one `retry` call spend up to `budget` waiting on `RetryAfter` /// results without counting those tries as attempts. #[must_use] @@ -114,6 +167,13 @@ impl Retrier { self } + fn report(&self, outcome: RetryOutcome) { + record_retry_outcome(self.client, outcome.as_str()); + if let Some(outcome_fn) = &self.outcome_fn { + outcome_fn(outcome); + } + } + /// This should only return true if the error code should be interpreted as /// temporary. fn should_retry(&self, code: Code) -> bool { @@ -174,7 +234,14 @@ impl Retrier { "Retry stream ended abruptly on attempt {attempt}", )); } - Some(RetryResult::Ok(value)) => return Ok(value), + Some(RetryResult::Ok(value)) => { + // A throttle within the budget is a retry too, + // though it leaves `attempt` where it was. + if attempt > 1 || !throttled.is_zero() { + self.report(RetryOutcome::Recovered); + } + return Ok(value); + } Some(RetryResult::Err(e)) => { return Err(e.append(format!("On attempt {attempt}"))); } @@ -185,6 +252,7 @@ impl Retrier { if err.code == Code::NotFound { info!(?err, "Not found, not retrying"); } else { + self.report(RetryOutcome::Permanent); error!(?attempt, ?err, "Not retrying permanent error"); } return Err(err); @@ -199,12 +267,16 @@ impl Retrier { { throttled = throttled.saturating_add(min_wait); attempt -= 1; + self.report(RetryOutcome::Retried); (self.sleep_fn)(min_wait).await; continue; } - let backoff = iter - .next() - .ok_or_else(|| err.append(format!("On attempt {attempt}")))?; + let Some(backoff) = iter.next() else { + self.report(RetryOutcome::Exhausted); + warn!(client = self.client, ?attempt, ?err, "Retries exhausted"); + return Err(err.append(format!("On attempt {attempt}"))); + }; + self.report(RetryOutcome::Retried); (self.sleep_fn)(backoff.max(min_wait)).await; } } diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index ae12c47c25..daa3d43010 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -21,7 +21,7 @@ use futures::stream::repeat_with; use nativelink_config::stores::Retry; use nativelink_error::{Code, Error, make_err}; use nativelink_macro::nativelink_test; -use nativelink_util::retry::{Retrier, RetryResult}; +use nativelink_util::retry::{Retrier, RetryOutcome, RetryResult}; use pretty_assertions::assert_eq; use tokio::time::Duration; @@ -212,6 +212,100 @@ async fn retry_calls_sleep_fn() -> Result<(), Error> { Ok(()) } +/// A retrier whose outcomes land in the returned list, with `max_retries`. +fn observed_retrier(max_retries: usize) -> (Retrier, Arc>>) { + let outcomes = Arc::new(Mutex::new(Vec::new())); + let sink = outcomes.clone(); + let retrier = Retrier::new( + Arc::new(|_duration| Box::pin(ready(()))), + Arc::new(move |_delay| Duration::from_millis(1)), + Retry { + max_retries, + ..Default::default() + }, + ) + .with_client("test") + .with_outcome_fn(Arc::new(move |outcome| sink.lock().unwrap().push(outcome))); + (retrier, outcomes) +} + +/// Each retry is reported, and a success after them is a recovery; a success +/// on the first attempt reports nothing. +#[nativelink_test] +async fn retry_reports_retries_and_recovery() -> Result<(), Error> { + let (retrier, outcomes) = observed_retrier(5); + let run_count = AtomicI32::new(0); + Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 2 { + RetryResult::Retry(make_err!(Code::Unavailable, "Dummy failure")) + } else { + RetryResult::Ok(true) + } + })) + .await?; + assert_eq!( + *outcomes.lock().unwrap(), + vec![ + RetryOutcome::Retried, + RetryOutcome::Retried, + RetryOutcome::Recovered + ] + ); + + let (retrier, outcomes) = observed_retrier(5); + Pin::new(&retrier) + .retry(repeat_with(|| RetryResult::Ok(true))) + .await?; + assert_eq!(*outcomes.lock().unwrap(), vec![]); + Ok(()) +} + +/// Retries that run out are reported as exhausted after the last retry. +#[nativelink_test] +async fn retry_reports_exhaustion() -> Result<(), Error> { + let (retrier, outcomes) = observed_retrier(2); + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + RetryResult::::Retry(make_err!(Code::Unavailable, "Dummy failure")) + })) + .await; + assert!(result.is_err()); + assert_eq!( + *outcomes.lock().unwrap(), + vec![ + RetryOutcome::Retried, + RetryOutcome::Retried, + RetryOutcome::Exhausted + ] + ); + Ok(()) +} + +/// An error that is never retried is reported as permanent, except +/// `NotFound`, which on a store is an ordinary miss. +#[nativelink_test] +async fn retry_reports_permanent_errors_but_not_misses() -> Result<(), Error> { + let (retrier, outcomes) = observed_retrier(5); + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + RetryResult::::Retry(make_err!(Code::PermissionDenied, "Dummy failure")) + })) + .await; + assert!(result.is_err()); + assert_eq!(*outcomes.lock().unwrap(), vec![RetryOutcome::Permanent]); + + let (retrier, outcomes) = observed_retrier(5); + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + RetryResult::::Retry(make_err!(Code::NotFound, "Dummy miss")) + })) + .await; + assert!(result.is_err()); + assert_eq!(*outcomes.lock().unwrap(), vec![]); + Ok(()) +} + fn recording_retrier(backoff: Duration, sleeps: &Arc>>) -> Retrier { let sleeps = Arc::clone(sleeps); Retrier::new( @@ -412,3 +506,32 @@ async fn retry_after_with_permanent_code_returns_at_once() -> Result<(), Error> assert!(sleeps.lock().unwrap().is_empty()); Ok(()) } + +/// A throttle waited out within the budget spends no attempt but is still a +/// retry, and a success after only such throttles is a recovery. +#[nativelink_test] +async fn throttles_within_the_budget_report_retries_and_recovery() -> Result<(), Error> { + let (retrier, outcomes) = observed_retrier(1); + let retrier = retrier.with_throttle_budget(Duration::from_secs(10)); + let run_count = AtomicI32::new(0); + Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 2 { + return RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ); + } + RetryResult::Ok(true) + })) + .await?; + assert_eq!( + *outcomes.lock().unwrap(), + vec![ + RetryOutcome::Retried, + RetryOutcome::Retried, + RetryOutcome::Recovered + ] + ); + Ok(()) +} diff --git a/web/apps/docs/content/docs/reference/metrics.mdx b/web/apps/docs/content/docs/reference/metrics.mdx index d2b70e632c..bdde26a82f 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 (9aa3e3e90) + Source: nativelink-util/src/metrics.rs @ v1.7.6 (ddde488d) Regenerate from web/: bun --filter @nativelink/docs gen:metrics-reference */} -NativeLink **v1.7.6** declares 49 OpenTelemetry instruments. This page is +NativeLink **v1.7.6** declares 50 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 @@ -73,6 +73,7 @@ wrong. | [`connection.pool.available`](#connection-pool-available) | `nativelink_connection_pool_available` (unreferenced) | Histogram | Yes | whenever the code path runs | | [`connection.pool.acquisitions`](#connection-pool-acquisitions) | `nativelink_connection_pool_acquisitions` (unreferenced) | Counter | Yes | whenever the code path runs | | [`connection.reconnects`](#connection-reconnects) | `nativelink_connection_reconnects` (unreferenced) | Counter | Yes | whenever the code path runs | +| [`retry.outcomes`](#retry-outcomes) | `nativelink_retry_outcomes` (unreferenced) | Counter | Yes | whenever the code path runs | ## How an instrument becomes a Prometheus series @@ -810,7 +811,7 @@ Throttled responses, by provider and operation. | 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:1476`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1476) | -| 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) | +| Emitted from | [`nativelink-store/src/azure_blob_store.rs:269`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-store/src/azure_blob_store.rs#L269) | ## Store tier metrics @@ -890,7 +891,7 @@ Free slots at the moment a connection was taken. | 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:1589`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1589) | -| 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) | +| 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:342`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L342), [`nativelink-util/src/connection_manager.rs:352`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L352) | **Histogram buckets:** 11 explicit boundaries, from `0.0` to `512.0` `{connection}`. Anything past the last boundary lands in the @@ -909,7 +910,7 @@ Acquisitions, by pool and result. | 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:1598`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1598) | -| 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) | +| 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:342`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L342), [`nativelink-util/src/connection_manager.rs:352`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/connection_manager.rs#L352) | ### `connection.reconnects` @@ -925,6 +926,26 @@ Reconnects, by pool. | Declared at | [`nativelink-util/src/metrics.rs:1604`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1604) | | 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) | +## RETRY_METRICS + +Global metrics for operations run under a [`crate::retry::Retrier`]. + +Declared as `RETRY_METRICS` in [`nativelink-util/src/metrics.rs:1685`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1685), backed by the `RetryMetrics` struct. + +### `retry.outcomes` + +Retry events, by client and outcome. + +| Property | Value | +| --- | --- | +| Instrument type | `Counter` | +| Field | `outcomes` | +| Unit | `{event}` | +| Prometheus series | `nativelink_retry_outcomes` (not referenced by any shipped rule or dashboard) | +| Emitted | Yes. whenever the code path runs | +| Declared at | [`nativelink-util/src/metrics.rs:1689`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/metrics.rs#L1689) | +| Emitted from | [`nativelink-util/src/retry.rs:171`](https://github.com/TraceMachina/nativelink/blob/v1.7.6/nativelink-util/src/retry.rs#L171) | + ## Attributes Every attribute key NativeLink attaches to a metric, and the closed set of