From fbb6cebeb4ee6fd18fe2324395289c12c14c4a10 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 3 Aug 2026 02:24:26 +0800 Subject: [PATCH] feat(kms): bound backend concurrency and failures (#5651) --- .../prometheus-rules/rustfs-kms-alerts.yml | 42 +- crates/kms/src/backends/aws.rs | 60 +- crates/kms/src/backends/vault.rs | 10 +- crates/kms/src/backends/vault_credentials.rs | 75 +- crates/kms/src/backends/vault_transit.rs | 12 +- crates/kms/src/backup/vault_restore.rs | 16 +- crates/kms/src/policy.rs | 1013 ++++++++++++++++- .../grafana/rustfs-kms-observability.json | 6 +- docs/architecture/global-state-inventory.md | 1 + docs/operations/kms-observability-runbook.md | 34 +- 10 files changed, 1212 insertions(+), 57 deletions(-) diff --git a/.docker/observability/prometheus-rules/rustfs-kms-alerts.yml b/.docker/observability/prometheus-rules/rustfs-kms-alerts.yml index e4f1faf03..1d60b8797 100644 --- a/.docker/observability/prometheus-rules/rustfs-kms-alerts.yml +++ b/.docker/observability/prometheus-rules/rustfs-kms-alerts.yml @@ -17,9 +17,9 @@ # ============================================================================= # # Metric source: the KMS operation-policy choke point in -# crates/kms/src/policy.rs. All label values are static enum strings -# (operation, op_class, outcome, error_class); key identifiers, key material, -# and tokens never appear in labels. +# crates/kms/src/policy.rs. All label values are bounded static strings +# (operation, op_class, outcome, error_class, backend, scope); key identifiers, +# key material, and tokens never appear in labels. # # Response procedures: docs/operations/kms-observability-runbook.md # @@ -70,8 +70,9 @@ groups: # ------------------------------------------------------------------ # 2. KmsBackendHighErrorRate # Sustained share of operations terminating without success - # (fatal, budget_exhausted, deadline_exceeded). The cancelled - # outcome is excluded because shutdowns legitimately produce it. + # (fatal, budget/deadline exhaustion, admission backpressure, + # or an open circuit). The cancelled outcome is excluded because + # shutdowns legitimately produce it. # The traffic guard keeps a single failure on a near-idle # cluster from firing the alert. # Threshold: 5% for 10m — conservative default, calibrate @@ -94,9 +95,11 @@ groups: summary: "KMS backend non-success ratio above 5% for 10m" description: >- {{ $value | humanizePercentage }} of KMS backend operations - are terminating in fatal, budget_exhausted, or - deadline_exceeded. Object encryption and decryption paths - depending on the KMS are degraded or failing. + are terminating in fatal, budget_exhausted, + deadline_exceeded, backpressure_timeout, + backpressure_rejected, or circuit_open. Object encryption + and decryption paths depending on the KMS are degraded or + failing. runbook_url: "https://github.com/rustfs/rustfs/blob/main/docs/operations/kms-observability-runbook.md#kmsbackendhigherrorrate" # ========================================================================== @@ -186,3 +189,26 @@ groups: Retryable failures are outlasting the retry budget, so callers are seeing hard failures. runbook_url: "https://github.com/rustfs/rustfs/blob/main/docs/operations/kms-observability-runbook.md#kmsbackendretrybudgetexhausted" + + # ------------------------------------------------------------------ + # 6. KmsBackendCircuitOpen + # Direct circuit-state signal, independent of operation traffic. + # A transient open can recover on its first half-open probe; alert + # only when the circuit remains open or half-open for one minute. + # ------------------------------------------------------------------ + - alert: KmsBackendCircuitOpen + expr: | + rustfs_kms_backend_circuit_open > 0 + for: 1m + labels: + severity: warning + component: kms + annotations: + summary: "KMS backend circuit open ({{ $labels.backend }}/{{ $labels.scope }})" + description: >- + The KMS backend circuit for {{ $labels.backend }} scope + {{ $labels.scope }} has remained open or half-open for one + minute. Operations in this scope can terminate as + circuit_open until the half-open probe succeeds or returns + a non-retryable failure. + runbook_url: "https://github.com/rustfs/rustfs/blob/main/docs/operations/kms-observability-runbook.md#kmsbackendcircuitopen" diff --git a/crates/kms/src/backends/aws.rs b/crates/kms/src/backends/aws.rs index 7853aca26..a00e01d4e 100644 --- a/crates/kms/src/backends/aws.rs +++ b/crates/kms/src/backends/aws.rs @@ -309,11 +309,11 @@ impl AwsKmsBackend { loader = loader.region(aws_sdk_kms::config::Region::new(region.clone())); } let sdk_config = loader.load().await; - if sdk_config.region().is_none() { + let Some(region) = sdk_config.region() else { return Err(KmsError::configuration_error( "AWS KMS backend could not resolve a region; set the backend region or AWS_REGION", )); - } + }; let mut builder = aws_sdk_kms::config::Builder::from(&sdk_config) // `crate::policy` owns retries and timeouts; leaving the SDK's own @@ -324,13 +324,18 @@ impl AwsKmsBackend { builder = builder.endpoint_url(endpoint_url); } - Ok(Self::with_client(aws_sdk_kms::Client::from_conf(builder.build()), &config)) + Ok(Self::with_client( + aws_sdk_kms::Client::from_conf(builder.build()), + &config, + aws_backend_config.endpoint_url.as_deref().unwrap_or_default(), + region.as_ref(), + )) } - fn with_client(client: aws_sdk_kms::Client, config: &KmsConfig) -> Self { + fn with_client(client: aws_sdk_kms::Client, config: &KmsConfig, endpoint: &str, region: &str) -> Self { Self { client, - retry: RetryPolicy::from_config(config), + retry: RetryPolicy::for_backend(config, "aws", endpoint, Some(region), "operations"), cancel: CancellationToken::new(), } } @@ -826,6 +831,7 @@ mod tests { use aws_smithy_types::body::SdkBody; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64; + use std::sync::atomic::{AtomicU64, Ordering}; /// AWS KMS speaks awsJson1_1; every request goes to `/` on the regional /// endpoint, so the replayed request side carries no useful assertion. @@ -857,6 +863,15 @@ mod tests { ) } + fn scripted_endpoint() -> String { + static NEXT_SCRIPTED_BACKEND_ID: AtomicU64 = AtomicU64::new(0); + + format!( + "https://scripted-{}.example.invalid", + NEXT_SCRIPTED_BACKEND_ID.fetch_add(1, Ordering::Relaxed) + ) + } + fn scripted_backend(events: Vec) -> (StaticReplayClient, AwsKmsBackend) { let http_client = StaticReplayClient::new(events); let sdk_config = aws_sdk_kms::Config::builder() @@ -867,10 +882,20 @@ mod tests { .retry_config(aws_sdk_kms::config::retry::RetryConfig::disabled()) .build(); let kms_config = KmsConfig::aws(Some("us-east-1".to_string())); - let backend = AwsKmsBackend::with_client(aws_sdk_kms::Client::from_conf(sdk_config), &kms_config); + let endpoint = scripted_endpoint(); + let backend = AwsKmsBackend::with_client(aws_sdk_kms::Client::from_conf(sdk_config), &kms_config, &endpoint, "us-east-1"); (http_client, backend) } + fn aws_config(endpoint: &str, region: &str) -> KmsConfig { + let mut config = KmsConfig::aws(Some(region.to_owned())); + let BackendConfig::Aws(aws) = &mut config.backend_config else { + panic!("AWS constructor must create an AWS backend configuration"); + }; + aws.endpoint_url = Some(endpoint.to_owned()); + config + } + fn key_metadata_json(key_id: &str, state: &str) -> serde_json::Value { serde_json::json!({ "KeyMetadata": { @@ -900,6 +925,27 @@ mod tests { .expect("capabilities should deserialize into a flat bool map") } + #[tokio::test] + async fn aws_backend_new_identity_includes_endpoint_and_region() { + let endpoint = scripted_endpoint(); + let first = AwsKmsBackend::new(aws_config(&endpoint, "us-east-1")) + .await + .expect("first AWS backend"); + let matching = AwsKmsBackend::new(aws_config(&endpoint, "us-east-1")) + .await + .expect("matching AWS backend"); + let other_region = AwsKmsBackend::new(aws_config(&endpoint, "us-west-2")) + .await + .expect("other-region AWS backend"); + let other_endpoint = AwsKmsBackend::new(aws_config(&scripted_endpoint(), "us-east-1")) + .await + .expect("other-endpoint AWS backend"); + + assert!(first.retry.shares_active_capacity_with(&matching.retry)); + assert!(!first.retry.shares_active_capacity_with(&other_region.retry)); + assert!(!first.retry.shares_active_capacity_with(&other_endpoint.retry)); + } + #[tokio::test] async fn aws_backend_capabilities_golden() { let (_http, backend) = scripted_backend(Vec::new()); @@ -1023,7 +1069,7 @@ mod tests { let mut kms_config = KmsConfig::aws(Some("us-east-1".to_string())); kms_config.timeout = std::time::Duration::from_millis(5_000); kms_config.retry_attempts = 1; - let backend = AwsKmsBackend::with_client(aws_sdk_kms::Client::from_conf(sdk_config), &kms_config); + let backend = AwsKmsBackend::with_client(aws_sdk_kms::Client::from_conf(sdk_config), &kms_config, "", "us-east-1"); let error = backend .describe_key(DescribeKeyRequest { diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index 81c973f62..c928e0f9b 100644 --- a/crates/kms/src/backends/vault.rs +++ b/crates/kms/src/backends/vault.rs @@ -227,7 +227,13 @@ impl VaultKmsClient { attempt_timeout: kms_config.effective_timeout(), }; let source = token_source_for(&config.auth_method, &settings)?; - let policy = VaultCredentialPolicy::from_kms_config(kms_config, &config.auth_method); + let policy = VaultCredentialPolicy::from_kms_config( + kms_config, + &config.auth_method, + "vault-kv2", + &config.address, + config.namespace.as_deref(), + ); let credentials = Arc::new(VaultCredentialProvider::new(settings, source, policy).await?); info!(address = %config.address, "Vault KMS backend connected"); @@ -237,7 +243,7 @@ impl VaultKmsClient { kv_mount: config.kv_mount.clone(), key_path_prefix: config.key_path_prefix.clone(), dek_crypto: AesDekCrypto::new(), - retry: RetryPolicy::from_config(kms_config), + retry: RetryPolicy::for_backend(kms_config, "vault-kv2", &config.address, config.namespace.as_deref(), "operations"), cancel: CancellationToken::new(), }) } diff --git a/crates/kms/src/backends/vault_credentials.rs b/crates/kms/src/backends/vault_credentials.rs index c5a300908..b8766442b 100644 --- a/crates/kms/src/backends/vault_credentials.rs +++ b/crates/kms/src/backends/vault_credentials.rs @@ -520,8 +520,10 @@ impl VaultConnectionSettings { /// Refresh and fail-closed tuning for a [`VaultCredentialProvider`]. #[derive(Debug, Clone)] pub(crate) struct VaultCredentialPolicy { - /// Retry budget for one login/renewal cycle. + /// Retry budget for one login cycle. pub(crate) retry: RetryPolicy, + /// Retry budget for one token-renewal cycle. + pub(crate) renew_retry: RetryPolicy, /// Fail-closed margin: once the current token is within this window of /// expiry without a successful refresh, [`VaultCredentialProvider::current`] /// refuses to hand it out. @@ -536,8 +538,14 @@ impl VaultCredentialPolicy { /// The default safety window equals the per-attempt timeout: a request /// issued now can stay in flight for up to one attempt timeout, so the /// token must outlive at least that. - pub(crate) fn from_kms_config(config: &KmsConfig, auth_method: &VaultAuthMethod) -> Self { - let retry = RetryPolicy::from_config(config); + pub(crate) fn from_kms_config( + config: &KmsConfig, + auth_method: &VaultAuthMethod, + backend: &'static str, + endpoint: &str, + namespace: Option<&str>, + ) -> Self { + let retry = RetryPolicy::for_credentials(config, backend, endpoint, namespace, "credentials-login"); let safety_window = match auth_method { VaultAuthMethod::AppRole { refresh_safety_window_secs: Some(secs), @@ -551,6 +559,7 @@ impl VaultCredentialPolicy { }; Self { retry, + renew_retry: RetryPolicy::for_credentials(config, backend, endpoint, namespace, "credentials-renew"), safety_window, retry_interval: DEFAULT_REFRESH_RETRY_INTERVAL, } @@ -717,7 +726,7 @@ impl VaultCredentialProvider { let renewable = current.lease.map(|lease| lease.renewable).unwrap_or(false); let renewed = if renewable { - match policy::execute("vault_token_renew", OpClass::Auth, &self.policy.retry, cancel, || { + match policy::execute("vault_token_renew", OpClass::Auth, &self.policy.renew_retry, cancel, || { self.source.renew(¤t.client) }) .await @@ -868,19 +877,56 @@ mod tests { /// Tight retry budget so paused-clock tests stay deterministic: one /// attempt per cycle, failed cycles spaced by `retry_interval`. fn test_policy(safety_window: Duration, retry_interval: Duration) -> VaultCredentialPolicy { + let retry = RetryPolicy::for_test( + Duration::from_secs(1), + Duration::from_secs(1), + 1, + Duration::from_millis(10), + Duration::from_millis(10), + ); + let renew_retry = RetryPolicy::for_test( + Duration::from_secs(1), + Duration::from_secs(1), + 1, + Duration::from_millis(10), + Duration::from_millis(10), + ); VaultCredentialPolicy { - retry: RetryPolicy { - attempt_timeout: Duration::from_secs(1), - op_deadline: Duration::from_secs(1), - max_attempts: 1, - base_backoff: Duration::from_millis(10), - max_backoff: Duration::from_millis(10), - }, + retry, + renew_retry, safety_window, retry_interval, } } + #[test] + fn configured_login_and_renewal_use_reserved_credential_capacity() { + let config = KmsConfig::default(); + let auth_method = VaultAuthMethod::Token { + token: TEST_TOKEN.to_string(), + }; + let credentials = VaultCredentialPolicy::from_kms_config( + &config, + &auth_method, + "vault-kv2", + "https://credential-policy.example.invalid", + Some("team-namespace"), + ); + let operations = RetryPolicy::for_backend( + &config, + "vault-kv2", + "https://credential-policy.example.invalid", + Some("team-namespace"), + "operations", + ); + + assert!(credentials.retry.uses_credential_reserve()); + assert!(credentials.renew_retry.uses_credential_reserve()); + assert!(!operations.uses_credential_reserve()); + assert!(credentials.retry.shares_active_capacity_with(&credentials.renew_retry)); + assert!(credentials.retry.shares_active_capacity_with(&operations)); + } + /// Shared observable state of a [`ScriptedSource`]. #[derive(Debug, Default)] struct ScriptedState { @@ -1084,11 +1130,12 @@ mod tests { "expected CredentialsUnavailable, got {error:?}" ); - // Recovery: the next retry cycle succeeds, installs a fresh - // generation, and the provider serves requests again. + // Recovery: after the failed-refresh circuit's cool-down, its single + // half-open probe succeeds, installs a fresh generation, and the + // provider serves requests again. state.fail_renew.store(false, Ordering::SeqCst); state.fail_login.store(false, Ordering::SeqCst); - tokio::time::sleep(Duration::from_secs(6)).await; + tokio::time::sleep(Duration::from_secs(31)).await; let handle = provider.current().expect("provider must recover after a successful refresh"); assert!(handle.generation >= 1); diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index 3f3f95786..7d5d9b0e5 100644 --- a/crates/kms/src/backends/vault_transit.rs +++ b/crates/kms/src/backends/vault_transit.rs @@ -230,8 +230,16 @@ impl VaultTransitKmsClient { attempt_timeout: kms_config.effective_timeout(), }; let source = token_source_for(&config.auth_method, &settings)?; - let policy = VaultCredentialPolicy::from_kms_config(kms_config, &config.auth_method); + let policy = VaultCredentialPolicy::from_kms_config( + kms_config, + &config.auth_method, + "vault-transit", + &config.address, + config.namespace.as_deref(), + ); let credentials = Arc::new(VaultCredentialProvider::new(settings, source, policy).await?); + let retry = + RetryPolicy::for_backend(kms_config, "vault-transit", &config.address, config.namespace.as_deref(), "operations"); Ok(Self { credentials, @@ -242,7 +250,7 @@ impl VaultTransitKmsClient { .max_capacity(METADATA_CACHE_CAPACITY) .time_to_live(METADATA_CACHE_TTL) .build(), - retry: RetryPolicy::from_config(kms_config), + retry, cancel: CancellationToken::new(), }) } diff --git a/crates/kms/src/backup/vault_restore.rs b/crates/kms/src/backup/vault_restore.rs index 3986ad640..b7d4884cc 100644 --- a/crates/kms/src/backup/vault_restore.rs +++ b/crates/kms/src/backup/vault_restore.rs @@ -452,12 +452,24 @@ impl VaultRestoreClient { attempt_timeout: kms_config.effective_timeout(), }; let source = token_source_for(&target.auth_method, &settings)?; - let policy = VaultCredentialPolicy::from_kms_config(kms_config, &target.auth_method); + let policy = VaultCredentialPolicy::from_kms_config( + kms_config, + &target.auth_method, + "vault-restore", + &target.address, + target.namespace.as_deref(), + ); let credentials = Arc::new(VaultCredentialProvider::new(settings, source, policy).await?); Ok(Self { credentials, kv_mount: target.kv_mount.clone(), - retry: RetryPolicy::from_config(kms_config), + retry: RetryPolicy::for_backend( + kms_config, + "vault-restore", + &target.address, + target.namespace.as_deref(), + "operations", + ), cancel: CancellationToken::new(), }) } diff --git a/crates/kms/src/policy.rs b/crates/kms/src/policy.rs index da478a224..40e0c264d 100644 --- a/crates/kms/src/policy.rs +++ b/crates/kms/src/policy.rs @@ -34,10 +34,14 @@ //! values — operation names, classes, outcomes — never key identifiers, key //! material, ciphertext, or tokens. +use std::collections::HashMap; use std::future::Future; +use std::sync::{Arc, LazyLock, Mutex, Weak}; use std::time::Duration; use rand::{RngExt, SeedableRng, rngs::StdRng}; +use sha2::{Digest, Sha256}; +use tokio::sync::{AcquireError, OwnedSemaphorePermit, Semaphore, TryAcquireError}; use tokio::time::Instant; use tokio_util::sync::CancellationToken; @@ -49,6 +53,13 @@ const DEFAULT_BASE_BACKOFF: Duration = Duration::from_millis(100); /// Default upper bound for a single backoff sleep. const DEFAULT_MAX_BACKOFF: Duration = Duration::from_secs(2); +const DEFAULT_MAX_CONCURRENT_OPERATIONS: usize = 64; +const RESERVED_CREDENTIAL_OPERATIONS: usize = 1; +const DEFAULT_MAX_QUEUED_OPERATIONS: usize = 64; +const DEFAULT_CIRCUIT_FAILURE_THRESHOLD: u32 = 5; +const DEFAULT_CIRCUIT_OPEN_DURATION: Duration = Duration::from_secs(30); +const CIRCUIT_OPEN_MESSAGE: &str = "KMS backend circuit is open; waiting for the half-open recovery probe"; +const BACKPRESSURE_MESSAGE: &str = "KMS backend capacity and wait queue are full"; /// Replay safety of a backend operation. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -147,7 +158,7 @@ impl AttemptError { } /// Budgets applied by [`execute`]. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone)] pub(crate) struct RetryPolicy { /// Upper bound for one backend attempt. pub(crate) attempt_timeout: Duration, @@ -160,6 +171,7 @@ pub(crate) struct RetryPolicy { pub(crate) base_backoff: Duration, /// Upper bound for a single backoff sleep. pub(crate) max_backoff: Duration, + runtime: Arc, } impl RetryPolicy { @@ -170,13 +182,47 @@ impl RetryPolicy { /// attempts plus backoff, so it bounds runaway loops without cutting any /// attempt short; an independently configurable deadline is left to the /// admin-API follow-up. - pub(crate) fn from_config(config: &KmsConfig) -> Self { + pub(crate) fn for_backend( + config: &KmsConfig, + backend: &'static str, + endpoint: &str, + namespace: Option<&str>, + scope: &'static str, + ) -> Self { + Self::for_capacity(config, backend, endpoint, namespace, scope, CapacityClass::Operations) + } + + pub(crate) fn for_credentials( + config: &KmsConfig, + backend: &'static str, + endpoint: &str, + namespace: Option<&str>, + scope: &'static str, + ) -> Self { + Self::for_capacity(config, backend, endpoint, namespace, scope, CapacityClass::Credentials) + } + + fn for_capacity( + config: &KmsConfig, + backend: &'static str, + endpoint: &str, + namespace: Option<&str>, + scope: &'static str, + capacity: CapacityClass, + ) -> Self { + let endpoint = url::Url::parse(endpoint).map_or_else(|_| endpoint.to_owned(), |url| url.to_string()); + let mut identity = Sha256::new(); + for part in [backend, endpoint.as_str(), namespace.unwrap_or_default()] { + identity.update(part.len().to_be_bytes()); + identity.update(part.as_bytes()); + } let mut policy = Self { attempt_timeout: config.effective_timeout(), op_deadline: Duration::ZERO, max_attempts: config.effective_retry_attempts(), base_backoff: DEFAULT_BASE_BACKOFF, max_backoff: DEFAULT_MAX_BACKOFF, + runtime: BackendRuntime::shared(identity.finalize().into(), backend, scope, capacity), }; policy.op_deadline = policy.worst_case_budget(); policy @@ -194,6 +240,237 @@ impl RetryPolicy { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CapacityClass { + Operations, + Credentials, +} + +static ACTIVE_REGISTRY: LazyLock>>> = LazyLock::new(|| Mutex::new(HashMap::new())); + +#[derive(Debug)] +struct BackendCapacity { + total: Arc, + operations: Arc, +} + +#[derive(Debug)] +struct BackendRuntime { + capacity: Arc, + capacity_class: CapacityClass, + queued: Arc, + breaker: Mutex, + in_flight: metrics::Gauge, + circuit_open: metrics::Gauge, +} + +impl BackendRuntime { + fn shared(identity: [u8; 32], backend: &'static str, scope: &'static str, capacity: CapacityClass) -> Arc { + let mut registry = ACTIVE_REGISTRY.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + registry.retain(|_, active| active.strong_count() > 0); + let active = registry.get(&identity).and_then(Weak::upgrade).unwrap_or_else(|| { + Arc::new(BackendCapacity { + total: Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT_OPERATIONS)), + operations: Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT_OPERATIONS - RESERVED_CREDENTIAL_OPERATIONS)), + }) + }); + registry.insert(identity, Arc::downgrade(&active)); + Arc::new(Self::new(backend, scope, active, capacity)) + } + + fn new(backend: &'static str, scope: &'static str, active: Arc, capacity: CapacityClass) -> Self { + let in_flight = metrics::gauge!(METRIC_IN_FLIGHT, "backend" => backend, "scope" => scope); + let circuit_open = metrics::gauge!(METRIC_CIRCUIT_OPEN, "backend" => backend, "scope" => scope); + in_flight.increment(0.0); + circuit_open.increment(0.0); + Self { + capacity: active, + capacity_class: capacity, + queued: Arc::new(Semaphore::new(DEFAULT_MAX_QUEUED_OPERATIONS)), + breaker: Mutex::new(CircuitState::default()), + in_flight, + circuit_open, + } + } + + fn try_acquire_active(&self) -> std::result::Result { + let operations = matches!(self.capacity_class, CapacityClass::Operations) + .then(|| Arc::clone(&self.capacity.operations).try_acquire_owned()) + .transpose()?; + match Arc::clone(&self.capacity.total).try_acquire_owned() { + Ok(total) => Ok(ActivePermits { + _total: total, + _operations: operations, + }), + Err(error) => { + drop(operations); + Err(error) + } + } + } + + async fn acquire_active(&self) -> std::result::Result { + let operations = match self.capacity_class { + CapacityClass::Operations => Some(Arc::clone(&self.capacity.operations).acquire_owned().await?), + CapacityClass::Credentials => None, + }; + let total = Arc::clone(&self.capacity.total).acquire_owned().await?; + Ok(ActivePermits { + _total: total, + _operations: operations, + }) + } + + fn breaker_state(&self) -> Result> { + self.breaker + .lock() + .map_err(|_| KmsError::backend_error("KMS backend circuit state is unavailable")) + } + + fn rejects(&self, now: Instant) -> Result { + let state = self.breaker_state()?; + Ok(matches!(state.open_until, Some(open_until) if now < open_until || state.probe_in_flight)) + } + + fn admit(&self, now: Instant) -> Result> { + let mut state = self.breaker_state()?; + Ok(match state.open_until { + None => Some(CircuitAdmission { + generation: state.generation, + probe: false, + }), + Some(open_until) if now >= open_until && !state.probe_in_flight => { + state.probe_in_flight = true; + Some(CircuitAdmission { + generation: state.generation, + probe: true, + }) + } + Some(_) => None, + }) + } + + fn transition(&self, state: &mut CircuitState, open: bool) { + let was_open = state.open_until.is_some(); + state.generation = state.generation.saturating_add(1); + state.consecutive_failures = 0; + state.open_until = open.then(|| Instant::now() + DEFAULT_CIRCUIT_OPEN_DURATION); + state.probe_in_flight = false; + if was_open != open { + if open { + self.circuit_open.increment(1.0); + } else { + self.circuit_open.decrement(1.0); + } + } + } + + fn complete(&self, admission: CircuitAdmission, failure: Option) -> bool { + let Ok(mut state) = self.breaker_state() else { + return true; + }; + if state.generation != admission.generation { + return state.open_until.is_some(); + } + match failure { + None | Some(ErrorClass::Fatal) => { + state.consecutive_failures = 0; + if admission.probe { + self.transition(&mut state, false); + } + } + Some(ErrorClass::RetryableConn | ErrorClass::RetryableStatus) => { + state.consecutive_failures = state.consecutive_failures.saturating_add(1); + if admission.probe || state.consecutive_failures >= DEFAULT_CIRCUIT_FAILURE_THRESHOLD { + self.transition(&mut state, true); + } + } + } + state.open_until.is_some() + } + + fn abandon(&self, admission: CircuitAdmission) { + if let Ok(mut state) = self.breaker_state() + && state.generation == admission.generation + && admission.probe + { + self.transition(&mut state, true); + } + } +} + +impl Drop for BackendRuntime { + fn drop(&mut self) { + let state = self.breaker.get_mut().unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.open_until.is_some() { + self.circuit_open.decrement(1.0); + } + } +} + +#[derive(Debug, Default)] +struct CircuitState { + generation: u64, + consecutive_failures: u32, + open_until: Option, + probe_in_flight: bool, +} + +#[derive(Debug, Clone, Copy)] +struct CircuitAdmission { + generation: u64, + probe: bool, +} + +struct RuntimePermit { + runtime: Arc, + admission: Option, + _active: ActivePermits, +} + +struct ActivePermits { + _total: OwnedSemaphorePermit, + _operations: Option, +} + +impl RuntimePermit { + #[cfg(test)] + fn new(runtime: Arc, admission: CircuitAdmission, active: OwnedSemaphorePermit) -> Self { + Self::with_active( + runtime, + admission, + ActivePermits { + _total: active, + _operations: None, + }, + ) + } + + fn with_active(runtime: Arc, admission: CircuitAdmission, active: ActivePermits) -> Self { + runtime.in_flight.increment(1.0); + Self { + runtime, + admission: Some(admission), + _active: active, + } + } + + fn complete(&mut self, failure: Option) -> bool { + self.admission + .take() + .is_some_and(|admission| self.runtime.complete(admission, failure)) + } +} + +impl Drop for RuntimePermit { + fn drop(&mut self) { + if let Some(admission) = self.admission.take() { + self.runtime.abandon(admission); + } + self.runtime.in_flight.decrement(1.0); + } +} + /// Exponential backoff cap after `completed_attempts` failed attempts. fn backoff_cap(policy: &RetryPolicy, completed_attempts: u32) -> Duration { let doublings = completed_attempts.saturating_sub(1).min(31); @@ -229,6 +506,10 @@ const METRIC_OPERATION_DURATION_SECONDS: &str = "rustfs_kms_backend_operation_du /// Histogram: attempts one operation used before completing, by `operation` /// and `outcome`. const METRIC_OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts"; +/// Gauge: backend attempts currently in flight. +const METRIC_IN_FLIGHT: &str = "rustfs_kms_backend_in_flight"; +/// Gauge: number of open or half-open backend circuits. +const METRIC_CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open"; impl OpClass { fn as_label(self) -> &'static str { @@ -261,6 +542,12 @@ enum Outcome { BudgetExhausted, /// The operation deadline ran out before another attempt could complete. DeadlineExceeded, + /// The operation deadline ran out while waiting for backend capacity. + BackpressureTimeout, + /// Backend capacity and the bounded wait queue were both full. + BackpressureRejected, + /// A retryable failure opened the circuit, or an open circuit rejected the operation. + CircuitOpen, Cancelled, } @@ -271,6 +558,9 @@ impl Outcome { Outcome::Fatal => "fatal", Outcome::BudgetExhausted => "budget_exhausted", Outcome::DeadlineExceeded => "deadline_exceeded", + Outcome::BackpressureTimeout => "backpressure_timeout", + Outcome::BackpressureRejected => "backpressure_rejected", + Outcome::CircuitOpen => "circuit_open", Outcome::Cancelled => "cancelled", } } @@ -296,6 +586,11 @@ fn describe_metrics() { METRIC_OPERATION_ATTEMPTS, "Number of attempts a KMS backend operation used before completing" ); + metrics::describe_gauge!(METRIC_IN_FLIGHT, "KMS backend attempts currently in flight, by backend and policy scope"); + metrics::describe_gauge!( + METRIC_CIRCUIT_OPEN, + "Number of open or half-open KMS backend circuits, by backend and policy scope" + ); }); } @@ -332,6 +627,83 @@ fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, a .record(f64::from(attempts)); } +async fn acquire_attempt( + runtime: Arc, + operation: &'static str, + cancel: &CancellationToken, + deadline: Instant, +) -> std::result::Result { + if cancel.is_cancelled() { + return Err(( + Outcome::Cancelled, + KmsError::operation_cancelled(format!("{operation} cancelled before backend admission")), + )); + } + match runtime.rejects(Instant::now()) { + Ok(false) => {} + Ok(true) => return Err((Outcome::CircuitOpen, KmsError::backend_error(CIRCUIT_OPEN_MESSAGE))), + Err(error) => return Err((Outcome::CircuitOpen, error)), + } + + let active = match runtime.try_acquire_active() { + Ok(active) => active, + Err(TryAcquireError::Closed) => { + return Err(( + Outcome::BackpressureRejected, + KmsError::backend_error("KMS backend admission is unavailable"), + )); + } + Err(TryAcquireError::NoPermits) => { + let queued = Arc::clone(&runtime.queued) + .try_acquire_owned() + .map_err(|_| (Outcome::BackpressureRejected, KmsError::backend_error(BACKPRESSURE_MESSAGE)))?; + let acquire = runtime.acquire_active(); + let active = tokio::select! { + biased; + _ = cancel.cancelled() => { + return Err((Outcome::Cancelled, KmsError::operation_cancelled(format!( + "{operation} cancelled while waiting for backend capacity" + )))); + } + result = tokio::time::timeout_at(deadline, acquire) => match result { + Ok(Ok(active)) => active, + Ok(Err(_)) => { + return Err((Outcome::BackpressureRejected, KmsError::backend_error( + "KMS backend admission is unavailable" + ))); + } + Err(_) => { + return Err((Outcome::BackpressureTimeout, KmsError::operation_timed_out(format!( + "{operation} exceeded its operation deadline while waiting for backend capacity" + )))); + } + } + }; + drop(queued); + active + } + }; + + if cancel.is_cancelled() { + return Err(( + Outcome::Cancelled, + KmsError::operation_cancelled(format!("{operation} cancelled before backend attempt")), + )); + } + if Instant::now() >= deadline { + return Err(( + Outcome::BackpressureTimeout, + KmsError::operation_timed_out(format!("{operation} exceeded its operation deadline before backend attempt")), + )); + } + let admission = match runtime.admit(Instant::now()) { + Ok(Some(admission)) => admission, + Ok(None) => return Err((Outcome::CircuitOpen, KmsError::backend_error(CIRCUIT_OPEN_MESSAGE))), + Err(error) => return Err((Outcome::CircuitOpen, error)), + }; + Ok(RuntimePermit::with_active(runtime, admission, active)) +} + /// Run `attempt` under the policy. /// /// Each attempt is bounded by `attempt_timeout` (further capped by whatever is @@ -423,8 +795,14 @@ where ); } + let mut runtime_permit = match acquire_attempt(Arc::clone(&policy.runtime), operation, cancel, deadline).await { + Ok(permit) => permit, + Err((outcome, error)) => return (outcome, Err(error)), + }; + attempt_no += 1; *attempts_made = attempt_no; + let remaining = deadline.saturating_duration_since(Instant::now()); let attempt_budget = policy.attempt_timeout.min(remaining); let outcome = tokio::select! { biased; @@ -438,7 +816,10 @@ where }; let failure = match outcome { - Ok(Ok(value)) => return (Outcome::Success, Ok(value)), + Ok(Ok(value)) => { + runtime_permit.complete(None); + return (Outcome::Success, Ok(value)); + } Ok(Err(failure)) => { record_attempt_failure(operation, failure.class.as_label()); failure @@ -454,9 +835,14 @@ where } }; + let circuit_open = runtime_permit.complete(Some(failure.class)); if failure.class == ErrorClass::Fatal { return (Outcome::Fatal, Err(failure.error)); } + drop(runtime_permit); + if circuit_open { + return (Outcome::CircuitOpen, Err(failure.error)); + } if attempt_no >= max_attempts { return (Outcome::BudgetExhausted, Err(failure.error)); } @@ -486,22 +872,74 @@ where } } +#[cfg(test)] +fn test_runtime(max_concurrent: usize, max_queued: usize) -> Arc { + let capacity = Arc::new(BackendCapacity { + total: Arc::new(Semaphore::new(max_concurrent)), + operations: Arc::new(Semaphore::new(max_concurrent)), + }); + let mut runtime = BackendRuntime::new("test-backend", "test-scope", capacity, CapacityClass::Credentials); + runtime.queued = Arc::new(Semaphore::new(max_queued)); + Arc::new(runtime) +} + +#[cfg(test)] +impl RetryPolicy { + pub(crate) fn for_test( + attempt_timeout: Duration, + op_deadline: Duration, + max_attempts: u32, + base_backoff: Duration, + max_backoff: Duration, + ) -> Self { + Self { + attempt_timeout, + op_deadline, + max_attempts, + base_backoff, + max_backoff, + runtime: test_runtime(DEFAULT_MAX_CONCURRENT_OPERATIONS, DEFAULT_MAX_QUEUED_OPERATIONS), + } + } + + pub(crate) fn shares_active_capacity_with(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.runtime.capacity, &other.runtime.capacity) + } + + pub(crate) fn uses_credential_reserve(&self) -> bool { + matches!(self.runtime.capacity_class, CapacityClass::Credentials) + } +} + #[cfg(test)] mod tests { use super::*; use std::sync::Arc; - use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; + use tokio::sync::Notify; type AttemptResult = std::result::Result; fn policy_of(attempt_timeout_ms: u64, op_deadline_ms: u64, max_attempts: u32, base_ms: u64, max_ms: u64) -> RetryPolicy { - RetryPolicy { - attempt_timeout: Duration::from_millis(attempt_timeout_ms), - op_deadline: Duration::from_millis(op_deadline_ms), + RetryPolicy::for_test( + Duration::from_millis(attempt_timeout_ms), + Duration::from_millis(op_deadline_ms), max_attempts, - base_backoff: Duration::from_millis(base_ms), - max_backoff: Duration::from_millis(max_ms), - } + Duration::from_millis(base_ms), + Duration::from_millis(max_ms), + ) + } + + fn policy_with_limit(max_concurrent: usize, max_queued: usize) -> RetryPolicy { + let mut policy = policy_of(1_000, 60_000, 1, 10, 10); + policy.runtime = test_runtime(max_concurrent, max_queued); + policy + } + + fn unique_endpoint(label: &str) -> String { + static NEXT_ENDPOINT_ID: AtomicU64 = AtomicU64::new(0); + + format!("https://{label}-{}.example.invalid", NEXT_ENDPOINT_ID.fetch_add(1, Ordering::Relaxed)) } /// Deterministic jitter: always sleep the full backoff cap. @@ -516,6 +954,20 @@ mod tests { } } + fn is_circuit_open(error: &KmsError) -> bool { + matches!(error, KmsError::BackendError { message } if message.contains("circuit is open")) + } + + fn trip_breaker(runtime: &BackendRuntime) { + for attempt in 1..=DEFAULT_CIRCUIT_FAILURE_THRESHOLD { + let admission = runtime.admit(Instant::now()).expect("state").expect("closed"); + assert_eq!( + runtime.complete(admission, Some(ErrorClass::RetryableConn)), + attempt == DEFAULT_CIRCUIT_FAILURE_THRESHOLD + ); + } + } + #[tokio::test(start_paused = true)] async fn hung_attempt_fails_within_attempt_timeout() { let policy = policy_of(5_000, 60_000, 1, 100, 2_000); @@ -660,6 +1112,362 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 0); } + #[test] + fn backend_generation_reuses_capacity_but_resets_queue_and_breaker() { + let endpoint = unique_endpoint("fresh-generation"); + let config = KmsConfig::default(); + let first = RetryPolicy::for_backend(&config, "vault", &endpoint, Some("namespace"), "operations"); + let second = RetryPolicy::for_backend(&config, "vault", &endpoint, Some("namespace"), "operations"); + + assert!(!Arc::ptr_eq(&first.runtime, &second.runtime)); + assert!(Arc::ptr_eq(&first.runtime.capacity, &second.runtime.capacity)); + assert!(!Arc::ptr_eq(&first.runtime.queued, &second.runtime.queued)); + + let _first_queue = Arc::clone(&first.runtime.queued) + .try_acquire_owned() + .expect("first generation queue capacity"); + assert_eq!(first.runtime.queued.available_permits(), DEFAULT_MAX_QUEUED_OPERATIONS - 1); + assert_eq!(second.runtime.queued.available_permits(), DEFAULT_MAX_QUEUED_OPERATIONS); + + trip_breaker(&first.runtime); + assert!(first.runtime.rejects(Instant::now()).expect("first generation breaker state")); + assert!( + !second + .runtime + .rejects(Instant::now()) + .expect("second generation breaker state") + ); + } + + #[test] + fn backend_capacity_domains_isolate_credentials_queue_and_breaker() { + let endpoint = unique_endpoint("shared-capacity"); + let config = KmsConfig::default(); + let operations = RetryPolicy::for_backend(&config, "vault", &endpoint, Some("namespace"), "operations"); + let login = RetryPolicy::for_credentials(&config, "vault", &endpoint, Some("namespace"), "login"); + let renew = RetryPolicy::for_credentials(&config, "vault", &endpoint, Some("namespace"), "renew"); + + assert!(!Arc::ptr_eq(&operations.runtime, &login.runtime)); + assert!(Arc::ptr_eq(&operations.runtime.capacity, &login.runtime.capacity)); + assert!(matches!(operations.runtime.capacity_class, CapacityClass::Operations)); + assert!(matches!(login.runtime.capacity_class, CapacityClass::Credentials)); + assert!(Arc::ptr_eq(&login.runtime.capacity, &renew.runtime.capacity)); + assert!(!Arc::ptr_eq(&operations.runtime.queued, &login.runtime.queued)); + + let _operations_queue = Arc::clone(&operations.runtime.queued) + .try_acquire_owned() + .expect("operations queue capacity"); + assert_eq!(operations.runtime.queued.available_permits(), DEFAULT_MAX_QUEUED_OPERATIONS - 1); + assert_eq!(login.runtime.queued.available_permits(), DEFAULT_MAX_QUEUED_OPERATIONS); + + trip_breaker(&operations.runtime); + assert!(operations.runtime.rejects(Instant::now()).expect("operations breaker state")); + assert!(!login.runtime.rejects(Instant::now()).expect("login breaker state")); + } + + #[tokio::test(start_paused = true)] + async fn credential_refresh_is_not_starved_by_operations_capacity() { + let endpoint = unique_endpoint("credential-capacity"); + let config = KmsConfig::default(); + let operations = RetryPolicy::for_backend(&config, "vault", &endpoint, Some("namespace"), "operations"); + let login = RetryPolicy::for_credentials(&config, "vault", &endpoint, Some("namespace"), "credentials-login"); + let _operation_permits = (0..DEFAULT_MAX_CONCURRENT_OPERATIONS - RESERVED_CREDENTIAL_OPERATIONS) + .map(|_| operations.runtime.try_acquire_active().expect("operations capacity")) + .collect::>(); + assert_eq!(login.runtime.capacity.total.available_permits(), RESERVED_CREDENTIAL_OPERATIONS); + + execute_with_jitter( + "credential_login", + OpClass::Auth, + &login, + &CancellationToken::new(), + full_jitter, + || async { Ok(()) }, + ) + .await + .expect("credential login must use reserved capacity"); + } + + #[test] + fn backend_identity_isolates_endpoint_and_namespace() { + let config = KmsConfig::default(); + let endpoint_a = unique_endpoint("endpoint-a"); + let endpoint_b = unique_endpoint("endpoint-b"); + let first_endpoint = RetryPolicy::for_backend(&config, "vault", &endpoint_a, Some("namespace"), "operations"); + let second_endpoint = RetryPolicy::for_backend(&config, "vault", &endpoint_b, Some("namespace"), "operations"); + assert!(!Arc::ptr_eq(&first_endpoint.runtime.capacity, &second_endpoint.runtime.capacity)); + + let namespace_endpoint = unique_endpoint("namespace"); + let first_namespace = RetryPolicy::for_backend(&config, "vault", &namespace_endpoint, Some("tenant-a"), "operations"); + let second_namespace = RetryPolicy::for_backend(&config, "vault", &namespace_endpoint, Some("tenant-b"), "operations"); + assert!(!Arc::ptr_eq(&first_namespace.runtime.capacity, &second_namespace.runtime.capacity)); + } + + #[tokio::test(start_paused = true)] + async fn admission_bounds_active_queue_and_deadline() { + let mut policy = policy_with_limit(1, 1); + policy.op_deadline = Duration::from_millis(100); + let active = Arc::clone(&policy.runtime.capacity.total) + .acquire_owned() + .await + .expect("active permit"); + let queued = Arc::clone(&policy.runtime.queued) + .acquire_owned() + .await + .expect("queued permit"); + let excess: Result<()> = execute_with_jitter( + "queue_full", + OpClass::ReadIdempotent, + &policy, + &CancellationToken::new(), + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(matches!(&excess, Err(KmsError::BackendError { message }) if message == BACKPRESSURE_MESSAGE)); + drop(queued); + + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(90)).await; + drop(active); + }); + let started = Instant::now(); + let timed_out: Result<()> = execute_with_jitter( + "queue_deadline", + OpClass::ReadIdempotent, + &policy, + &CancellationToken::new(), + full_jitter, + std::future::pending::>, + ) + .await; + assert!(matches!(timed_out, Err(KmsError::OperationTimedOut { .. }))); + assert_eq!(started.elapsed(), Duration::from_millis(100)); + + let active = Arc::clone(&policy.runtime.capacity.total) + .acquire_owned() + .await + .expect("active permit"); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + drop(active); + }); + let boundary = acquire_attempt( + Arc::clone(&policy.runtime), + "deadline_boundary", + &CancellationToken::new(), + Instant::now() + Duration::from_millis(100), + ) + .await; + assert!(matches!(boundary, Err((Outcome::BackpressureTimeout, _)))); + } + + #[tokio::test(start_paused = true)] + async fn queued_admission_cancellation_releases_queue_capacity() { + let policy = Arc::new(policy_with_limit(1, 1)); + let active = Arc::clone(&policy.runtime.capacity.total) + .acquire_owned() + .await + .expect("active permit"); + let cancel = CancellationToken::new(); + let calls = Arc::new(AtomicU32::new(0)); + let queued = tokio::spawn({ + let policy = Arc::clone(&policy); + let cancel = cancel.clone(); + let calls = Arc::clone(&calls); + async move { + execute_with_jitter("cancel_queued", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || { + calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(())) + }) + .await + } + }); + + for _ in 0..100 { + if policy.runtime.queued.available_permits() == 0 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(policy.runtime.queued.available_permits(), 0, "request must be queued"); + cancel.cancel(); + + let result = queued.await.expect("queued task join"); + assert!(matches!(result, Err(KmsError::OperationCancelled { .. })), "got {result:?}"); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert_eq!(policy.runtime.queued.available_permits(), 1); + drop(active); + } + + #[tokio::test(start_paused = true)] + async fn queued_admission_observes_circuit_opened_while_waiting() { + let policy = Arc::new(policy_with_limit(1, 1)); + let active = Arc::clone(&policy.runtime.capacity.total) + .acquire_owned() + .await + .expect("active permit"); + let calls = Arc::new(AtomicU32::new(0)); + let queued = tokio::spawn({ + let policy = Arc::clone(&policy); + let calls = Arc::clone(&calls); + async move { + execute_with_jitter( + "open_while_queued", + OpClass::ReadIdempotent, + &policy, + &CancellationToken::new(), + full_jitter, + move || { + calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(())) + }, + ) + .await + } + }); + + for _ in 0..100 { + if policy.runtime.queued.available_permits() == 0 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(policy.runtime.queued.available_permits(), 0, "request must be queued"); + trip_breaker(&policy.runtime); + drop(active); + + let result = queued.await.expect("queued task join"); + assert!(result.as_ref().is_err_and(is_circuit_open), "got {result:?}"); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert_eq!(policy.runtime.queued.available_permits(), 1); + assert_eq!(policy.runtime.capacity.total.available_permits(), 1); + } + + #[tokio::test(start_paused = true)] + async fn breaker_bounds_attempts_and_recovers_with_one_probe() { + let mut policy = policy_with_limit(2, 2); + policy.max_attempts = 10; + let policy = Arc::new(policy); + let calls = Arc::new(AtomicU32::new(0)); + let cancel = CancellationToken::new(); + let failed: Result<()> = execute_with_jitter("breaker_open", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, { + let calls = Arc::clone(&calls); + move || { + let calls = Arc::clone(&calls); + async move { + calls.fetch_add(1, Ordering::SeqCst); + Err(retryable_conn_error()) + } + } + }) + .await; + failed.expect_err("breaker threshold must stop retries"); + assert_eq!(calls.load(Ordering::SeqCst), 5); + for _ in 0..3 { + let rejected: Result<()> = + execute_with_jitter("breaker_rejected", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || async { + Ok(()) + }) + .await; + assert!(rejected.as_ref().is_err_and(is_circuit_open), "got {rejected:?}"); + } + let already_cancelled = CancellationToken::new(); + already_cancelled.cancel(); + let cancelled: Result<()> = execute_with_jitter( + "breaker_cancelled", + OpClass::ReadIdempotent, + &policy, + &already_cancelled, + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(matches!(cancelled, Err(KmsError::OperationCancelled { .. }))); + + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let probe = tokio::spawn({ + let policy = Arc::clone(&policy); + let calls = Arc::clone(&calls); + let started = Arc::clone(&started); + let release = Arc::clone(&release); + async move { + execute_with_jitter( + "breaker_probe", + OpClass::ReadIdempotent, + &policy, + &CancellationToken::new(), + full_jitter, + move || { + let calls = Arc::clone(&calls); + let started = Arc::clone(&started); + let release = Arc::clone(&release); + async move { + calls.fetch_add(1, Ordering::SeqCst); + started.notify_one(); + release.notified().await; + Ok(()) + } + }, + ) + .await + } + }); + started.notified().await; + let concurrent: Result<()> = execute_with_jitter( + "breaker_concurrent_probe", + OpClass::ReadIdempotent, + &policy, + &cancel, + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(concurrent.as_ref().is_err_and(is_circuit_open), "got {concurrent:?}"); + release.notify_one(); + probe.await.expect("probe join").expect("successful probe must close circuit"); + execute_with_jitter("breaker_recovered", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || async { + Ok(()) + }) + .await + .expect("closed circuit must recover"); + assert_eq!(calls.load(Ordering::SeqCst), 6); + } + + #[tokio::test(start_paused = true)] + async fn breaker_stale_and_failed_probes_preserve_state() { + let runtime = test_runtime(2, 2); + let stale = runtime.admit(Instant::now()).expect("state").expect("closed"); + trip_breaker(&runtime); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + let active = Arc::clone(&runtime.capacity.total).try_acquire_owned().expect("capacity"); + let probe = runtime.admit(Instant::now()).expect("state").expect("probe"); + drop(RuntimePermit::new(Arc::clone(&runtime), probe, active)); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION - Duration::from_millis(1)).await; + assert!(runtime.rejects(Instant::now()).expect("state")); + tokio::time::advance(Duration::from_millis(1)).await; + let probe = runtime.admit(Instant::now()).expect("state").expect("probe"); + assert!(runtime.complete(probe, Some(ErrorClass::RetryableStatus))); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + let recovery = runtime.admit(Instant::now()).expect("state").expect("probe"); + assert!(!runtime.complete(recovery, Some(ErrorClass::Fatal))); + assert!(!runtime.complete(stale, Some(ErrorClass::RetryableConn))); + + let reset = test_runtime(2, 2); + for _ in 0..4 { + let admission = reset.admit(Instant::now()).expect("state").expect("closed"); + assert!(!reset.complete(admission, Some(ErrorClass::RetryableConn))); + } + let fatal = reset.admit(Instant::now()).expect("state").expect("closed"); + assert!(!reset.complete(fatal, Some(ErrorClass::Fatal))); + for _ in 0..4 { + let admission = reset.admit(Instant::now()).expect("state").expect("closed"); + assert!(!reset.complete(admission, Some(ErrorClass::RetryableConn))); + } + } + #[tokio::test(start_paused = true)] async fn total_duration_never_exceeds_deadline() { // Worst case without a deadline would be 5 * 10s + backoff; the 25s @@ -750,7 +1558,7 @@ mod tests { retry_attempts: 50, ..KmsConfig::default() }; - let policy = RetryPolicy::from_config(&config); + let policy = RetryPolicy::for_backend(&config, "test-backend", "https://example.invalid", None, "config-clamp"); assert_eq!(policy.attempt_timeout, Duration::from_secs(300)); assert_eq!(policy.max_attempts, 10); // The deadline must cover the full worst case so it never cuts a @@ -758,7 +1566,7 @@ mod tests { assert!(policy.op_deadline >= policy.attempt_timeout.saturating_mul(policy.max_attempts)); let in_range = KmsConfig::default(); - let policy = RetryPolicy::from_config(&in_range); + let policy = RetryPolicy::for_backend(&in_range, "test-backend", "https://example.invalid", None, "config-in-range"); assert_eq!(policy.attempt_timeout, in_range.timeout); assert_eq!(policy.max_attempts, in_range.retry_attempts); } @@ -865,6 +1673,17 @@ mod tests { .sum() } + fn gauge_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> Option { + snapshot.iter().find_map(|(composite, _unit, _description, value)| { + let matches = + composite.kind() == MetricKind::Gauge && composite.key().name() == name && labels_match(composite.key(), labels); + match (matches, value) { + (true, DebugValue::Gauge(value)) => Some(value.into_inner()), + _ => None, + } + }) + } + fn histogram_values(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> Vec { snapshot .iter() @@ -1116,4 +1935,174 @@ mod tests { vec![1.0] ); } + + #[test] + fn metrics_record_backpressure_and_circuit_outcomes() { + let (snapshot, ()) = record_metrics(|| { + Box::pin(async { + let rejected = policy_with_limit(1, 0); + let rejected_active = Arc::clone(&rejected.runtime.capacity.total) + .try_acquire_owned() + .expect("active capacity"); + let result: Result<()> = execute_with_jitter( + "metrics_backpressure_rejected", + OpClass::ReadIdempotent, + &rejected, + &CancellationToken::new(), + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(matches!(result, Err(KmsError::BackendError { .. }))); + drop(rejected_active); + + let mut timed_out = policy_with_limit(1, 1); + timed_out.op_deadline = Duration::from_millis(100); + let timeout_active = Arc::clone(&timed_out.runtime.capacity.total) + .try_acquire_owned() + .expect("active capacity"); + let result: Result<()> = execute_with_jitter( + "metrics_backpressure_timeout", + OpClass::ReadIdempotent, + &timed_out, + &CancellationToken::new(), + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(matches!(result, Err(KmsError::OperationTimedOut { .. }))); + drop(timeout_active); + + let open = policy_with_limit(1, 1); + trip_breaker(&open.runtime); + let result: Result<()> = execute_with_jitter( + "metrics_circuit_open", + OpClass::ReadIdempotent, + &open, + &CancellationToken::new(), + full_jitter, + || async { Ok(()) }, + ) + .await; + assert!(result.as_ref().is_err_and(is_circuit_open), "got {result:?}"); + + let mut trip = policy_with_limit(1, 1); + trip.max_attempts = DEFAULT_CIRCUIT_FAILURE_THRESHOLD + 1; + let result: Result<()> = execute_with_jitter( + "metrics_circuit_trip", + OpClass::ReadIdempotent, + &trip, + &CancellationToken::new(), + full_jitter, + || async { Err(retryable_conn_error()) }, + ) + .await; + assert!(matches!(result, Err(KmsError::BackendError { .. }))); + }) + }); + + for (operation, outcome, attempts) in [ + ("metrics_backpressure_rejected", "backpressure_rejected", 0.0), + ("metrics_backpressure_timeout", "backpressure_timeout", 0.0), + ("metrics_circuit_open", "circuit_open", 0.0), + ("metrics_circuit_trip", "circuit_open", f64::from(DEFAULT_CIRCUIT_FAILURE_THRESHOLD)), + ] { + assert_eq!( + counter_value(&snapshot, METRIC_OPERATIONS_TOTAL, &[("operation", operation), ("outcome", outcome)]), + 1 + ); + assert_eq!( + histogram_values(&snapshot, METRIC_OPERATION_ATTEMPTS, &[("operation", operation), ("outcome", outcome)]), + vec![attempts] + ); + } + } + + #[test] + fn successful_and_fatal_half_open_probes_clear_the_open_gauge() { + for failure in [None, Some(ErrorClass::Fatal)] { + let (snapshot, runtime) = record_metrics(move || { + Box::pin(async move { + let runtime = test_runtime(1, 1); + trip_breaker(&runtime); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + let probe = runtime.admit(Instant::now()).expect("state").expect("half-open probe"); + assert!(!runtime.complete(probe, failure)); + runtime + }) + }); + + assert!(!runtime.rejects(Instant::now()).expect("closed breaker state")); + assert_eq!( + gauge_value(&snapshot, METRIC_CIRCUIT_OPEN, &[("backend", "test-backend"), ("scope", "test-scope")]), + Some(0.0) + ); + } + } + + #[test] + fn cancelling_half_open_attempt_releases_capacity_and_reopens_breaker() { + let (snapshot, policy) = record_metrics(|| { + Box::pin(async { + let policy = policy_with_limit(1, 1); + trip_breaker(&policy.runtime); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + + let cancel = CancellationToken::new(); + let canceller = cancel.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(1)).await; + canceller.cancel(); + }); + let result: Result<()> = + execute_with_jitter("cancel_half_open_probe", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || { + std::future::pending::>() + }) + .await; + + assert!(matches!(result, Err(KmsError::OperationCancelled { .. }))); + assert_eq!(policy.runtime.capacity.total.available_permits(), 1); + assert_eq!(policy.runtime.queued.available_permits(), 1); + assert!(policy.runtime.rejects(Instant::now()).expect("reopened breaker state")); + tokio::time::advance(DEFAULT_CIRCUIT_OPEN_DURATION).await; + let probe = policy + .runtime + .admit(Instant::now()) + .expect("state") + .expect("replacement half-open probe"); + assert!(probe.probe); + policy + }) + }); + + let labels = [("backend", "test-backend"), ("scope", "test-scope")]; + assert_eq!(gauge_value(&snapshot, METRIC_IN_FLIGHT, &labels), Some(0.0)); + assert_eq!(gauge_value(&snapshot, METRIC_CIRCUIT_OPEN, &labels), Some(1.0)); + drop(policy); + } + + #[test] + fn runtime_metrics_track_in_flight_and_open_lifecycle() { + let in_flight_recorder = DebuggingRecorder::new(); + let in_flight_snapshotter = in_flight_recorder.snapshotter(); + let in_flight = metrics::with_local_recorder(&in_flight_recorder, || { + let runtime = test_runtime(1, 1); + let active = Arc::clone(&runtime.capacity.total).try_acquire_owned().expect("capacity"); + let admission = runtime.admit(Instant::now()).expect("state").expect("closed"); + let _permit = RuntimePermit::new(Arc::clone(&runtime), admission, active); + in_flight_snapshotter.snapshot().into_vec() + }); + + let dropped_recorder = DebuggingRecorder::new(); + let dropped_snapshotter = dropped_recorder.snapshotter(); + let dropped = metrics::with_local_recorder(&dropped_recorder, || { + let runtime = test_runtime(1, 1); + trip_breaker(&runtime); + drop(runtime); + dropped_snapshotter.snapshot().into_vec() + }); + let labels = [("backend", "test-backend"), ("scope", "test-scope")]; + assert_eq!(gauge_value(&in_flight, METRIC_IN_FLIGHT, &labels), Some(1.0)); + assert_eq!(gauge_value(&dropped, METRIC_CIRCUIT_OPEN, &labels), Some(0.0)); + } } diff --git a/deploy/observability/grafana/rustfs-kms-observability.json b/deploy/observability/grafana/rustfs-kms-observability.json index 0810a3e4a..0b874cf41 100644 --- a/deploy/observability/grafana/rustfs-kms-observability.json +++ b/deploy/observability/grafana/rustfs-kms-observability.json @@ -21,7 +21,7 @@ } ] }, - "description": "KMS backend operation metrics emitted at the operation-policy choke point (crates/kms/src/policy.rs). All label values are static enum strings; key identifiers, key material, and tokens never appear in labels. These metrics do not carry the RustFS `server` label — use your scrape topology (job/instance or promoted OTel resource attributes) to split by node. Alert response procedures: docs/operations/kms-observability-runbook.md.", + "description": "KMS backend operation metrics emitted at the operation-policy choke point (crates/kms/src/policy.rs). All label values are bounded enums or fixed call-site tokens; key identifiers, key material, and tokens never appear in labels. These metrics do not carry the RustFS `server` label — use your scrape topology (job/instance or promoted OTel resource attributes) to split by node. Alert response procedures: docs/operations/kms-observability-runbook.md.", "editable": true, "fiscalYearStartMonth": 0, "graphTooltip": 1, @@ -34,7 +34,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "description": "Terminal outcomes of KMS backend operations. `fatal` means a non-retryable failure ended the operation on first observation; `budget_exhausted` and `deadline_exceeded` mean retries ran out; `cancelled` is normal during shutdown.", + "description": "Terminal outcomes of KMS backend operations. `fatal` is non-retryable; `budget_exhausted` and `deadline_exceeded` mean retry limits ran out; `backpressure_timeout` and `backpressure_rejected` are admission failures; `circuit_open` means the breaker opened or rejected the operation; `cancelled` is normal during shutdown.", "fieldConfig": { "defaults": { "color": { @@ -218,7 +218,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "description": "Share of operations that terminated in fatal, budget_exhausted, or deadline_exceeded. The cancelled outcome is plotted separately because shutdown windows legitimately spike it. The ratio is meaningless at near-zero traffic — read it together with the operation rate panels.", + "description": "Share of operations that terminated in fatal, budget_exhausted, deadline_exceeded, backpressure_timeout, backpressure_rejected, or circuit_open. The cancelled outcome is plotted separately because shutdown windows legitimately spike it. The ratio is meaningless at near-zero traffic — read it together with the operation rate panels.", "fieldConfig": { "defaults": { "color": { diff --git a/docs/architecture/global-state-inventory.md b/docs/architecture/global-state-inventory.md index eacb3e708..a594bc569 100644 --- a/docs/architecture/global-state-inventory.md +++ b/docs/architecture/global-state-inventory.md @@ -78,6 +78,7 @@ of reaching across module boundaries. | `DRIVE_TIMEOUT_PROFILE_CACHE`, `DRIVE_TIMEOUT_HEALTH_POLICY_CACHE` | `crates/ecstore/src/disk/disk_store.rs` | Cache or constant / owner-local cache | Drive timeout environment caches stay local to the disk-store owner. | | `TIER_FREE_VERSION_RECOVERY_STARTED`, `TIER_DELETE_JOURNAL_RECOVERY_STARTED` | `crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs` | Cache or constant / owner-local static guard | Lifecycle recovery single-run guards stay local to lifecycle operations. | | `REMOTE_DELETE_INFLIGHT`, `REMOTE_DELETE_LIMITER`, `REMOTE_DELETE_BREAKER`, `REMOTE_TIER_DELETE_TEST_HOOK` | `crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs` | Cache or constant / owner-local static guard | Remote tier delete concurrency, breaker, and test hook state stay local to the tier sweeper owner. | +| `ACTIVE_REGISTRY`, `BackendCapacity` | `crates/kms/src/policy.rs` | Process-global owner-local admission capacity registry | KMS policy generations share only active semaphore capacity by backend identity; each generation owns fresh bounded queues and circuit breakers. Callers access this state only through `RetryPolicy`. | ## RustFS Owner-Local Static Inventory diff --git a/docs/operations/kms-observability-runbook.md b/docs/operations/kms-observability-runbook.md index 30bf9019a..28b3b4744 100644 --- a/docs/operations/kms-observability-runbook.md +++ b/docs/operations/kms-observability-runbook.md @@ -4,11 +4,11 @@ This runbook covers the KMS metrics, the Grafana dashboard that visualizes them, ## Metric reference -Across every family below, label values are exclusively static enum strings — key identifiers, key material, ciphertext, paths, and tokens never appear in metric labels, and any change that would add such a label is a regression. +Across every family below, label values come from bounded enums or fixed call-site tokens — key identifiers, key material, ciphertext, paths, and tokens never appear in metric labels, and any change that would add such a label is a regression. ### Backend operation metrics -All four are emitted at the single operation-policy choke point (`crates/kms/src/policy.rs`) that every instrumented KMS backend call flows through. +All six are emitted at the single operation-policy choke point (`crates/kms/src/policy.rs`) that every instrumented KMS backend call flows through. | Metric | Type | Labels | Meaning | | --- | --- | --- | --- | @@ -16,15 +16,19 @@ All four are emitted at the single operation-policy choke point (`crates/kms/src | `rustfs_kms_backend_attempt_failures_total` | counter | `operation`, `error_class` | Individual failed attempts, including attempts the retry policy later absorbed | | `rustfs_kms_backend_operation_duration_seconds` | histogram | `operation`, `outcome` | Wall-clock duration of a whole operation, including retries and backoff sleeps | | `rustfs_kms_backend_operation_attempts` | histogram | `operation`, `outcome` | Number of attempts one operation used before completing | +| `rustfs_kms_backend_in_flight` | gauge | `backend`, `scope` | External backend attempts currently in flight after admission | +| `rustfs_kms_backend_circuit_open` | gauge | `backend`, `scope` | Open or half-open circuits; `0` means closed | Label values: -- `outcome`: `success`, `fatal` (a non-retryable failure ended the operation on first observation), `budget_exhausted` (the attempt budget ran out on retryable failures), `deadline_exceeded` (the operation deadline ran out before another attempt could complete), `cancelled` (shutdown or caller cancellation). +- `outcome`: `success`, `fatal` (a non-retryable failure ended the operation on first observation), `budget_exhausted` (the attempt budget ran out on retryable failures), `deadline_exceeded` (the operation deadline ran out before another attempt could complete), `backpressure_timeout` (the deadline elapsed before capacity admission completed), `backpressure_rejected` (active capacity and the bounded queue were full or unavailable), `circuit_open` (a retryable failure opened the breaker or an open breaker rejected the operation), `cancelled` (shutdown or caller cancellation). - `op_class`: `read_idempotent` (safe to retry), `mutating_non_idempotent` (never replayed — a retryable failure terminates after a single attempt because the server may have processed the request), `auth` (login and token renewal). - `error_class`: `retryable_conn` (connection-level failure: dial, TLS, broken connection), `retryable_status` (retryable backend status, e.g. Vault 5xx or a sealed Vault's 503), `attempt_timeout` (the per-attempt timeout cut the attempt off; retried like a connection failure because the server may still have processed the request), `fatal` (non-retryable: authentication, permissions, malformed request, missing key or version). - `operation`: static per-call-site names, e.g. `vault_kv2_read_key_version`, `vault_kv2_cas_write_key`, `vault_transit_encrypt`, `vault_transit_decrypt`, `vault_login`, `vault_token_renew`. -Instrumentation boundary: the Local and Static backends do not flow through the choke point and emit no operation metrics; bringing them under the same instrumentation is tracked separately (rustfs/backlog#1569). Absence of these four series on a cluster using those backends is expected, not an outage. The families below sit above the backend layer and are emitted regardless. +Admission sharing follows two different boundaries. Total active backend capacity is shared by backend identity and capped at 64; ordinary operations are limited to 63 so login and renewal always retain one reserved slot without exceeding the total cap. Each backend configuration generation owns fresh bounded queues and circuit breakers for its policy scopes, so a failed reconfiguration candidate cannot inherit or mutate the running generation's admission state. + +Instrumentation boundary: the Local and Static backends do not flow through the choke point and emit no operation metrics; bringing them under the same instrumentation is tracked separately (rustfs/backlog#1569). Absence of these six series on a cluster using those backends is expected, not an outage. The families below sit above the backend layer and are emitted regardless. ### Key metadata cache metrics @@ -116,14 +120,16 @@ Related signals: the "Attempt Failure Rate by Error Class" and "Backend Operatio ### KmsBackendHighErrorRate -Meaning: more than 5% of KMS operations are terminating without success (`fatal`, `budget_exhausted`, or `deadline_exceeded`; `cancelled` is excluded because shutdown windows legitimately produce it). A traffic guard suppresses the alert below ~0.02 ops/s so a single failure on a near-idle cluster does not page. +Meaning: more than 5% of KMS operations are terminating without success (`fatal`, `budget_exhausted`, `deadline_exceeded`, `backpressure_timeout`, `backpressure_rejected`, or `circuit_open`; `cancelled` is excluded because shutdown windows legitimately produce it). A traffic guard suppresses the alert below ~0.02 ops/s so a single failure on a near-idle cluster does not page. Investigation: 1. Break the failures down by outcome: `sum by (outcome) (rate(rustfs_kms_backend_operations_total{outcome!~"success|cancelled"}[5m]))`. 2. If `fatal` dominates, follow [KmsBackendFatalErrors](#kmsbackendfatalerrors). 3. If `budget_exhausted` or `deadline_exceeded` dominates, follow [KmsBackendRetryBudgetExhausted](#kmsbackendretrybudgetexhausted) — the backend is unavailable or too slow for longer than the retry policy can bridge. -4. Correlate with client impact: encrypted-object PUT/GET failures and S3 error rates on buckets with encryption configured. +4. If `backpressure_timeout` or `backpressure_rejected` dominates, compare `rustfs_kms_backend_in_flight` by `backend` and `scope`; total active capacity is shared by backend identity, one slot is reserved for credential refresh, and each configuration generation has fresh scope-local bounded queues. +5. If `circuit_open` dominates, follow [KmsBackendCircuitOpen](#kmsbackendcircuitopen). +6. Correlate with client impact: encrypted-object PUT/GET failures and S3 error rates on buckets with encryption configured. Related signals: the "Non-Success Outcome Ratio" dashboard panel; the KMS-related warnings listed under the other alerts in this runbook. @@ -170,9 +176,23 @@ Investigation: Related signals: the "Backend Operation Rate by Outcome" panel; retry-backoff warnings in RustFS logs; Vault availability monitoring. +### KmsBackendCircuitOpen + +Meaning: `rustfs_kms_backend_circuit_open` has remained above `0` for a `backend` and `scope` for one minute. This direct gauge alert does not depend on operation traffic: it remains visible when the circuit is open and rejecting calls, and while the single half-open recovery probe is running. + +Investigation: + +1. Identify the affected scope with `rustfs_kms_backend_circuit_open > 0`. +2. Break recent rejections down by operation: `sum by (operation) (rate(rustfs_kms_backend_operations_total{outcome="circuit_open"}[5m]))`. +3. Check `sum by (error_class) (rate(rustfs_kms_backend_attempt_failures_total{error_class=~"retryable_conn|retryable_status|attempt_timeout"}[5m]))` to distinguish transport failures, retryable backend responses such as a sealed Vault, and attempt timeouts. An attempt timeout counts toward the breaker as a retryable connection failure. +4. After the open interval, the next eligible operation is the only half-open probe. A success or non-retryable failure closes the circuit; a retryable failure reopens it. A non-retryable probe still fails as `fatal`, so follow [KmsBackendFatalErrors](#kmsbackendfatalerrors) even after the circuit gauge clears. Do not restart RustFS just to clear the state. +5. Remember the sharing boundary: each configuration generation has fresh scope-local breaker and queue state, while total active capacity is shared by backend identity with one slot reserved for credential refresh. Check other scopes for capacity pressure even when their circuits remain closed. + +Related signals: `circuit_open`, `backpressure_timeout`, and `backpressure_rejected` on the "Backend Operation Rate by Outcome" panel; `rustfs_kms_backend_in_flight`; Vault availability and seal status. + ## Threshold calibration -Every numeric threshold in `rustfs-kms-alerts.yml` (5% error ratio, 2s p99, 0.5/s attempt failures, 0.05/s budget exhaustion) is a conservative default chosen without a production baseline, biased toward not paging on healthy-but-busy systems. Before relying on these alerts for paging: run the workload in staging for at least a week, record the steady-state values of the expressions above, then tighten thresholds to sit clearly above observed peaks. Once a stable baseline exists, consider converting `KmsBackendAttemptFailureSpike` to a baseline-relative form (`offset 1d` ratio, see `.docker/observability/prometheus-rules/rustfs-get-optimization-alerts.yaml` for the pattern). Formal SLO targets for KMS operations are deliberately out of scope until that baseline exists (rustfs/backlog#1584). +Every numeric traffic or latency threshold in `rustfs-kms-alerts.yml` (5% error ratio, 2s p99, 0.5/s attempt failures, 0.05/s budget exhaustion) is a conservative default chosen without a production baseline, biased toward not paging on healthy-but-busy systems. Before relying on these alerts for paging: run the workload in staging for at least a week, record the steady-state values of the expressions above, then tighten thresholds to sit clearly above observed peaks. `KmsBackendCircuitOpen` is different: its gauge is direct state, and the one-minute hold only suppresses a circuit that recovers immediately. Once a stable baseline exists, consider converting `KmsBackendAttemptFailureSpike` to a baseline-relative form (`offset 1d` ratio, see `.docker/observability/prometheus-rules/rustfs-get-optimization-alerts.yaml` for the pattern). Formal SLO targets for KMS operations are deliberately out of scope until that baseline exists (rustfs/backlog#1584). ## Coverage gaps