From 83cdea1f186ce6d38fce2b7270620ecd368bde81 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 7 Aug 2026 08:52:02 +0800 Subject: [PATCH] feat(rpc): expose and auto-size replay cache capacity (#5781) * feat(metrics): expose replay cache pressure Co-Authored-By: heihutu * feat(rpc): auto-size replay cache capacity Co-Authored-By: heihutu * feat(cache): split runtime memory feature Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 1 + Cargo.toml | 2 +- crates/config/src/constants/internode.rs | 7 +- crates/ecstore/Cargo.toml | 1 + crates/ecstore/src/cluster/rpc/http_auth.rs | 502 ++++++++++++++++-- crates/io-metrics/src/internode_metrics.rs | 176 +++++- crates/object-data-cache/Cargo.toml | 32 +- crates/object-data-cache/src/config.rs | 4 +- crates/object-data-cache/src/lib.rs | 25 + crates/object-data-cache/src/memory.rs | 110 +--- .../object-data-cache/src/runtime_memory.rs | 124 +++++ rustfs/Cargo.toml | 2 +- 12 files changed, 795 insertions(+), 191 deletions(-) create mode 100644 crates/object-data-cache/src/runtime_memory.rs diff --git a/Cargo.lock b/Cargo.lock index cedb9fa3e..92a3ec9a9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9334,6 +9334,7 @@ dependencies = [ "rustfs-lock", "rustfs-madmin", "rustfs-object-capacity", + "rustfs-object-data-cache", "rustfs-policy", "rustfs-protos", "rustfs-replication", diff --git a/Cargo.toml b/Cargo.toml index 1170cd85d..07ed29a39 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,7 +108,7 @@ rustfs-notify = { path = "crates/notify", version = "1.0.0-beta.12" } rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-beta.12" } rustfs-io-core = { path = "crates/io-core", version = "1.0.0-beta.12" } rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-beta.12" } -rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.12" } +rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.12", default-features = false } rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-beta.12" } rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.12" } rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.12" } diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index 45f979ddc..4a7910632 100644 --- a/crates/config/src/constants/internode.rs +++ b/crates/config/src/constants/internode.rs @@ -177,10 +177,9 @@ const _: () = assert!(!DEFAULT_INTERNODE_RPC_REPLAY_SCOPE_STRICT); /// /// The cache retains each nonce for the ~10-minute signature freshness envelope. Once peers use /// replay-scoped v3 authentication, every authenticated RPC consumes one entry, so the steady -/// state holds roughly `authenticated RPC RPS x 601s` entries. The default sustains about 1,700 -/// authenticated RPCs per second (about 120 MiB worst case, allocated only under sustained load); -/// operators must size it for the node's aggregate peak RPC rate before enabling strict replay -/// scope. Overflow fails closed — legitimate signed traffic is the only thing that can fill the +/// state holds roughly `authenticated RPC RPS x 601s` entries. This default is the minimum floor: +/// explicit operator values and resource-aware auto sizing both clamp upward to at least this +/// value. Overflow fails closed — legitimate signed traffic is the only thing that can fill the /// cache (replays are rejected before insertion, and an attacker cannot mint valid nonces without /// the shared secret) — and increments /// `rustfs_system_network_internode_replay_cache_overflow_total`, so a sustained non-zero overflow diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 83499c027..35470aa1f 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -144,6 +144,7 @@ rustfs-lifecycle.workspace = true rustfs-s3-types = { workspace = true } rustfs-data-usage.workspace = true rustfs-object-capacity.workspace = true +rustfs-object-data-cache = { workspace = true, features = ["runtime-memory"] } arc-swap.workspace = true async-trait.workspace = true bytes = { workspace = true, features = ["serde"] } diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index c06033fd7..797eed459 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -36,15 +36,20 @@ use http::{HeaderMap, HeaderValue, Method, Uri}; #[cfg(test)] use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE}; use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token}; -use rustfs_io_metrics::internode_metrics::global_internode_metrics; +use rustfs_io_metrics::internode_metrics::{ + INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, +}; +use rustfs_object_data_cache::{MemoryBasis, resolve_effective_memory}; use rustfs_utils::get_env_bool; use sha2::Digest as _; use sha2::Sha256; use std::collections::{HashSet, VecDeque}; use std::sync::{LazyLock, Mutex, Once}; +use std::thread; use std::time::{Duration, Instant}; use time::OffsetDateTime; -use tracing::error; +use tracing::{error, info, warn}; use uuid::Uuid; type HmacSha256 = Hmac; @@ -70,6 +75,11 @@ const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601); +const REPLAY_CACHE_RETENTION_SECS: usize = 601; +const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128; +const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 4; +const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 1024; +const REPLAY_CACHE_AUTO_MAX_CAPACITY: usize = 8_388_608; const NS_SCANNER_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-capability-v3"; pub const TONIC_RPC_PREFIX: &str = "/node_service.NodeService"; static INTERNODE_RPC_SIGNATURE_STRICT: LazyLock = LazyLock::new(|| { @@ -91,18 +101,211 @@ static INTERNODE_RPC_REPLAY_SCOPE_STRICT: LazyLock = LazyLock::new(|| { ) }); // Sized for peak legitimate authenticated RPC RPS x the retention window once replay scope is -// active; overflow fails closed and increments the replay-cache overflow counter. Clamped to at -// least 1 so a misconfigured zero cannot disable replay protection by rejecting every request. -static REPLAY_CACHE_CAPACITY: LazyLock = LazyLock::new(|| { - rustfs_utils::get_env_usize( - rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, - rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, - ) - .max(1) -}); +// active; overflow fails closed and increments the replay-cache overflow counter. Explicit operator +// values and auto-sizing are both floored at the historical default so under-sizing cannot turn +// legitimate high-throughput traffic into `No valid auth token` failures. +static REPLAY_CACHE_CAPACITY: LazyLock = LazyLock::new(resolve_replay_cache_capacity); static RPC_SECRET_RESOLUTION_LOG_ONCE: Once = Once::new(); static RPC_BOOT_EPOCH: LazyLock = LazyLock::new(Uuid::new_v4); +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReplayCacheCapacitySource { + Env, + EnvClampedToDefault, + Auto, + AutoClampedToDefault, + AutoInvalidEnv, + AutoInvalidEnvClampedToDefault, +} + +impl ReplayCacheCapacitySource { + fn as_str(self) -> &'static str { + match self { + Self::Env => "env", + Self::EnvClampedToDefault => "env_clamped_to_default", + Self::Auto => "auto", + Self::AutoClampedToDefault => "auto_clamped_to_default", + Self::AutoInvalidEnv => "auto_invalid_env", + Self::AutoInvalidEnvClampedToDefault => "auto_invalid_env_clamped_to_default", + } + } + + fn is_env_clamped(self) -> bool { + matches!(self, Self::EnvClampedToDefault) + } + + fn is_env(self) -> bool { + matches!(self, Self::Env) + } + + fn is_invalid_env(self) -> bool { + matches!(self, Self::AutoInvalidEnv | Self::AutoInvalidEnvClampedToDefault) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ReplayCacheCapacityDecision { + capacity: usize, + source: ReplayCacheCapacitySource, + cpu_count: usize, + memory_limit_bytes: Option, + memory_basis: Option, + memory_based_capacity: usize, + cpu_based_capacity: usize, +} + +fn saturating_usize_from_u64(value: u64) -> usize { + usize::try_from(value).unwrap_or(usize::MAX) +} + +fn replay_cache_capacity_from_resources(cpu_count: usize, memory_limit_bytes: Option) -> (usize, usize, usize) { + let cpu_count = cpu_count.max(1); + let cpu_based_capacity = cpu_count + .saturating_mul(REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU) + .saturating_mul(REPLAY_CACHE_RETENTION_SECS); + let memory_based_capacity = memory_limit_bytes + .map(|bytes| { + let budget = bytes.saturating_mul(REPLAY_CACHE_AUTO_MEMORY_PERCENT) / 100; + saturating_usize_from_u64(budget / REPLAY_CACHE_ENTRY_BYTES_ESTIMATE) + }) + .unwrap_or(REPLAY_CACHE_AUTO_MAX_CAPACITY); + let capacity = memory_based_capacity + .min(cpu_based_capacity) + .clamp(rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, REPLAY_CACHE_AUTO_MAX_CAPACITY); + (capacity, memory_based_capacity, cpu_based_capacity) +} + +fn replay_cache_capacity_decision( + env: rustfs_utils::EnvParseOutcome, + cpu_count: usize, + memory_limit_bytes: Option, + memory_basis: Option, +) -> ReplayCacheCapacityDecision { + let default = rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY; + match env { + rustfs_utils::EnvParseOutcome::Parsed(configured) => { + let capacity = configured.max(default); + let source = if configured < default { + ReplayCacheCapacitySource::EnvClampedToDefault + } else { + ReplayCacheCapacitySource::Env + }; + ReplayCacheCapacityDecision { + capacity, + source, + cpu_count: cpu_count.max(1), + memory_limit_bytes, + memory_basis, + memory_based_capacity: 0, + cpu_based_capacity: 0, + } + } + rustfs_utils::EnvParseOutcome::Absent | rustfs_utils::EnvParseOutcome::Invalid => { + let (capacity, memory_based_capacity, cpu_based_capacity) = + replay_cache_capacity_from_resources(cpu_count, memory_limit_bytes); + let clamped_to_default = capacity == default && memory_based_capacity.min(cpu_based_capacity) < default; + let invalid_env = matches!(env, rustfs_utils::EnvParseOutcome::Invalid); + let source = match (invalid_env, clamped_to_default) { + (true, true) => ReplayCacheCapacitySource::AutoInvalidEnvClampedToDefault, + (true, false) => ReplayCacheCapacitySource::AutoInvalidEnv, + (false, true) => ReplayCacheCapacitySource::AutoClampedToDefault, + (false, false) => ReplayCacheCapacitySource::Auto, + }; + ReplayCacheCapacityDecision { + capacity, + source, + cpu_count: cpu_count.max(1), + memory_limit_bytes, + memory_basis, + memory_based_capacity, + cpu_based_capacity, + } + } + } +} + +fn detected_replay_cache_resources() -> (usize, Option, Option) { + let cpu_count = thread::available_parallelism().map(usize::from).unwrap_or(1).max(1); + let memory = resolve_effective_memory(); + let memory_limit_bytes = (memory.total_bytes > 0).then_some(memory.total_bytes); + (cpu_count, memory_limit_bytes, Some(memory.basis)) +} + +fn log_replay_cache_capacity_decision(decision: ReplayCacheCapacityDecision) { + let source = decision.source.as_str(); + if decision.source.is_env_clamped() { + warn!( + event = "internode_rpc_replay_cache_capacity_resolved", + component = "ecstore", + subsystem = "rpc_auth", + capacity = decision.capacity, + source, + default_capacity = rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, + env = rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, + "internode rpc replay cache capacity clamped to default" + ); + return; + } + if decision.source.is_env() { + info!( + event = "internode_rpc_replay_cache_capacity_resolved", + component = "ecstore", + subsystem = "rpc_auth", + capacity = decision.capacity, + source, + default_capacity = rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, + env = rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, + "internode rpc replay cache capacity resolved from env" + ); + return; + } + if decision.source.is_invalid_env() { + warn!( + event = "internode_rpc_replay_cache_capacity_resolved", + component = "ecstore", + subsystem = "rpc_auth", + capacity = decision.capacity, + source, + cpu_count = decision.cpu_count, + memory_limit_bytes = decision.memory_limit_bytes, + memory_basis = decision.memory_basis.map(MemoryBasis::as_str), + memory_based_capacity = decision.memory_based_capacity, + cpu_based_capacity = decision.cpu_based_capacity, + auto_max_capacity = REPLAY_CACHE_AUTO_MAX_CAPACITY, + env = rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY, + "internode rpc replay cache capacity auto-sized after invalid env" + ); + return; + } + info!( + event = "internode_rpc_replay_cache_capacity_resolved", + component = "ecstore", + subsystem = "rpc_auth", + capacity = decision.capacity, + source, + cpu_count = decision.cpu_count, + memory_limit_bytes = decision.memory_limit_bytes, + memory_basis = decision.memory_basis.map(MemoryBasis::as_str), + memory_based_capacity = decision.memory_based_capacity, + cpu_based_capacity = decision.cpu_based_capacity, + auto_max_capacity = REPLAY_CACHE_AUTO_MAX_CAPACITY, + "internode rpc replay cache capacity resolved" + ); +} + +fn resolve_replay_cache_capacity() -> usize { + let (cpu_count, memory_limit_bytes, memory_basis) = detected_replay_cache_resources(); + let decision = replay_cache_capacity_decision( + rustfs_utils::get_env_parse_outcome(rustfs_config::ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY), + cpu_count, + memory_limit_bytes, + memory_basis, + ); + global_internode_metrics().record_replay_cache_state(0, decision.capacity); + log_replay_cache_capacity_decision(decision); + decision.capacity +} + #[derive(Default)] struct RpcNonceCache { nonces: HashSet, @@ -110,8 +313,50 @@ struct RpcNonceCache { max_wall_time: i64, } +#[derive(Clone, Copy)] +struct RpcReplayCacheMetricScope<'a> { + operation: &'static str, + backend: &'static str, + rpc_path: &'a str, +} + +#[derive(Clone, Copy)] +struct RpcNonceRecord<'a> { + nonce: Uuid, + signed_at: i64, + now: Instant, + wall_time: i64, + expires_at: Instant, + capacity: usize, + metric_scope: RpcReplayCacheMetricScope<'a>, +} + +struct RpcNonceCacheMetrics<'a> { + expired: usize, + entries: usize, + capacity: usize, + overflow_scope: Option>, +} + +fn publish_nonce_cache_metrics(metrics: Option>) { + let Some(metrics) = metrics else { + return; + }; + let internode_metrics = global_internode_metrics(); + internode_metrics.record_replay_cache_evictions("expired", metrics.expired); + internode_metrics.record_replay_cache_state(metrics.entries, metrics.capacity); + if let Some(scope) = metrics.overflow_scope { + internode_metrics.record_replay_cache_overflow_for_operation_and_backend_path( + scope.operation, + scope.backend, + scope.rpc_path, + ); + } +} + impl RpcNonceCache { - fn remove_expired(&mut self, now: Instant, wall_time: i64) { + fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize { + let mut removed = 0; while matches!( self.expirations.front(), Some((expires_at, valid_until, _)) if *expires_at < now && *valid_until < wall_time @@ -120,37 +365,48 @@ impl RpcNonceCache { break; }; self.nonces.remove(&nonce); + removed += 1; } + removed } - fn check_and_record( - &mut self, - nonce: Uuid, - signed_at: i64, - now: Instant, - wall_time: i64, - expires_at: Instant, - capacity: usize, - ) -> std::io::Result<()> { - self.max_wall_time = self.max_wall_time.max(wall_time); - if self.max_wall_time.saturating_sub(signed_at) > SIGNATURE_VALID_DURATION { - return Err(std::io::Error::other("RPC request timestamp expired after clock regression")); + fn check_and_record<'a>(&mut self, record: RpcNonceRecord<'a>) -> (std::io::Result<()>, Option>) { + self.max_wall_time = self.max_wall_time.max(record.wall_time); + if self.max_wall_time.saturating_sub(record.signed_at) > SIGNATURE_VALID_DURATION { + return (Err(std::io::Error::other("RPC request timestamp expired after clock regression")), None); } - self.remove_expired(now, self.max_wall_time); - if self.nonces.contains(&nonce) { - return Err(std::io::Error::other("RPC request replay detected")); + let expired = self.remove_expired(record.now, self.max_wall_time); + let metrics = RpcNonceCacheMetrics { + expired, + entries: self.nonces.len(), + capacity: record.capacity, + overflow_scope: None, + }; + if self.nonces.contains(&record.nonce) { + return (Err(std::io::Error::other("RPC request replay detected")), Some(metrics)); } - if self.nonces.len() >= capacity { + if self.nonces.len() >= record.capacity { // Fail closed and alert: only legitimately signed traffic can fill the cache, so a // sustained overflow means RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY is undersized // for this node's peak mutation rate and writes are being refused. - global_internode_metrics().record_replay_cache_overflow(); - return Err(std::io::Error::other("RPC replay cache capacity exceeded")); + return ( + Err(std::io::Error::other("RPC replay cache capacity exceeded")), + Some(RpcNonceCacheMetrics { + overflow_scope: Some(record.metric_scope), + ..metrics + }), + ); } - self.nonces.insert(nonce); + self.nonces.insert(record.nonce); self.expirations - .push_back((expires_at, signed_at.saturating_add(SIGNATURE_VALID_DURATION), nonce)); - Ok(()) + .push_back((record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce)); + ( + Ok(()), + Some(RpcNonceCacheMetrics { + entries: self.nonces.len(), + ..metrics + }), + ) } } @@ -541,18 +797,43 @@ fn check_timestamp(timestamp: i64) -> std::io::Result<()> { Ok(()) } -fn check_and_record_nonce(nonce: Uuid, signed_at: i64) -> std::io::Result<()> { +fn tonic_rpc_metric_operation(path: &str) -> &'static str { + match parse_tonic_rpc_path(path).ok().map(|(_, rpc_method)| rpc_method) { + Some("ReadAll") => INTERNODE_OPERATION_GRPC_READ_ALL, + Some("ReadMultiple") => INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + Some("WriteAll") => INTERNODE_OPERATION_GRPC_WRITE_ALL, + _ => INTERNODE_OPERATION_GRPC_OTHER, + } +} + +fn check_and_record_nonce(nonce: Uuid, signed_at: i64, rpc_path: &str) -> std::io::Result<()> { let wall_time = OffsetDateTime::now_utc().unix_timestamp(); - let mut cache = LOCAL_RPC_NONCE_CACHE - .lock() - .map_err(|_| std::io::Error::other("RPC replay cache unavailable"))?; - // Take the monotonic timestamp after acquiring the lock so expiration - // entries remain ordered by the same serialization point as insertion. - let now = Instant::now(); - let expires_at = now - .checked_add(REPLAY_CACHE_RETENTION) - .ok_or_else(|| std::io::Error::other("RPC replay expiry overflow"))?; - cache.check_and_record(nonce, signed_at, now, wall_time, expires_at, *REPLAY_CACHE_CAPACITY) + let (result, metrics) = { + let mut cache = LOCAL_RPC_NONCE_CACHE + .lock() + .map_err(|_| std::io::Error::other("RPC replay cache unavailable"))?; + // Take the monotonic timestamp after acquiring the lock so expiration + // entries remain ordered by the same serialization point as insertion. + let now = Instant::now(); + let expires_at = now + .checked_add(REPLAY_CACHE_RETENTION) + .ok_or_else(|| std::io::Error::other("RPC replay expiry overflow"))?; + cache.check_and_record(RpcNonceRecord { + nonce, + signed_at, + now, + wall_time, + expires_at, + capacity: *REPLAY_CACHE_CAPACITY, + metric_scope: RpcReplayCacheMetricScope { + operation: tonic_rpc_metric_operation(rpc_path), + backend: INTERNODE_TRANSPORT_BACKEND_GRPC, + rpc_path, + }, + }) + }; + publish_nonce_cache_metrics(metrics); + result } /// Build headers with authentication signature @@ -814,7 +1095,7 @@ fn verify_tonic_replay_scope_signature(audience: &str, path: &str, headers: &Hea if boot_epoch != tonic_rpc_boot_epoch() { return Err(std::io::Error::other("RPC boot epoch is stale")); } - check_and_record_nonce(nonce, signed_at) + check_and_record_nonce(nonce, signed_at, path) } /// Verify gRPC authentication, preferring v2 without downgrade on malformed v2 metadata. @@ -1005,7 +1286,7 @@ fn verify_tonic_rpc_signature_with_strictness( return Err(std::io::Error::other("Invalid RPC v2 signature")); } if let Some(nonce) = parsed_nonce { - check_and_record_nonce(nonce, timestamp)?; + check_and_record_nonce(nonce, timestamp, path)?; } Ok(()) } @@ -1968,6 +2249,114 @@ mod tests { assert_eq!(error.to_string(), "RPC mutation requires v2 authentication"); } + #[test] + fn tonic_rpc_metric_operation_classifies_get_hot_path_methods() { + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/ReadAll"), + INTERNODE_OPERATION_GRPC_READ_ALL + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/ReadMultiple"), + INTERNODE_OPERATION_GRPC_READ_MULTIPLE + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/WriteAll"), + INTERNODE_OPERATION_GRPC_WRITE_ALL + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/SignalService"), + INTERNODE_OPERATION_GRPC_OTHER + ); + assert_eq!(tonic_rpc_metric_operation("not-a-grpc-path"), INTERNODE_OPERATION_GRPC_OTHER); + } + + #[test] + fn replay_cache_capacity_uses_env_with_default_floor() { + let default = rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY; + + let high = replay_cache_capacity_decision( + rustfs_utils::EnvParseOutcome::Parsed(default * 16), + 2, + Some(512 * 1024 * 1024), + Some(MemoryBasis::Host), + ); + assert_eq!(high.capacity, default * 16); + assert_eq!(high.source, ReplayCacheCapacitySource::Env); + + let low = replay_cache_capacity_decision( + rustfs_utils::EnvParseOutcome::Parsed(1), + 64, + Some(128 * 1024 * 1024 * 1024), + Some(MemoryBasis::Host), + ); + assert_eq!(low.capacity, default); + assert_eq!(low.source, ReplayCacheCapacitySource::EnvClampedToDefault); + } + + #[test] + fn replay_cache_capacity_auto_sizes_from_cpu_and_memory() { + let gib = 1024_u64 * 1024 * 1024; + let decision = + replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 8, Some(16 * gib), Some(MemoryBasis::Host)); + + assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); + assert_eq!(decision.memory_basis, Some(MemoryBasis::Host)); + assert_eq!(decision.memory_based_capacity, 5_368_709); + assert_eq!(decision.cpu_based_capacity, 4_923_392); + assert_eq!(decision.capacity, 4_923_392); + } + + #[test] + fn replay_cache_capacity_auto_keeps_default_floor_for_small_nodes() { + let decision = replay_cache_capacity_decision( + rustfs_utils::EnvParseOutcome::Absent, + 1, + Some(512 * 1024 * 1024), + Some(MemoryBasis::Host), + ); + + assert_eq!(decision.capacity, rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY); + assert_eq!(decision.source, ReplayCacheCapacitySource::AutoClampedToDefault); + assert!(decision.memory_based_capacity < rustfs_config::DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY); + } + + #[test] + fn replay_cache_capacity_invalid_env_uses_auto_sizing() { + let decision = replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Invalid, 8, None, None); + + assert_eq!(decision.source, ReplayCacheCapacitySource::AutoInvalidEnv); + assert_eq!(decision.capacity, 4_923_392); + } + + fn check_test_nonce_record(cache: &mut RpcNonceCache, record: RpcNonceRecord<'_>) -> std::io::Result<()> { + let (result, metrics) = cache.check_and_record(record); + publish_nonce_cache_metrics(metrics); + result + } + + fn test_nonce_record( + nonce: Uuid, + signed_at: i64, + now: Instant, + wall_time: i64, + expires_at: Instant, + capacity: usize, + ) -> RpcNonceRecord<'static> { + RpcNonceRecord { + nonce, + signed_at, + now, + wall_time, + expires_at, + capacity, + metric_scope: RpcReplayCacheMetricScope { + operation: INTERNODE_OPERATION_GRPC_READ_ALL, + backend: INTERNODE_TRANSPORT_BACKEND_GRPC, + rpc_path: "/node_service.NodeService/ReadAll", + }, + } + } + #[test] fn nonce_cache_expires_by_monotonic_deadline_and_fails_closed_at_capacity() { let now = Instant::now(); @@ -1977,15 +2366,12 @@ mod tests { let nonce_b = Uuid::new_v4(); let mut cache = RpcNonceCache::default(); - cache - .check_and_record(nonce_a, 100, now, 100, expiry, 1) + check_test_nonce_record(&mut cache, test_nonce_record(nonce_a, 100, now, 100, expiry, 1)) .expect("first nonce should be recorded"); - let capacity = cache - .check_and_record(nonce_b, 100, now, 100, expiry, 1) + let capacity = check_test_nonce_record(&mut cache, test_nonce_record(nonce_b, 100, now, 100, expiry, 1)) .expect_err("a full replay cache must fail closed"); assert_eq!(capacity.to_string(), "RPC replay cache capacity exceeded"); - cache - .check_and_record(nonce_b, 702, after_expiry, 702, after_expiry, 1) + check_test_nonce_record(&mut cache, test_nonce_record(nonce_b, 702, after_expiry, 702, after_expiry, 1)) .expect("expired nonce should release capacity"); assert!(!cache.nonces.contains(&nonce_a)); assert!(cache.nonces.contains(&nonce_b)); @@ -2188,17 +2574,15 @@ mod tests { let nonce = Uuid::new_v4(); let mut cache = RpcNonceCache::default(); - cache - .check_and_record(nonce, 1_000, now, 1_000, expiry, 2) + check_test_nonce_record(&mut cache, test_nonce_record(nonce, 1_000, now, 1_000, expiry, 2)) .expect("first nonce should be recorded"); - let replay = cache - .check_and_record(nonce, 1_000, after_expiry, 900, after_expiry, 2) + let replay = check_test_nonce_record(&mut cache, test_nonce_record(nonce, 1_000, after_expiry, 900, after_expiry, 2)) .expect_err("wall clock regression must not make an old signature reusable"); assert_eq!(replay.to_string(), "RPC request replay detected"); - let stale = cache - .check_and_record(Uuid::new_v4(), 600, after_expiry, 900, after_expiry, 2) - .expect_err("the monotonic wall-clock high-water mark must fail closed"); + let stale = + check_test_nonce_record(&mut cache, test_nonce_record(Uuid::new_v4(), 600, after_expiry, 900, after_expiry, 2)) + .expect_err("the monotonic wall-clock high-water mark must fail closed"); assert_eq!(stale.to_string(), "RPC request timestamp expired after clock regression"); } } diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 4cebf3202..020201779 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -47,6 +47,8 @@ const STAGE_LABEL: &str = "stage"; const DOMINANT_ERROR_LABEL: &str = "dominant_error"; const HTTP_VERSION_LABEL: &str = "http_version"; const FAILURE_REASON_LABEL: &str = "failure_reason"; +const RPC_PATH_LABEL: &str = "rpc_path"; +const REASON_LABEL: &str = "reason"; const DIRECTION_LABEL: &str = "direction"; const MESSAGE_LABEL: &str = "message"; const CODEC_LABEL: &str = "codec"; @@ -73,6 +75,11 @@ const INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL: &str = "rustfs_system_network_inter const INTERNODE_BODY_DIGEST_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_body_digest_fallback_total"; const INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_replay_scope_fallback_total"; const INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL: &str = "rustfs_system_network_internode_replay_cache_overflow_total"; +const INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL: &str = + "rustfs_system_network_internode_replay_cache_overflow_by_operation_total"; +const INTERNODE_REPLAY_CACHE_ENTRIES: &str = "rustfs_system_network_internode_replay_cache_entries"; +const INTERNODE_REPLAY_CACHE_CAPACITY: &str = "rustfs_system_network_internode_replay_cache_capacity"; +const INTERNODE_REPLAY_CACHE_EVICTIONS_TOTAL: &str = "rustfs_system_network_internode_replay_cache_evictions_total"; const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -87,6 +94,9 @@ const SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS: &[&str] = const SERVER_OPERATION_BACKEND_HTTP_VERSION_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL]; const SERVER_OPERATION_BACKEND_FAILURE_REASON_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL]; +const SERVER_OPERATION_BACKEND_RPC_PATH_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL]; +const SERVER_LABELS: &[&str] = &[SERVER_LABEL]; +const SERVER_REASON_LABELS: &[&str] = &[SERVER_LABEL, REASON_LABEL]; const SERVER_QUORUM_FAILURE_LABELS: &[&str] = &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]; pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &[ @@ -142,6 +152,22 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: INTERNODE_RPC_AUTH_FAILURES_TOTAL, labels: SERVER_OPERATION_BACKEND_FAILURE_REASON_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL, + labels: SERVER_OPERATION_BACKEND_RPC_PATH_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_REPLAY_CACHE_ENTRIES, + labels: SERVER_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_REPLAY_CACHE_CAPACITY, + labels: SERVER_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_REPLAY_CACHE_EVICTIONS_TOTAL, + labels: SERVER_REASON_LABELS, + }, InternodeOperationMetricDescriptor { name: ERASURE_WRITE_QUORUM_FAILURES_TOTAL, labels: SERVER_QUORUM_FAILURE_LABELS, @@ -192,6 +218,9 @@ pub struct InternodeMetricsSnapshot { pub body_digest_fallback_total: u64, pub replay_scope_fallback_total: u64, pub replay_cache_overflow_total: u64, + pub replay_cache_entries: u64, + pub replay_cache_capacity: u64, + pub replay_cache_evictions_total: u64, } #[derive(Debug, Default)] @@ -215,6 +244,13 @@ pub struct InternodeMetrics { body_digest_fallback_total: AtomicU64, replay_scope_fallback_total: AtomicU64, replay_cache_overflow_total: AtomicU64, + replay_cache_entries: AtomicU64, + replay_cache_capacity: AtomicU64, + replay_cache_evictions_total: AtomicU64, +} + +fn usize_to_u64_saturating(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) } impl InternodeMetrics { @@ -565,6 +601,46 @@ impl InternodeMetrics { counter!(INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL, SERVER_LABEL => current_server_label()).increment(1); } + pub fn record_replay_cache_overflow_for_operation_and_backend_path( + &self, + operation: &'static str, + backend: &'static str, + rpc_path: &str, + ) { + self.record_replay_cache_overflow(); + counter!( + INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL, + SERVER_LABEL => current_server_label(), + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + RPC_PATH_LABEL => rpc_path.to_owned() + ) + .increment(1); + } + + pub fn record_replay_cache_state(&self, entries: usize, capacity: usize) { + let entries = usize_to_u64_saturating(entries); + let capacity = usize_to_u64_saturating(capacity); + self.replay_cache_entries.store(entries, Ordering::Relaxed); + self.replay_cache_capacity.store(capacity, Ordering::Relaxed); + gauge!(INTERNODE_REPLAY_CACHE_ENTRIES, SERVER_LABEL => current_server_label()).set(entries as f64); + gauge!(INTERNODE_REPLAY_CACHE_CAPACITY, SERVER_LABEL => current_server_label()).set(capacity as f64); + } + + pub fn record_replay_cache_evictions(&self, reason: &'static str, count: usize) { + if count == 0 { + return; + } + let count = usize_to_u64_saturating(count); + self.replay_cache_evictions_total.fetch_add(count, Ordering::Relaxed); + counter!( + INTERNODE_REPLAY_CACHE_EVICTIONS_TOTAL, + SERVER_LABEL => current_server_label(), + REASON_LABEL => reason + ) + .increment(count); + } + pub fn record_erasure_write_quorum_failure(&self, stage: &'static str, dominant_error: &'static str) { counter!( ERASURE_WRITE_QUORUM_FAILURES_TOTAL, @@ -618,6 +694,9 @@ impl InternodeMetrics { body_digest_fallback_total: self.body_digest_fallback_total.load(Ordering::Relaxed), replay_scope_fallback_total: self.replay_scope_fallback_total.load(Ordering::Relaxed), replay_cache_overflow_total: self.replay_cache_overflow_total.load(Ordering::Relaxed), + replay_cache_entries: self.replay_cache_entries.load(Ordering::Relaxed), + replay_cache_capacity: self.replay_cache_capacity.load(Ordering::Relaxed), + replay_cache_evictions_total: self.replay_cache_evictions_total.load(Ordering::Relaxed), } } @@ -642,6 +721,9 @@ impl InternodeMetrics { self.body_digest_fallback_total.store(0, Ordering::Relaxed); self.replay_scope_fallback_total.store(0, Ordering::Relaxed); self.replay_cache_overflow_total.store(0, Ordering::Relaxed); + self.replay_cache_entries.store(0, Ordering::Relaxed); + self.replay_cache_capacity.store(0, Ordering::Relaxed); + self.replay_cache_evictions_total.store(0, Ordering::Relaxed); } } @@ -864,6 +946,8 @@ mod tests { INTERNODE_TRANSPORT_BACKEND_GRPC, "missing_v2_signature", ); + metrics.record_replay_cache_state(64, 1024); + metrics.record_replay_cache_evictions("expired", 3); let snapshot = metrics.snapshot(); assert_eq!(snapshot.sent_bytes_total, 128); @@ -872,11 +956,14 @@ mod tests { assert_eq!(snapshot.incoming_requests_total, 1); assert_eq!(snapshot.errors_total, 1); assert_eq!(snapshot.rpc_auth_failures_total, 1); + assert_eq!(snapshot.replay_cache_entries, 64); + assert_eq!(snapshot.replay_cache_capacity, 1024); + assert_eq!(snapshot.replay_cache_evictions_total, 3); } #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 16); + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 20); for metric in &INTERNODE_OPERATION_METRICS[..6] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } @@ -894,10 +981,18 @@ mod tests { INTERNODE_OPERATION_METRICS[12].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL] ); - assert_eq!(INTERNODE_OPERATION_METRICS[13].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]); + assert_eq!( + INTERNODE_OPERATION_METRICS[13].labels, + &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL] + ); + for metric in &INTERNODE_OPERATION_METRICS[14..16] { + assert_eq!(metric.labels, &[SERVER_LABEL]); + } + assert_eq!(INTERNODE_OPERATION_METRICS[16].labels, &[SERVER_LABEL, REASON_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[17].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]); // Payload histogram + large-payload counter carry operation+backend labels. - assert_eq!(INTERNODE_OPERATION_METRICS[14].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); - assert_eq!(INTERNODE_OPERATION_METRICS[15].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[19].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } #[test] @@ -947,14 +1042,30 @@ mod tests { ); assert_eq!( INTERNODE_OPERATION_METRICS[13].name, - "rustfs_system_storage_erasure_write_quorum_failures_total" + "rustfs_system_network_internode_replay_cache_overflow_by_operation_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[14].name, - "rustfs_system_network_internode_operation_payload_bytes" + "rustfs_system_network_internode_replay_cache_entries" ); assert_eq!( INTERNODE_OPERATION_METRICS[15].name, + "rustfs_system_network_internode_replay_cache_capacity" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[16].name, + "rustfs_system_network_internode_replay_cache_evictions_total" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[17].name, + "rustfs_system_storage_erasure_write_quorum_failures_total" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[18].name, + "rustfs_system_network_internode_operation_payload_bytes" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[19].name, "rustfs_system_network_internode_operation_large_payloads_total" ); assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); @@ -979,6 +1090,8 @@ mod tests { "rustfs_system_network_internode_signature_v1_fallback_total" ); assert_eq!(FAILURE_REASON_LABEL, "failure_reason"); + assert_eq!(RPC_PATH_LABEL, "rpc_path"); + assert_eq!(REASON_LABEL, "reason"); } #[test] @@ -1015,6 +1128,57 @@ mod tests { assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); } + #[test] + fn replay_cache_metrics_record_state_eviction_and_overflow_scope() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let metrics = InternodeMetrics::default(); + + with_local_recorder(&recorder, || { + metrics.record_replay_cache_state(7, 11); + metrics.record_replay_cache_evictions("expired", 5); + metrics.record_replay_cache_overflow_for_operation_and_backend_path( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + "/node_service.NodeService/ReadAll", + ); + }); + + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.replay_cache_entries, 7); + assert_eq!(snapshot.replay_cache_capacity, 11); + assert_eq!(snapshot.replay_cache_evictions_total, 5); + assert_eq!(snapshot.replay_cache_overflow_total, 1); + + let entries: Vec<_> = snapshotter.snapshot().into_vec(); + assert!( + entries + .iter() + .any(|(composite, _, _, _)| composite.key().name() == INTERNODE_REPLAY_CACHE_ENTRIES) + ); + assert!( + entries + .iter() + .any(|(composite, _, _, _)| composite.key().name() == INTERNODE_REPLAY_CACHE_CAPACITY) + ); + + let overflow: Vec<_> = entries + .iter() + .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL) + .collect(); + assert_eq!(overflow.len(), 1); + let labels: HashMap<_, _> = overflow[0] + .0 + .key() + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect(); + assert_eq!(labels.get(OPERATION_LABEL).map(String::as_str), Some(INTERNODE_OPERATION_GRPC_READ_ALL)); + assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC)); + assert_eq!(labels.get(RPC_PATH_LABEL).map(String::as_str), Some("/node_service.NodeService/ReadAll")); + assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); + } + #[test] fn direct_internode_metrics_emit_stable_server_label() { let recorder = DebuggingRecorder::new(); diff --git a/crates/object-data-cache/Cargo.toml b/crates/object-data-cache/Cargo.toml index 53a7ad5d3..704f3df6f 100644 --- a/crates/object-data-cache/Cargo.toml +++ b/crates/object-data-cache/Cargo.toml @@ -28,21 +28,33 @@ categories = ["web-programming", "development-tools"] doctest = false [features] -default = [] -hotpath = ["hotpath/hotpath", "hotpath/tokio"] +default = ["cache"] +runtime-memory = ["dep:sysinfo"] +cache = [ + "runtime-memory", + "dep:bytes", + "dep:metrics", + "dep:moka", + "dep:starshard", + "dep:thiserror", + "dep:tokio", + "dep:tracing", + "sysinfo/multithread", +] +hotpath = ["cache", "hotpath/hotpath", "hotpath/tokio"] hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"] hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"] [dependencies] hotpath.workspace = true -bytes = { workspace = true, features = ["serde"] } -metrics = { workspace = true } -moka = { workspace = true, features = ["future"] } -starshard = { workspace = true, features = ["rayon", "async", "serde"] } -sysinfo = { workspace = true, features = ["multithread"] } -thiserror.workspace = true -tokio = { workspace = true, features = ["sync", "time", "fs", "rt-multi-thread"] } -tracing.workspace = true +bytes = { workspace = true, optional = true, features = ["serde"] } +metrics = { workspace = true, optional = true } +moka = { workspace = true, optional = true, features = ["future"] } +starshard = { workspace = true, optional = true, features = ["rayon", "async", "serde"] } +sysinfo = { workspace = true, optional = true } +thiserror = { workspace = true, optional = true } +tokio = { workspace = true, optional = true, features = ["sync", "time", "fs", "rt-multi-thread"] } +tracing = { workspace = true, optional = true } [dev-dependencies] criterion = { workspace = true, features = ["html_reports"] } diff --git a/crates/object-data-cache/src/config.rs b/crates/object-data-cache/src/config.rs index 962450b2c..7c0e9cad4 100644 --- a/crates/object-data-cache/src/config.rs +++ b/crates/object-data-cache/src/config.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::error::ObjectDataCacheConfigError; -use crate::memory::{EffectiveMemory, MemoryBasis, resolve_effective_memory}; +use crate::{EffectiveMemory, MemoryBasis, resolve_effective_memory}; use std::sync::Once; use std::time::Duration; @@ -296,7 +296,7 @@ mod tests { clamp_derived_max_bytes, }; use crate::error::ObjectDataCacheConfigError; - use crate::memory::{EffectiveMemory, MemoryBasis}; + use crate::{EffectiveMemory, MemoryBasis}; use std::time::Duration; #[test] diff --git a/crates/object-data-cache/src/lib.rs b/crates/object-data-cache/src/lib.rs index e0ec6497b..c3860c766 100644 --- a/crates/object-data-cache/src/lib.rs +++ b/crates/object-data-cache/src/lib.rs @@ -41,27 +41,52 @@ //! the cache disabled for those buckets. Adding timing noise is not a viable //! mitigation: it would cost exactly the latency the cache exists to save. +#[cfg(feature = "cache")] pub mod backend; +#[cfg(feature = "cache")] pub mod cache; +#[cfg(feature = "cache")] pub mod config; +#[cfg(feature = "cache")] pub mod entry; +#[cfg(feature = "cache")] pub mod error; +#[cfg(feature = "cache")] pub mod index; +#[cfg(feature = "cache")] pub mod key; +#[cfg(feature = "cache")] pub mod memory; +#[cfg(feature = "cache")] pub mod metrics; +#[cfg(feature = "cache")] pub mod moka_backend; +#[cfg(feature = "cache")] pub mod noop; +#[cfg(feature = "runtime-memory")] +mod runtime_memory; +#[cfg(feature = "cache")] pub mod singleflight; +#[cfg(feature = "cache")] pub mod starshard_index; +#[cfg(feature = "cache")] pub mod stats; +#[cfg(feature = "cache")] pub use cache::{ ObjectDataCache, ObjectDataCacheBodyReservation, ObjectDataCacheFillResult, ObjectDataCacheGetPlan, ObjectDataCacheGetRequest, ObjectDataCacheInvalidationReason, ObjectDataCacheInvalidationResult, ObjectDataCacheLookup, ObjectDataCacheReservedBody, }; +#[cfg(feature = "cache")] pub use config::{ObjectDataCacheConfig, ObjectDataCacheMode}; +#[cfg(feature = "cache")] pub use error::ObjectDataCacheConfigError; +#[cfg(feature = "cache")] pub use key::{NULL_VERSION_ID, ObjectDataCacheBodyVariant, ObjectDataCacheIdentity, ObjectDataCacheKey}; +#[cfg(feature = "runtime-memory")] +pub use runtime_memory::{ + EffectiveMemory, MemoryBasis, effective_memory_from_system, resolve_effective_memory, select_effective_memory, +}; +#[cfg(feature = "cache")] pub use stats::{ObjectDataCacheStats, ObjectDataCacheStatsSnapshot}; diff --git a/crates/object-data-cache/src/memory.rs b/crates/object-data-cache/src/memory.rs index 354991c87..cacc3db40 100644 --- a/crates/object-data-cache/src/memory.rs +++ b/crates/object-data-cache/src/memory.rs @@ -14,6 +14,7 @@ use crate::config::ObjectDataCacheConfig; use crate::metrics::record_memory_pressure; +use crate::runtime_memory::resolve_effective_memory; use crate::stats::ObjectDataCacheStats; use bytes::Bytes; use std::hint::spin_loop; @@ -22,7 +23,6 @@ use std::sync::Arc; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::{Duration, Instant}; -use sysinfo::System; const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(5); const DEFAULT_TELEMETRY_STALENESS: Duration = Duration::from_secs(15); @@ -31,73 +31,6 @@ const DEFAULT_TELEMETRY_STALENESS: Duration = Duration::from_secs(15); // exhaustion safely skips only the cache fill. const MAX_STATE_RETRIES: usize = 8; -/// Source used to resolve the effective memory limits. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum MemoryBasis { - /// Limits come from host memory reported by sysinfo. - Host, - /// Limits come from a constraining cgroup (container) memory limit. - Cgroup, -} - -impl MemoryBasis { - pub(crate) const fn as_str(self) -> &'static str { - match self { - Self::Host => "host", - Self::Cgroup => "cgroup", - } - } -} - -/// Effective memory totals after reconciling host memory with cgroup limits. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) struct EffectiveMemory { - /// Effective total memory in bytes. - pub(crate) total_bytes: u64, - /// Effective available memory in bytes. - pub(crate) available_bytes: u64, - /// Whether the limits are host- or cgroup-derived. - pub(crate) basis: MemoryBasis, -} - -/// Reconciles host memory with an optional cgroup limit. -/// -/// `cgroup` carries `(total_memory, free_memory)` as reported for the cgroup -/// hierarchy (sysinfo already caps `total_memory` at the host total). A cgroup -/// only counts when it actually constrains below the host, so an unlimited -/// cgroup transparently falls back to host values. -pub(crate) fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Option<(u64, u64)>) -> EffectiveMemory { - match cgroup { - Some((cgroup_total, cgroup_free)) if cgroup_total > 0 && cgroup_total < host_total => EffectiveMemory { - total_bytes: cgroup_total, - available_bytes: cgroup_free.min(cgroup_total), - basis: MemoryBasis::Cgroup, - }, - _ => EffectiveMemory { - total_bytes: host_total, - available_bytes: host_available, - basis: MemoryBasis::Host, - }, - } -} - -/// Resolves the effective memory from an already-refreshed system handle. -/// -/// `cgroup_limits()` is computed fresh on each call and is only implemented on -/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host -/// values. -pub(crate) fn effective_memory_from_system(system: &System) -> EffectiveMemory { - let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory)); - select_effective_memory(system.total_memory(), system.available_memory(), cgroup) -} - -/// Resolves the effective memory using a fresh, memory-refreshed system handle. -pub(crate) fn resolve_effective_memory() -> EffectiveMemory { - let mut system = System::new(); - system.refresh_memory(); - effective_memory_from_system(&system) -} - /// Immutable memory snapshot used by the cache fill gate. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ObjectDataCacheMemorySnapshot { @@ -797,10 +730,7 @@ fn lock_or_recover(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { #[cfg(test)] mod tests { - use super::{ - DEFAULT_TELEMETRY_STALENESS, MemoryBasis, ObjectDataCacheMemoryGate, ObjectDataCacheMemorySnapshot, - select_effective_memory, - }; + use super::{DEFAULT_TELEMETRY_STALENESS, ObjectDataCacheMemoryGate, ObjectDataCacheMemorySnapshot}; use crate::config::ObjectDataCacheConfig; use crate::stats::ObjectDataCacheStats; use bytes::Bytes; @@ -1240,42 +1170,6 @@ mod tests { assert_eq!(gate.claimed_bytes_for_test(), 0); } - #[test] - fn select_effective_memory_prefers_constraining_cgroup() { - let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB))); - - assert_eq!(effective.basis, MemoryBasis::Cgroup); - assert_eq!(effective.total_bytes, 2 * GIB); - assert_eq!(effective.available_bytes, GIB); - } - - #[test] - fn select_effective_memory_ignores_non_constraining_cgroup() { - // A cgroup total equal to (or above) the host total means no real limit. - let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((64 * GIB, 10 * GIB))); - - assert_eq!(effective.basis, MemoryBasis::Host); - assert_eq!(effective.total_bytes, 64 * GIB); - assert_eq!(effective.available_bytes, 40 * GIB); - } - - #[test] - fn select_effective_memory_falls_back_to_host_without_cgroup() { - let effective = select_effective_memory(8 * GIB, 4 * GIB, None); - - assert_eq!(effective.basis, MemoryBasis::Host); - assert_eq!(effective.total_bytes, 8 * GIB); - assert_eq!(effective.available_bytes, 4 * GIB); - } - - #[test] - fn select_effective_memory_caps_available_at_total() { - let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, 3 * GIB))); - - assert_eq!(effective.total_bytes, 2 * GIB); - assert_eq!(effective.available_bytes, 2 * GIB); - } - #[test] fn gate_pauses_fill_when_container_memory_is_low() { // Simulate a pod-sized snapshot (256 MiB total, 16 MiB free): below the diff --git a/crates/object-data-cache/src/runtime_memory.rs b/crates/object-data-cache/src/runtime_memory.rs new file mode 100644 index 000000000..06eec2ea0 --- /dev/null +++ b/crates/object-data-cache/src/runtime_memory.rs @@ -0,0 +1,124 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use sysinfo::System; + +/// Source used to resolve the effective memory limits. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MemoryBasis { + /// Limits come from host memory reported by sysinfo. + Host, + /// Limits come from a constraining cgroup (container) memory limit. + Cgroup, +} + +impl MemoryBasis { + pub const fn as_str(self) -> &'static str { + match self { + Self::Host => "host", + Self::Cgroup => "cgroup", + } + } +} + +/// Effective memory totals after reconciling host memory with cgroup limits. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EffectiveMemory { + /// Effective total memory in bytes. + pub total_bytes: u64, + /// Effective available memory in bytes. + pub available_bytes: u64, + /// Whether the limits are host- or cgroup-derived. + pub basis: MemoryBasis, +} + +/// Reconciles host memory with an optional cgroup limit. +/// +/// `cgroup` carries `(total_memory, free_memory)` as reported for the cgroup +/// hierarchy (sysinfo already caps `total_memory` at the host total). A cgroup +/// only counts when it actually constrains below the host, so an unlimited +/// cgroup transparently falls back to host values. +pub fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Option<(u64, u64)>) -> EffectiveMemory { + match cgroup { + Some((cgroup_total, cgroup_free)) if cgroup_total > 0 && cgroup_total < host_total => EffectiveMemory { + total_bytes: cgroup_total, + available_bytes: cgroup_free.min(cgroup_total), + basis: MemoryBasis::Cgroup, + }, + _ => EffectiveMemory { + total_bytes: host_total, + available_bytes: host_available, + basis: MemoryBasis::Host, + }, + } +} + +/// Resolves the effective memory from an already-refreshed system handle. +/// +/// `cgroup_limits()` is computed fresh on each call and is only implemented on +/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host +/// values. +pub fn effective_memory_from_system(system: &System) -> EffectiveMemory { + let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory)); + select_effective_memory(system.total_memory(), system.available_memory(), cgroup) +} + +/// Resolves the effective memory using a fresh, memory-refreshed system handle. +pub fn resolve_effective_memory() -> EffectiveMemory { + let mut system = System::new(); + system.refresh_memory(); + effective_memory_from_system(&system) +} + +#[cfg(test)] +mod tests { + use super::{MemoryBasis, select_effective_memory}; + + const GIB: u64 = 1024 * 1024 * 1024; + + #[test] + fn select_effective_memory_prefers_constraining_cgroup() { + let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB))); + + assert_eq!(effective.basis, MemoryBasis::Cgroup); + assert_eq!(effective.total_bytes, 2 * GIB); + assert_eq!(effective.available_bytes, GIB); + } + + #[test] + fn select_effective_memory_ignores_non_constraining_cgroup() { + let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((64 * GIB, 10 * GIB))); + + assert_eq!(effective.basis, MemoryBasis::Host); + assert_eq!(effective.total_bytes, 64 * GIB); + assert_eq!(effective.available_bytes, 40 * GIB); + } + + #[test] + fn select_effective_memory_falls_back_to_host_without_cgroup() { + let effective = select_effective_memory(8 * GIB, 4 * GIB, None); + + assert_eq!(effective.basis, MemoryBasis::Host); + assert_eq!(effective.total_bytes, 8 * GIB); + assert_eq!(effective.available_bytes, 4 * GIB); + } + + #[test] + fn select_effective_memory_caps_available_at_total() { + let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, 3 * GIB))); + + assert_eq!(effective.total_bytes, 2 * GIB); + assert_eq!(effective.available_bytes, 2 * GIB); + } +} diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 76382fa00..5884a5cbd 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -235,7 +235,7 @@ rustfs-zip = { workspace = true } rustfs-io-core = { workspace = true } rustfs-io-metrics = { workspace = true } rustfs-object-capacity = { workspace = true } -rustfs-object-data-cache = { workspace = true } +rustfs-object-data-cache = { workspace = true, features = ["cache"] } rustfs-concurrency = { workspace = true } rustfs-scanner = { workspace = true } tempfile = { workspace = true }