diff --git a/crates/kms/src/backends/vault_credentials.rs b/crates/kms/src/backends/vault_credentials.rs index 066e0bb3a..c5a300908 100644 --- a/crates/kms/src/backends/vault_credentials.rs +++ b/crates/kms/src/backends/vault_credentials.rs @@ -57,6 +57,42 @@ const DEFAULT_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(5); /// Default seconds between token file re-reads for [`TokenFileSource`]. const DEFAULT_TOKEN_FILE_POLL_INTERVAL_SECS: u64 = 30; +// --------------------------------------------------------------------------- +// Metrics +// +// Both gauges describe the one credential generation currently installed, so +// they carry no labels: the Vault address, mount, auth path and token are all +// off limits as label values, and there is exactly one generation to describe. +// The renewal loop republishes them on a bounded cadence while it waits, so a +// scrape landing between refresh cycles never reads a TTL frozen at the last +// refresh, or a fail-closed state that flipped after it. +// --------------------------------------------------------------------------- + +/// Gauge: seconds left before the active Vault token expires; `0` once it has. +const METRIC_TOKEN_TTL_SECONDS: &str = "rustfs_kms_vault_token_ttl_seconds"; +/// Gauge: `1` while [`VaultCredentialProvider::current`] refuses to hand out +/// the token because it is inside the fail-closed safety window, `0` otherwise. +const METRIC_CREDENTIALS_FAIL_CLOSED: &str = "rustfs_kms_vault_credentials_fail_closed"; + +/// How often the renewal loop republishes the credential gauges while waiting. +/// Bounds how stale a scrape can be, without any additional Vault traffic. +const CREDENTIAL_GAUGE_INTERVAL: Duration = Duration::from_secs(10); + +/// Register metric descriptions once per process. +fn describe_credential_metrics() { + static DESCRIBE: std::sync::Once = std::sync::Once::new(); + DESCRIBE.call_once(|| { + metrics::describe_gauge!( + METRIC_TOKEN_TTL_SECONDS, + "Seconds remaining before the Vault token backing the KMS backend expires" + ); + metrics::describe_gauge!( + METRIC_CREDENTIALS_FAIL_CLOSED, + "1 while the Vault credential provider refuses to serve its token because the token is inside the fail-closed safety window" + ); + }); +} + /// A crate-owned secret value, zeroized on drop and redacted in Debug output. #[derive(Clone, Zeroize, ZeroizeOnDrop)] pub(crate) struct SecretString(String); @@ -627,6 +663,45 @@ impl VaultCredentialProvider { Ok(handle) } + /// Publish the credential gauges for the generation currently installed. + /// + /// The fail-closed gauge re-evaluates the very gate + /// [`VaultCredentialProvider::current`] applies, so what operators see and + /// what the request path does cannot drift apart. + fn record_credential_gauges(&self) { + let handle = self.current.load(); + let now = Instant::now(); + let fail_closed = match handle.expires_at() { + Some(expires_at) => { + metrics::gauge!(METRIC_TOKEN_TTL_SECONDS).set(expires_at.saturating_duration_since(now).as_secs_f64()); + now + self.policy.safety_window >= expires_at + } + // A generation without an expiry has no remaining TTL to report + // and can never lapse, so it can never fail closed either. + None => false, + }; + metrics::gauge!(METRIC_CREDENTIALS_FAIL_CLOSED).set(if fail_closed { 1.0 } else { 0.0 }); + } + + /// Wait until `deadline`, republishing the credential gauges on the + /// observation cadence. Reports `false` when cancellation cut the wait + /// short. + async fn wait_publishing_gauges(&self, deadline: Instant, cancel: &CancellationToken) -> bool { + loop { + self.record_credential_gauges(); + let now = Instant::now(); + if now >= deadline { + return true; + } + let slice = (deadline - now).min(CREDENTIAL_GAUGE_INTERVAL); + tokio::select! { + biased; + _ = cancel.cancelled() => return false, + _ = tokio::time::sleep(slice) => {} + } + } + } + /// Refresh the credentials if generation `observed` is still current. /// /// Single-flight: concurrent callers serialize on the refresh lock, and a @@ -711,17 +786,22 @@ impl fmt::Debug for VaultCredentialProvider { /// again immediately (the renewal point is already in the past), so the /// provider keeps trying to recover even after the fail-closed window has /// been reached. +/// +/// Both waits run through [`VaultCredentialProvider::wait_publishing_gauges`], +/// which is the only place the credential gauges are published: the loop is +/// already the component that tracks token expiry, and doing it here keeps the +/// request path free of any metric work. async fn renewal_loop(provider: Arc, cancel: CancellationToken) { + describe_credential_metrics(); loop { let handle = provider.snapshot(); let Some(renew_at) = handle.renew_at() else { // The current generation never expires; nothing left to schedule. + provider.record_credential_gauges(); return; }; - tokio::select! { - biased; - _ = cancel.cancelled() => return, - _ = tokio::time::sleep_until(renew_at) => {} + if !provider.wait_publishing_gauges(renew_at, &cancel).await { + return; } match provider.refresh(handle.generation, &cancel).await { @@ -733,10 +813,9 @@ async fn renewal_loop(provider: Arc, cancel: Cancellati error = %error, "Vault credential refresh failed; retrying until the credentials recover" ); - tokio::select! { - biased; - _ = cancel.cancelled() => return, - _ = tokio::time::sleep(provider.policy.retry_interval) => {} + let retry_at = Instant::now() + provider.policy.retry_interval; + if !provider.wait_publishing_gauges(retry_at, &cancel).await { + return; } } } @@ -1354,4 +1433,125 @@ mod tests { assert!(!rendered.contains(TEST_TOKEN), "debug output must not leak the vault token: {rendered}"); assert!(rendered.contains("vault-token"), "the file path is not a secret"); } + + // -- Metric emission ---------------------------------------------------- + // + // Each test installs a thread-local debugging recorder and drives a + // paused-clock current-thread runtime inside it, so the renewal loop's + // gauge publications land on exact virtual timestamps. + + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + type MetricEntry = ( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + ); + + /// Run `test` on a paused current-thread runtime under a debugging + /// recorder and return one snapshot of everything it emitted. + /// + /// A single snapshot per test on purpose: `Snapshotter::snapshot` drains + /// the recorded state, so taking it per assertion would only show the + /// first assertion any data. + fn record_metrics( + test: impl FnOnce() -> std::pin::Pin>>, + ) -> (Vec, Out) { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let out = metrics::with_local_recorder(&recorder, || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(true) + .build() + .expect("current-thread runtime must build"); + runtime.block_on(test()) + }); + (snapshotter.snapshot().into_vec(), out) + } + + /// Last value of a gauge, plus the labels it was published with. + fn gauge(snapshot: &[MetricEntry], name: &str) -> Option<(f64, Vec)> { + snapshot.iter().find_map(|(composite, _unit, _description, value)| { + let matches = composite.kind() == MetricKind::Gauge && composite.key().name() == name; + match (matches, value) { + (true, DebugValue::Gauge(value)) => Some(( + value.into_inner(), + composite.key().labels().map(|label| label.key().to_string()).collect(), + )), + _ => None, + } + }) + } + + #[test] + fn renewal_loop_republishes_token_ttl_while_it_waits() { + let (snapshot, ()) = record_metrics(|| { + Box::pin(async { + let (provider, _state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + let task = provider.spawn_renewal_task().expect("lease-bound tokens need renewal"); + + // Well inside the first half of the lease: nothing has been + // renewed yet, so only the observation cadence can have moved + // the gauge off its initial 60s. + tokio::time::sleep(Duration::from_secs(25)).await; + task.shutdown().await; + }) + }); + + let (ttl, labels) = gauge(&snapshot, METRIC_TOKEN_TTL_SECONDS).expect("token TTL gauge must be published"); + assert!( + (ttl - 40.0).abs() < 1.0, + "expected ~40s left of a 60s lease at the last observation, got {ttl}" + ); + assert!(labels.is_empty(), "credential gauges must carry no labels, got {labels:?}"); + assert_eq!( + gauge(&snapshot, METRIC_CREDENTIALS_FAIL_CLOSED).map(|(value, _)| value), + Some(0.0), + "a token outside its safety window must not report fail-closed" + ); + } + + #[test] + fn fail_closed_gauge_tracks_the_gate_the_request_path_applies() { + let (snapshot, refused) = record_metrics(|| { + Box::pin(async { + let (provider, state) = scripted_provider( + Duration::from_secs(60), + true, + test_policy(Duration::from_secs(10), Duration::from_secs(5)), + ) + .await; + state.fail_renew.store(true, Ordering::SeqCst); + state.fail_login.store(true, Ordering::SeqCst); + let task = provider.spawn_renewal_task().expect("renewal task"); + + // 60s lease minus the 10s safety window: by 51s the provider + // is refusing the token, and the gauge must already say so. + tokio::time::sleep(Duration::from_secs(51)).await; + let refused = provider.current().is_err(); + task.shutdown().await; + refused + }) + }); + + assert!(refused, "a token inside the safety window must be refused"); + assert_eq!( + gauge(&snapshot, METRIC_CREDENTIALS_FAIL_CLOSED).map(|(value, _)| value), + Some(1.0), + "the gauge must report the same fail-closed state the request path enforces" + ); + let (ttl, _labels) = gauge(&snapshot, METRIC_TOKEN_TTL_SECONDS).expect("token TTL gauge must be published"); + assert!( + (ttl - 10.0).abs() < 1.0, + "expected the TTL gauge to track the lease down into its safety window, got {ttl}" + ); + } } diff --git a/crates/kms/src/deletion_worker.rs b/crates/kms/src/deletion_worker.rs index 49594499a..5edd89dc5 100644 --- a/crates/kms/src/deletion_worker.rs +++ b/crates/kms/src/deletion_worker.rs @@ -25,7 +25,7 @@ use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink}; use crate::backends::{ExpiredKeyRemoval, KmsBackend}; use crate::error::Result; -use crate::types::{KeyStatus, ListKeysRequest, OperationContext}; +use crate::types::{KeyInfo, KeyStatus, ListKeysRequest, OperationContext}; use async_trait::async_trait; use jiff::Zoned; use std::sync::Arc; @@ -36,6 +36,109 @@ use tracing::{debug, info, warn}; /// How often the worker looks for expired pending deletions. pub const DEFAULT_SWEEP_INTERVAL: Duration = Duration::from_secs(60); +// --------------------------------------------------------------------------- +// Metrics +// +// The sweep already pages through the whole key set, so everything below is +// derived from the pages it has in hand: observing the key lifecycle costs no +// extra backend call. Each metric is an aggregate — the gauges carry no labels +// at all and the counter only a static outcome — because a per-key label would +// carry key identifiers into the metric stream and grow the series count with +// the key set. "This key is overdue for rotation" is a threshold on the +// aggregate age and belongs in an alerting rule, not in a label. +// --------------------------------------------------------------------------- + +/// Gauge: keys scheduled for deletion whose deadline has not passed, as of the +/// end of the last sweep that saw the whole key set. +const METRIC_PENDING_DELETION_KEYS: &str = "rustfs_kms_pending_deletion_keys"; +/// Gauge: keys left tombstoned by an interrupted removal and still awaiting +/// the sweep, as of the end of the last sweep that saw the whole key set. +const METRIC_TOMBSTONE_KEYS: &str = "rustfs_kms_deletion_tombstone_keys"; +/// Gauge: seconds since the least recently rotated usable key was rotated +/// (its creation time when it was never rotated); `0` when there are none. +const METRIC_OLDEST_ROTATION_AGE_SECONDS: &str = "rustfs_kms_oldest_key_rotation_age_seconds"; +/// Counter: keys the sweep acted on, by `outcome` (`removed`, `blocked`, +/// `skipped`, `failed`). +const METRIC_SWEEP_KEYS_TOTAL: &str = "rustfs_kms_deletion_sweep_keys_total"; + +/// Register metric descriptions once per process. +fn describe_metrics() { + static DESCRIBE: std::sync::Once = std::sync::Once::new(); + DESCRIBE.call_once(|| { + metrics::describe_gauge!( + METRIC_PENDING_DELETION_KEYS, + "KMS keys scheduled for deletion whose deadline has not passed yet" + ); + metrics::describe_gauge!( + METRIC_TOMBSTONE_KEYS, + "KMS keys left tombstoned by an interrupted removal, still awaiting the deletion sweep" + ); + metrics::describe_gauge!( + METRIC_OLDEST_ROTATION_AGE_SECONDS, + "Seconds since the least recently rotated usable KMS key was last rotated, counting from creation for keys that were never rotated" + ); + metrics::describe_counter!(METRIC_SWEEP_KEYS_TOTAL, "Total keys acted on by the KMS deletion sweep, by outcome"); + }); +} + +/// Aggregate view of the key set, accumulated while the sweep pages through it. +/// +/// Counts and one maximum age only: these feed label-less gauges, so no key +/// identifier can reach the metric stream through them. +#[derive(Debug, Default, Clone, Copy, PartialEq)] +struct KeyCensus { + pending_deletion: usize, + tombstones: usize, + /// Longest time since last rotation across keys that are still usable, + /// counting from creation for keys that were never rotated. Keys on their + /// way out are excluded: they will never be rotated again, and would + /// otherwise pin the gauge high until the sweep finishes removing them. + oldest_rotation_age_seconds: f64, +} + +impl KeyCensus { + fn observe(&mut self, key: &KeyInfo, now: &Zoned) { + match key.status { + KeyStatus::PendingDeletion => self.pending_deletion += 1, + KeyStatus::Deleted => self.tombstones += 1, + KeyStatus::Active | KeyStatus::Disabled => { + let rotated_at = key.rotated_at.as_ref().unwrap_or(&key.created_at); + self.oldest_rotation_age_seconds = self.oldest_rotation_age_seconds.max(seconds_between(rotated_at, now)); + } + } + } +} + +/// Seconds from `earlier` to `now`, clamped at zero so clock skew (or a +/// timestamp persisted by a node running ahead) cannot produce a negative age. +fn seconds_between(earlier: &Zoned, now: &Zoned) -> f64 { + (now.timestamp().as_second() - earlier.timestamp().as_second()).max(0) as f64 +} + +/// Publish one sweep's counters, plus the lifecycle gauges when `census` covers +/// the whole key set. +fn record_sweep(report: &SweepReport, census: Option) { + describe_metrics(); + for (outcome, count) in [ + ("removed", report.removed.len()), + ("blocked", report.blocked.len()), + ("skipped", report.skipped), + ("failed", report.failed), + ] { + // Emitted even at zero so every outcome series exists from the first + // sweep on and a rate over it is defined. + metrics::counter!(METRIC_SWEEP_KEYS_TOTAL, "outcome" => outcome).increment(count as u64); + } + + // A sweep that could not finish listing saw only part of the key set; + // publishing its counts would understate every gauge, so the previous + // (complete) values are left standing instead. + let Some(census) = census else { return }; + metrics::gauge!(METRIC_PENDING_DELETION_KEYS).set(census.pending_deletion as f64); + metrics::gauge!(METRIC_TOMBSTONE_KEYS).set(census.tombstones as f64); + metrics::gauge!(METRIC_OLDEST_ROTATION_AGE_SECONDS).set(census.oldest_rotation_age_seconds); +} + /// Reports configuration that still references a KMS key. /// /// Consulted before any material is destroyed; a non-empty result blocks the @@ -127,10 +230,15 @@ impl DeletionWorker { /// Run one sweep at the given time. Exposed separately so tests can drive /// the expiry logic deterministically. + /// + /// Also feeds the lifecycle gauges: every key it lists is counted into a + /// [`KeyCensus`] on the way past, so the observability comes out of the + /// pages the sweep already had to fetch. pub(crate) async fn sweep(&self, now: &Zoned) -> SweepReport { let mut report = SweepReport::default(); + let mut census = KeyCensus::default(); let mut marker: Option = None; - loop { + let listed_everything = loop { let request = ListKeysRequest { limit: Some(100), marker: marker.clone(), @@ -142,40 +250,52 @@ impl DeletionWorker { Err(error) => { warn!(%error, "KMS deletion sweep could not list keys"); report.failed += 1; - return report; + break false; } }; for key in &response.keys { - if matches!(key.status, KeyStatus::PendingDeletion | KeyStatus::Deleted) { - self.process_key(&key.key_id, now, &mut report).await; + // Keys this sweep destroys are left out of the census: the + // gauges describe the key set as it stands once the sweep is + // done, not as it was when the page was listed. + let removed = if matches!(key.status, KeyStatus::PendingDeletion | KeyStatus::Deleted) { + self.process_key(&key.key_id, now, &mut report).await + } else { + false + }; + if !removed { + census.observe(key, now); } } if !response.truncated { - break; + break true; } match response.next_marker { Some(next_marker) => marker = Some(next_marker), - None => break, + // Truncated without a marker: the rest of the key set is out + // of reach, so the census is incomplete. + None => break false, } - } + }; + record_sweep(&report, listed_everything.then_some(census)); report } - async fn process_key(&self, key_id: &str, now: &Zoned, report: &mut SweepReport) { + /// Handle one key that is on its way out; reports whether it was removed. + async fn process_key(&self, key_id: &str, now: &Zoned, report: &mut SweepReport) -> bool { // Never remove a key that live configuration still points at. The // default key check is built in; broader references (bucket // encryption settings, ...) come from the injected checker. if self.default_key_id.as_deref() == Some(key_id) { warn!(key_id, "expired KMS key is still the default key; refusing removal"); report.blocked.push(key_id.to_string()); - return; + return false; } if let Some(checker) = &self.reference_checker { let references = checker.references(key_id).await; if !references.is_empty() { warn!(key_id, ?references, "expired KMS key is still referenced; refusing removal"); report.blocked.push(key_id.to_string()); - return; + return false; } } @@ -187,12 +307,17 @@ impl DeletionWorker { Ok(ExpiredKeyRemoval::Removed) => { report.removed.push(key_id.to_string()); self.audit_removal(key_id, started, &outcome); + true + } + Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => { + report.skipped += 1; + false } - Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => report.skipped += 1, Err(error) => { warn!(key_id, %error, "failed to remove expired KMS key; will retry next sweep"); report.failed += 1; self.audit_removal(key_id, started, &outcome); + false } } } @@ -479,4 +604,156 @@ mod tests { cancel.cancel(); task.await.expect("worker task must stop after cancellation"); } + + // -- Metric emission ---------------------------------------------------- + + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + type MetricEntry = ( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + ); + + /// Run `test` on a current-thread runtime under a debugging recorder and + /// return one snapshot of everything it emitted. + /// + /// A single snapshot per test on purpose: `Snapshotter::snapshot` drains + /// the recorded state, so taking it per assertion would only show the + /// first assertion any data. + fn record_metrics( + test: impl FnOnce() -> std::pin::Pin>>, + ) -> (Vec, Out) { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let out = metrics::with_local_recorder(&recorder, || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime must build"); + runtime.block_on(test()) + }); + (snapshotter.snapshot().into_vec(), out) + } + + fn gauge_value(snapshot: &[MetricEntry], name: &str) -> Option { + snapshot.iter().find_map(|(composite, _unit, _description, value)| { + let matches = composite.kind() == MetricKind::Gauge && composite.key().name() == name; + match (matches, value) { + (true, DebugValue::Gauge(value)) => Some(value.into_inner()), + _ => None, + } + }) + } + + fn counter_value(snapshot: &[MetricEntry], name: &str, outcome: &str) -> u64 { + snapshot + .iter() + .filter_map(|(composite, _unit, _description, value)| { + let matches = composite.kind() == MetricKind::Counter + && composite.key().name() == name + && composite + .key() + .labels() + .any(|label| label.key() == "outcome" && label.value() == outcome); + match (matches, value) { + (true, DebugValue::Counter(count)) => Some(*count), + _ => None, + } + }) + .sum() + } + + fn key_info(key_id: &str, status: KeyStatus, created_at: Zoned, rotated_at: Option) -> KeyInfo { + KeyInfo { + key_id: key_id.to_string(), + description: None, + algorithm: "AES-256".to_string(), + usage: KeyUsage::EncryptDecrypt, + status, + version: 1, + metadata: std::collections::HashMap::new(), + tags: std::collections::HashMap::new(), + created_at, + rotated_at, + created_by: None, + } + } + + #[test] + fn census_ages_from_rotation_and_ignores_departing_keys() { + let now = Zoned::now(); + let day = Duration::from_secs(86400); + let mut census = KeyCensus::default(); + + // Rotation beats creation as the age baseline... + census.observe( + &key_info("rotated", KeyStatus::Active, now.clone() - 30 * day, Some(now.clone() - 2 * day)), + &now, + ); + // ...and a key that was never rotated ages from its creation. + census.observe(&key_info("never-rotated", KeyStatus::Disabled, now.clone() - 5 * day, None), &now); + // Keys on their way out only ever move the counts: they will not be + // rotated again, so their age must not drive the rotation gauge. + census.observe(&key_info("pending", KeyStatus::PendingDeletion, now.clone() - 400 * day, None), &now); + census.observe(&key_info("tombstone", KeyStatus::Deleted, now.clone() - 400 * day, None), &now); + + assert_eq!(census.pending_deletion, 1); + assert_eq!(census.tombstones, 1); + let expected = 5.0 * 86400.0; + assert!( + (census.oldest_rotation_age_seconds - expected).abs() < 1.0, + "expected the never-rotated key to set the age, got {}", + census.oldest_rotation_age_seconds + ); + } + + #[test] + fn sweep_publishes_lifecycle_gauges_without_key_labels() { + let (snapshot, key_ids) = record_metrics(|| { + Box::pin(async { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let live = create_key(&backend, "live-key").await; + let doomed = create_key(&backend, "doomed-key").await; + schedule(&backend, &doomed).await; + + // Three days in, still inside the seven-day window: the sweep + // observes the scheduled key instead of removing it. + let report = worker(backend.clone()) + .sweep(&(Zoned::now() + Duration::from_secs(3 * 86400))) + .await; + assert_eq!(report.skipped, 1); + assert!(report.removed.is_empty()); + vec![live, doomed] + }) + }); + + assert_eq!(gauge_value(&snapshot, METRIC_PENDING_DELETION_KEYS), Some(1.0)); + assert_eq!(gauge_value(&snapshot, METRIC_TOMBSTONE_KEYS), Some(0.0)); + let age = gauge_value(&snapshot, METRIC_OLDEST_ROTATION_AGE_SECONDS).expect("rotation age gauge must be published"); + // Both keys were just created, and only the usable one counts, so the + // reported age is the sweep's own offset into the future. + assert!( + (age - 3.0 * 86400.0).abs() < 60.0, + "expected the sweep offset as the rotation age, got {age}" + ); + assert_eq!(counter_value(&snapshot, METRIC_SWEEP_KEYS_TOTAL, "skipped"), 1); + assert_eq!(counter_value(&snapshot, METRIC_SWEEP_KEYS_TOTAL, "removed"), 0); + + for (composite, ..) in &snapshot { + for label in composite.key().labels() { + for key_id in &key_ids { + assert!( + !label.value().contains(key_id.as_str()), + "metric {} leaked a key identifier through label {}", + composite.key().name(), + label.key() + ); + } + } + } + } }