Count retries, recoveries and exhausted retries per client

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,

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

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"),
         }
     }
 }

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,

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,

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

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

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

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<RetryMetrics> = 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<u64>,
+}
+
+/// 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),
+        ],
+    );
+}

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<T> {
     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<dyn Fn(RetryOutcome) + Send + Sync>;
+
 /// 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<OutcomeFn>,
     /// 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;
             }
         }