From 7484a61fa3e64f8c06524163ff79a838bad7c969 Mon Sep 17 00:00:00 2001 From: wood Date: Tue, 30 Jun 2026 19:05:05 +0800 Subject: [PATCH] feat(compression): show cluster-level compression stat in grafana (#4112) --- .../grafana/dashboards/rustfs.json | 572 ++++++++++++++++++ crates/data-usage/src/data_usage.rs | 11 + crates/ecstore/src/api/mod.rs | 11 +- crates/ecstore/src/data_usage/mod.rs | 285 ++++++++- crates/ecstore/src/io_support/compress.rs | 4 + crates/ecstore/src/set_disk/mod.rs | 7 + .../obs/src/metrics/collectors/compression.rs | 110 ++++ crates/obs/src/metrics/collectors/mod.rs | 2 + crates/obs/src/metrics/mod.rs | 5 +- crates/obs/src/metrics/scheduler.rs | 33 +- crates/obs/src/metrics/schema/compression.rs | 61 ++ .../src/metrics/schema/entry/metric_name.rs | 14 + .../obs/src/metrics/schema/entry/subsystem.rs | 4 + crates/obs/src/metrics/schema/mod.rs | 1 + crates/obs/src/metrics/stats_collector.rs | 32 +- crates/obs/src/metrics/storage_api.rs | 6 +- rustfs/src/startup_observability.rs | 7 +- rustfs/src/startup_services.rs | 2 +- rustfs/src/startup_shutdown.rs | 11 + rustfs/src/storage/storage_api.rs | 16 +- rustfs/src/storage_api.rs | 8 +- 21 files changed, 1173 insertions(+), 29 deletions(-) create mode 100644 crates/obs/src/metrics/collectors/compression.rs create mode 100644 crates/obs/src/metrics/schema/compression.rs diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index fca4ef27e..7a6768f8e 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -10928,6 +10928,578 @@ ], "title": "Host CPU Usage by Instance", "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 320 + }, + "id": 512, + "panels": [], + "title": "Disk Compression", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "bytes" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 0, + "y": 321 + }, + "id": 513, + "options": { + "colorMode": "background", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_saved_bytes_total{job=~\"$job\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Bytes Saved", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "fixedColor": "red", + "mode": "fixed" + }, + "mappings": [], + "max": 1, + "min": 0, + "unit": "percentunit" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 4, + "x": 5, + "y": 321 + }, + "id": 514, + "options": { + "colorMode": "background", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "rustfs_compression_compression_ratio{job=~\"$job\"}", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Compression Ratio", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "fixedColor": "#FF7383", + "mode": "fixed" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "text", + "value": null + } + ] + }, + "unit": "bytes" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 9, + "y": 321 + }, + "id": 517, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_original_bytes_total{job=~\"$job\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Original Bytes", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "fixedColor": "#73BF69", + "mode": "fixed" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "text", + "value": null + } + ] + }, + "unit": "bytes" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 14, + "y": 321 + }, + "id": 518, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_compressed_bytes_total{job=~\"$job\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Compressed Bytes", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "fixedColor": "blue", + "mode": "fixed" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "text", + "value": null + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 5, + "x": 19, + "y": 321 + }, + "id": 519, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_compression_operation_total{job=~\"$job\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Total Ops", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "bytes" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Original Bytes" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "#FF7383", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Compressed Bytes" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "#73BF69", + "mode": "fixed" + } + } + ] + } + ] + }, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 325 + }, + "id": 515, + "options": { + "legend": { + "calcs": [ + "lastNotNull" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_original_bytes_total{job=~\"$job\"})", + "legendFormat": "Original Bytes", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rustfs_compression_compressed_bytes_total{job=~\"$job\"})", + "legendFormat": "Compressed Bytes", + "range": true, + "refId": "B" + } + ], + "title": "Original vs Compressed Bytes", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 325 + }, + "id": 516, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(rate(rustfs_compression_compression_operation_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "Ops/s", + "range": true, + "refId": "A" + } + ], + "title": "Compression Operations Rate", + "type": "timeseries" } ], "preload": false, diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index af176630b..a34238cb4 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -1257,6 +1257,17 @@ impl SizeSummary { } } +/// Aggregated compression metrics: original size, compressed size, and operation count. +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct CompressionTotalInfo { + // Total bytes before compression since compression is used. + pub original_bytes_total: u64, + // Total bytes after compression since compression is used. + pub compressed_bytes_total: u64, + // Total number of compression operations since compression is used. + pub compression_operations_total: u64, +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 2d130e66c..1bd46194f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -57,7 +57,7 @@ pub mod cluster { } pub mod compression { - pub use crate::io_support::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; + pub use crate::io_support::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled}; } pub mod config { @@ -69,10 +69,11 @@ pub mod config { pub mod data_usage { pub use crate::data_usage::{ - DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, load_data_usage_from_backend, - record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, - record_bucket_object_write_memory, refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, - replace_bucket_usage_memory_from_info, + DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, + load_compression_total_from_memory, load_data_usage_from_backend, record_bucket_delete_marker_memory, + record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, + record_compression_total_memory, refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, + replace_bucket_usage_memory_from_info, store_compression_total_in_backend, }; } diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 512b76281..f35b4984c 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -32,13 +32,14 @@ use crate::{ }; pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path}; use rustfs_data_usage::{ - BucketTargetUsageInfo, BucketUsageInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, SizeSummary, + BucketTargetUsageInfo, BucketUsageInfo, CompressionTotalInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, + SizeSummary, }; use rustfs_io_metrics::record_system_path_failure; use rustfs_utils::path::SLASH_SEPARATOR; use std::{ collections::{HashMap, HashSet, hash_map::Entry}, - sync::{Arc, OnceLock}, + sync::{Arc, LazyLock, OnceLock}, time::{Duration, SystemTime}, }; use tokio::fs; @@ -48,6 +49,7 @@ use tracing::{debug, error, info, instrument}; // Data usage storage constants pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR; const DATA_USAGE_OBJ_NAME: &str = ".usage.json"; +const DATA_COMPRESSION_TOTAL_NAME: &str = ".compression.json"; const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin"; pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin"; const DATA_USAGE_CACHE_TTL_SECS: u64 = 30; @@ -69,6 +71,37 @@ type CacheUpdating = Arc>; static USAGE_MEMORY_CACHE: OnceLock = OnceLock::new(); static USAGE_CACHE_UPDATING: OnceLock = OnceLock::new(); +/// Deferred persist thresholds for compression totals: persist after this many +/// operations recorded, but no more often than the min interval. +const COMPRESSION_PERSIST_BATCH_SIZE: u64 = 100; +const COMPRESSION_PERSIST_MIN_INTERVAL: Duration = Duration::from_secs(30); + +/// In-memory compression accumulator with debounced persistence state. +#[derive(Debug, Clone)] +struct CompressionTotalState { + info: CompressionTotalInfo, + /// Operations recorded since the last persist attempt. + ops_since_persist: u64, + /// When the last persist was attempted (used for min-interval gating). + last_persist: tokio::time::Instant, + /// When `true`, recording is skipped (embedded mode without observability). + inited: bool, +} + +impl Default for CompressionTotalState { + fn default() -> Self { + Self { + info: CompressionTotalInfo::default(), + ops_since_persist: 0, + last_persist: tokio::time::Instant::now(), + inited: false, + } + } +} + +static COMPRESSION_TOTAL_MEMORY_CACHE: LazyLock>>> = + LazyLock::new(|| Arc::new(RwLock::new(Some(CompressionTotalState::default())))); + fn memory_cache() -> &'static UsageMemoryCache { USAGE_MEMORY_CACHE.get_or_init(|| Arc::new(RwLock::new(HashMap::new()))) } @@ -94,6 +127,11 @@ lazy_static::lazy_static! { SLASH_SEPARATOR, DATA_USAGE_BLOOM_NAME ); + pub static ref DATA_COMPRESSION_TOTAL_NAME_PATH: String = format!("{}{}{}", + crate::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_COMPRESSION_TOTAL_NAME + ); } /// Store data usage info to backend storage @@ -1007,6 +1045,138 @@ pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate: Ok(()) } +/// Persist the current in-memory compression total to the backend. +/// Resets the debounce counter so the next auto-persist won't fire +/// immediately after this manual flush (intended for shutdown paths). +/// Storage will not be triggered when uninitialized. +pub async fn store_compression_total_in_backend() { + let snapshot = { + let mut guard = COMPRESSION_TOTAL_MEMORY_CACHE.write().await; + let Some(state) = guard.as_mut() else { + return; + }; + if !state.inited { + return; + } + state.ops_since_persist = 0; + state.last_persist = tokio::time::Instant::now(); + state.info.clone() + }; + try_flush_compression_total(&snapshot).await; +} + +/// Load compression total info from backend storage +#[instrument(skip(store))] +pub async fn load_compression_total_from_backend(store: Arc) -> Result { + let buf: Vec = match read_config(store.clone(), &DATA_COMPRESSION_TOTAL_NAME_PATH).await { + Ok(data) => data, + Err(e) => { + if e == Error::ConfigNotFound { + return Ok(CompressionTotalInfo::default()); + } + let reason = classify_system_path_failure_reason(&e); + record_system_path_failure("compression_total", "read_primary", reason); + error!( + path_kind = "compression_total", + operation = "read_primary", + reason, + object = %DATA_COMPRESSION_TOTAL_NAME_PATH.as_str(), + error = %e, + "system path read failed" + ); + return Err(Error::other(e)); + } + }; + let compression_total: CompressionTotalInfo = + serde_json::from_slice(&buf).map_err(|e| Error::other(format!("Failed to deserialize compression total info: {e}")))?; + + info!( + "Loaded compression total info: original={}, compressed={}, operations={}", + compression_total.original_bytes_total, + compression_total.compressed_bytes_total, + compression_total.compression_operations_total + ); + + Ok(compression_total) +} + +pub async fn load_compression_total_from_memory() -> Option { + COMPRESSION_TOTAL_MEMORY_CACHE.read().await.as_ref().map(|s| s.info.clone()) +} + +/// Record a compression operation. Accumulates totals in memory and triggers a +/// background persist to backend after every `COMPRESSION_PERSIST_BATCH_SIZE` +/// operations, gated by a minimum interval to avoid excessive IO under high +/// throughput. +pub async fn record_compression_total_memory(original_size: u64, compressed_size: u64) { + let mut guard = COMPRESSION_TOTAL_MEMORY_CACHE.write().await; + let Some(state) = guard.as_mut() else { + error!("compression total memory cache not initialized, discarding record"); + return; + }; + // Compression totals are only recorded while this cache is initialized. + if !state.inited { + return; + } + state.info.original_bytes_total = state.info.original_bytes_total.saturating_add(original_size); + state.info.compressed_bytes_total = state.info.compressed_bytes_total.saturating_add(compressed_size); + state.info.compression_operations_total = state.info.compression_operations_total.saturating_add(1); + state.ops_since_persist = state.ops_since_persist.saturating_add(1); + if state.ops_since_persist >= COMPRESSION_PERSIST_BATCH_SIZE + || state.last_persist.elapsed() >= COMPRESSION_PERSIST_MIN_INTERVAL + { + state.ops_since_persist = 0; + state.last_persist = tokio::time::Instant::now(); + // Fire-and-forget: hold no lock during IO + let info = state.info.clone(); + drop(guard); + tokio::spawn(async move { + try_flush_compression_total(&info).await; + }); + } +} + +/// Persist the given `CompressionTotalInfo` snapshot to backend storage. +/// Used by both the debounce path and the manual flush path. +async fn try_flush_compression_total(info: &CompressionTotalInfo) { + let Some(store) = runtime_sources::object_store_handle() else { + error!("object store not initialized, skipping compression total persist"); + return; + }; + match serde_json::to_vec(info) { + Ok(data) => { + if let Err(e) = crate::config::com::save_config(store.clone(), &DATA_COMPRESSION_TOTAL_NAME_PATH, data).await { + error!("Failed to persist compression total to backend: {}", e); + } + } + Err(e) => { + error!("Failed to serialize compression total info: {}", e); + } + } +} + +/// Initialize the compression total memory cache from backend storage. +/// Should be called at startup after the store is available if compression is enabled. +/// Compression totals are only recorded while this cache is initialized, so this must +/// be called before any compression operations are recorded. +#[instrument(skip(store))] +pub async fn init_compression_total_memory_from_backend(store: Arc) { + let info = match load_compression_total_from_backend(store).await { + Ok(info) => info, + Err(e) => { + // If load fails (e.g. ConfigNotFound or corrupt backup), start from zero. + info!("Failed to init compression total from backend, starting from zero: {}", e); + CompressionTotalInfo::default() + } + }; + *COMPRESSION_TOTAL_MEMORY_CACHE.write().await = Some(CompressionTotalState { + info, + ops_since_persist: 0, + last_persist: tokio::time::Instant::now(), + inited: true, + }); +} + #[cfg(test)] mod tests { use super::*; @@ -1515,4 +1685,115 @@ mod tests { Some((1000, 10_000)) ); } + + // --- CompressionTotalState tests --- + + /// Reset the compression total cache to a known state for isolated tests. + async fn reset_compression_cache(state: CompressionTotalState) { + *COMPRESSION_TOTAL_MEMORY_CACHE.write().await = Some(state); + } + + #[test] + fn compression_state_default_values() { + let state = CompressionTotalState::default(); + assert!(!state.inited, "default state should not be inited"); + assert_eq!(state.ops_since_persist, 0); + assert_eq!(state.info.original_bytes_total, 0); + assert_eq!(state.info.compressed_bytes_total, 0); + assert_eq!(state.info.compression_operations_total, 0); + } + + #[tokio::test] + #[serial] + async fn record_compression_skips_when_not_inited() { + let mut pre_state = CompressionTotalState::default(); + // Set known values that should remain unchanged when inited=false + pre_state.info.original_bytes_total = 42; + pre_state.info.compressed_bytes_total = 21; + pre_state.info.compression_operations_total = 7; + pre_state.ops_since_persist = 5; + // inited is already false from default() + + reset_compression_cache(pre_state.clone()).await; + + // Call with sizes that would be accumulated if inited were true + record_compression_total_memory(100, 50).await; + + let guard = COMPRESSION_TOTAL_MEMORY_CACHE.read().await; + let state = guard.as_ref().expect("cache should contain a state"); + assert!(!state.inited); + assert_eq!(state.info.original_bytes_total, 42, "original should be unchanged"); + assert_eq!(state.info.compressed_bytes_total, 21, "compressed should be unchanged"); + assert_eq!(state.info.compression_operations_total, 7, "operations should be unchanged"); + assert_eq!(state.ops_since_persist, 5, "ops_since_persist should be unchanged"); + } + + #[tokio::test] + #[serial] + async fn record_compression_accumulates_totals() { + let state = CompressionTotalState { + info: CompressionTotalInfo::default(), + ops_since_persist: 0, + last_persist: tokio::time::Instant::now(), + inited: true, + }; + + reset_compression_cache(state).await; + + for _ in 0..3 { + record_compression_total_memory(100, 50).await; + } + + let guard = COMPRESSION_TOTAL_MEMORY_CACHE.read().await; + let state = guard.as_ref().expect("cache should contain a state"); + assert!(state.inited); + assert_eq!(state.info.original_bytes_total, 300); + assert_eq!(state.info.compressed_bytes_total, 150); + assert_eq!(state.info.compression_operations_total, 3); + assert_eq!(state.ops_since_persist, 3); + } + + #[tokio::test] + #[serial] + async fn record_compression_triggers_persist_on_batch_full() { + // Simulate 99 previous records, so the next (100th) hits the batch threshold. + let state = CompressionTotalState { + info: CompressionTotalInfo { + original_bytes_total: 9900, + compressed_bytes_total: 4950, + compression_operations_total: 99, + }, + ops_since_persist: 99, + last_persist: tokio::time::Instant::now(), + inited: true, + }; + + reset_compression_cache(state).await; + + // 100th record — should trigger the debounce flush. + record_compression_total_memory(100, 50).await; + + let guard = COMPRESSION_TOTAL_MEMORY_CACHE.read().await; + let state = guard.as_ref().expect("cache should contain a state"); + + // Totals are correctly accumulated (not reset by the flush). + assert_eq!(state.info.original_bytes_total, 10_000); // 9900 + 100 + assert_eq!(state.info.compressed_bytes_total, 5_000); // 4950 + 50 + assert_eq!(state.info.compression_operations_total, 100); // 99 + 1 + + // Debounce counter is reset after the persist trigger. + assert_eq!(state.ops_since_persist, 0); + + // last_persist was refreshed; it should be very recent. + assert!( + state.last_persist.elapsed() < Duration::from_secs(3), + "last_persist should have been refreshed to now" + ); + + // try_flush_compression_total is spawned asynchronously here. + // In a unit test, runtime_sources::object_store_handle() returns None, + // so the spawned task will hit the "object store not initialized" error + // path and return gracefully. The state mutation (counter reset + totals + // accumulation) is the critical behaviour verified above. + } } diff --git a/crates/ecstore/src/io_support/compress.rs b/crates/ecstore/src/io_support/compress.rs index b4187319b..356974afe 100644 --- a/crates/ecstore/src/io_support/compress.rs +++ b/crates/ecstore/src/io_support/compress.rs @@ -167,6 +167,10 @@ pub fn is_disk_compressible(headers: &http::HeaderMap, object_name: &str) -> boo is_disk_compressible_with_config(headers, object_name, DISK_COMPRESSION_CONFIG.get_or_init(parse_disk_compression_config)) } +pub fn is_disk_compression_enabled() -> bool { + DISK_COMPRESSION_CONFIG.get_or_init(parse_disk_compression_config).enabled +} + fn is_disk_compressible_with_config(headers: &http::HeaderMap, object_name: &str, config: &DiskCompressionConfig) -> bool { // Check if disk compression is enabled (read once at first use, then fixed for process lifetime) if !config.enabled { diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index abb1c17b3..ef9fda9e1 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -25,6 +25,7 @@ use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl}; use crate::cluster::rpc::heal_bucket_local_on_disks; +use crate::data_usage::record_compression_total_memory; use crate::diagnostics::get::{ GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, @@ -2214,6 +2215,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } + if fi.is_compressed() { + record_compression_total_memory(actual_size as u64, w_size as u64).await; + } record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); fi.replication_state_internal = Some(opts.put_replication_state()); @@ -4395,6 +4399,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } + if fi.is_compressed() { + record_compression_total_memory(actual_size as u64, w_size as u64).await; + } let checksums = data.as_hash_reader().content_crc(); let part_info = ObjectPartInfo { diff --git a/crates/obs/src/metrics/collectors/compression.rs b/crates/obs/src/metrics/collectors/compression.rs new file mode 100644 index 000000000..237aa36a6 --- /dev/null +++ b/crates/obs/src/metrics/collectors/compression.rs @@ -0,0 +1,110 @@ +// 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 crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::compression::*; + +/// Cluster-level compression statistics. +#[derive(Debug, Clone, Default)] +pub struct CompressionClusterStats { + pub original_bytes_total: u64, + pub compressed_bytes_total: u64, + pub bytes_saved_total: u64, + pub compression_ratio: f64, + pub compression_operations_total: u64, +} + +/// Collect cluster-level compression metrics from cluster stats. +pub fn collect_compression_cluster_metrics(stats: &CompressionClusterStats) -> Vec { + vec![ + PrometheusMetric::from_descriptor(&COMPRESSION_OPERATIONS_TOTAL, stats.compression_operations_total as f64), + PrometheusMetric::from_descriptor(&COMPRESSION_BYTES_ORIGINAL_TOTAL, stats.original_bytes_total as f64), + PrometheusMetric::from_descriptor(&COMPRESSION_BYTES_COMPRESSED_TOTAL, stats.compressed_bytes_total as f64), + PrometheusMetric::from_descriptor(&COMPRESSION_BYTES_SAVED_TOTAL, stats.bytes_saved_total as f64), + PrometheusMetric::from_descriptor(&COMPRESSION_RATIO, stats.compression_ratio), + ] +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::metrics::report::report_metrics; + + #[test] + fn test_collect_compression_cluster_metrics() { + let stats = CompressionClusterStats { + original_bytes_total: 1000, + compressed_bytes_total: 700, + bytes_saved_total: 300, + compression_ratio: 0.7, + compression_operations_total: 10, + }; + + let metrics = collect_compression_cluster_metrics(&stats); + report_metrics(&metrics); + + assert_eq!(metrics.len(), 5); + + // Verify original bytes + let name = COMPRESSION_BYTES_ORIGINAL_TOTAL.get_full_metric_name(); + let m = metrics.iter().find(|m| m.name == name && m.value == 1000.0); + assert!(m.is_some(), "missing or wrong value for original_bytes_total"); + + // Verify compressed bytes + let name = COMPRESSION_BYTES_COMPRESSED_TOTAL.get_full_metric_name(); + let m = metrics.iter().find(|m| m.name == name && m.value == 700.0); + assert!(m.is_some(), "missing or wrong value for compressed_bytes_total"); + + // Verify saved bytes + let name = COMPRESSION_BYTES_SAVED_TOTAL.get_full_metric_name(); + let m = metrics.iter().find(|m| m.name == name && m.value == 300.0); + assert!(m.is_some(), "missing or wrong value for bytes_saved_total"); + + // Verify compression ratio + let name = COMPRESSION_RATIO.get_full_metric_name(); + let m = metrics.iter().find(|m| m.name == name && m.value == 0.7); + assert!(m.is_some(), "missing or wrong value for compression_ratio"); + + // Verify operations total + let name = COMPRESSION_OPERATIONS_TOTAL.get_full_metric_name(); + let m = metrics.iter().find(|m| m.name == name && m.value == 10.0); + assert!(m.is_some(), "missing or wrong value for compression_operations_total"); + } + + #[test] + fn test_collect_compression_cluster_metrics_empty() { + let stats = CompressionClusterStats::default(); + + let metrics = collect_compression_cluster_metrics(&stats); + report_metrics(&metrics); + + assert_eq!(metrics.len(), 5); + + // All values should be zero + for metric in &metrics { + assert_eq!(metric.value, 0.0, "expected zero value for {}", metric.name); + assert!(metric.labels.is_empty()); + } + } + + #[test] + fn test_compression_cluster_stats_default() { + let stats = CompressionClusterStats::default(); + assert_eq!(stats.original_bytes_total, 0); + assert_eq!(stats.compressed_bytes_total, 0); + assert_eq!(stats.bytes_saved_total, 0); + assert_eq!(stats.compression_ratio, 0.0); + assert_eq!(stats.compression_operations_total, 0); + } +} diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index dc0860322..902b3edb8 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -21,6 +21,7 @@ pub mod cluster_erasure_set; pub mod cluster_health; pub mod cluster_iam; pub mod cluster_usage; +pub mod compression; pub mod dial9; pub mod ilm; pub mod node; @@ -51,6 +52,7 @@ pub use cluster_erasure_set::{ErasureSetStats, collect_erasure_set_metrics}; pub use cluster_health::{ClusterHealthStats, collect_cluster_health_metrics}; pub use cluster_iam::{IamStats, collect_iam_metrics}; pub use cluster_usage::{BucketUsageStats, ClusterUsageStats, collect_bucket_usage_metrics, collect_cluster_usage_metrics}; +pub use compression::{CompressionClusterStats, collect_compression_cluster_metrics}; pub use dial9::{Dial9Stats, collect_dial9_metrics, is_dial9_enabled}; pub use ilm::{IlmStats, collect_ilm_metrics}; pub use node::{DiskStats, collect_node_metrics}; diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index 966b3b610..c4556c75a 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -33,6 +33,7 @@ pub use scheduler::{ pub(crate) use storage_api::metrics::{ BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsEcstoreResult, ObsReplicationStats, ObsStore, StorageAdminApi, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_global_replication_stats, obs_get_quota_config, - obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_load_data_usage_from_backend, - obs_resolve_object_store_handle, obs_transition_state_handle, + obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, + obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_resolve_object_store_handle, + obs_transition_state_handle, }; diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 95d19672a..428bba37b 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -42,6 +42,7 @@ use crate::metrics::collectors::{ collect_cluster_health_metrics, collect_cluster_metrics, collect_cluster_usage_metrics, + collect_compression_cluster_metrics, collect_cpu_metrics, collect_drive_count_metrics, collect_drive_detailed_metrics, @@ -70,6 +71,7 @@ use crate::metrics::config::{ ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, ENV_CLUSTER_METRICS_INTERVAL, ENV_DEFAULT_METRICS_INTERVAL, ENV_NODE_METRICS_INTERVAL, ENV_NOTIFICATION_METRICS_INTERVAL, ENV_RESOURCE_METRICS_INTERVAL, }; +use crate::metrics::obs_is_disk_compression_enabled; use crate::metrics::report::{PrometheusMetric, report_metrics}; use crate::metrics::runtime_sources::bucket_monitor_available; use crate::metrics::schema::bucket_replication::{ @@ -78,9 +80,10 @@ use crate::metrics::schema::bucket_replication::{ use crate::metrics::stats_collector::{ ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_detail_stats, collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats, - collect_disk_and_system_drive_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, - collect_ilm_metric_stats, collect_internode_network_stats, collect_process_metric_bundle, collect_replication_stats, - collect_scanner_metric_stats, collect_system_cpu_and_memory_stats_with, + collect_compression_cluster_stats, collect_disk_and_system_drive_stats, collect_erasure_set_stats, + collect_host_network_stats, collect_iam_stats, collect_ilm_metric_stats, collect_internode_network_stats, + collect_process_metric_bundle, collect_replication_stats, collect_scanner_metric_stats, + collect_system_cpu_and_memory_stats_with, }; use rustfs_audit::audit_target_metrics; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; @@ -801,6 +804,30 @@ pub fn init_metrics_runtime(token: CancellationToken) { } }); + // Spawn task for compression metrics. + if obs_is_disk_compression_enabled() { + let token_clone = token.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(cluster_interval); + loop { + tokio::select! { + _ = interval.tick() => { + if let Some(stats) = collect_compression_cluster_stats().await { + let metrics = collect_compression_cluster_metrics(&stats); + if !metrics.is_empty() { + report_metrics(&metrics); + } + } + } + _ = token_clone.cancelled() => { + warn!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "compression_cluster_stats", state = "cancelled", "metrics runtime state changed"); + return; + } + } + } + }); + } + // Spawn task for internode/system network metrics. let token_clone = token; tokio::spawn(async move { diff --git a/crates/obs/src/metrics/schema/compression.rs b/crates/obs/src/metrics/schema/compression.rs new file mode 100644 index 000000000..5e031e0f0 --- /dev/null +++ b/crates/obs/src/metrics/schema/compression.rs @@ -0,0 +1,61 @@ +// 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 crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems}; +use std::sync::LazyLock; + +pub static COMPRESSION_BYTES_ORIGINAL_TOTAL: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::OriginedBytesTotal, + "Total bytes before compression (original)", + &["compression"], + subsystems::COMPRESSION, + ) +}); + +pub static COMPRESSION_BYTES_COMPRESSED_TOTAL: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::CompressedBytesTotal, + "Total bytes after compression", + &["compression"], + subsystems::COMPRESSION, + ) +}); + +pub static COMPRESSION_BYTES_SAVED_TOTAL: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::SavedBytesTotal, + "Total bytes saved by compression", + &["compression"], + subsystems::COMPRESSION, + ) +}); + +pub static COMPRESSION_RATIO: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::CompressionRatio, + "Compression ratio (0.0 - 1.0)", + &["compression"], + subsystems::COMPRESSION, + ) +}); + +pub static COMPRESSION_OPERATIONS_TOTAL: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::CompressionOperationsTotal, + "Total number of compression operations performed", + &["compression"], + subsystems::COMPRESSION, + ) +}); diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs index da02a9090..1eb890edb 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -338,6 +338,13 @@ pub enum MetricName { ScannerPartialCycles, ScannerPartialCyclesByReason, + // Compression-related metrics + CompressedBytesTotal, + OriginedBytesTotal, + SavedBytesTotal, + CompressionRatio, + CompressionOperationsTotal, + // CPU system-related metrics SysCPUAvgIdle, SysCPUAvgIOWait, @@ -737,6 +744,13 @@ impl MetricName { Self::ScannerPartialCycles => "partial_cycles".to_string(), Self::ScannerPartialCyclesByReason => "partial_cycles_by_reason".to_string(), + // Compression-related metrics + Self::CompressedBytesTotal => "compressed_bytes".to_string(), + Self::OriginedBytesTotal => "original_bytes".to_string(), + Self::SavedBytesTotal => "saved_bytes".to_string(), + Self::CompressionRatio => "compression_ratio".to_string(), + Self::CompressionOperationsTotal => "compression_operation_total".to_string(), + // CPU system-related metrics Self::SysCPUAvgIdle => "avg_idle".to_string(), Self::SysCPUAvgIOWait => "avg_iowait".to_string(), diff --git a/crates/obs/src/metrics/schema/entry/subsystem.rs b/crates/obs/src/metrics/schema/entry/subsystem.rs index d0f0c611f..1c313f6e8 100644 --- a/crates/obs/src/metrics/schema/entry/subsystem.rs +++ b/crates/obs/src/metrics/schema/entry/subsystem.rs @@ -51,6 +51,7 @@ pub enum MetricSubsystem { Replication, Notification, Scanner, + Compression, // Custom paths Custom(String), @@ -93,6 +94,7 @@ impl MetricSubsystem { Self::Replication => "/replication", Self::Notification => "/notification", Self::Scanner => "/scanner", + Self::Compression => "/compression", // Custom paths Self::Custom(path) => path, @@ -141,6 +143,7 @@ impl MetricSubsystem { "/replication" => Self::Replication, "/notification" => Self::Notification, "/scanner" => Self::Scanner, + "/compression" => Self::Compression, // Treat other paths as custom subsystems _ => Self::Custom(path.to_string()), @@ -203,6 +206,7 @@ pub mod subsystems { pub const REPLICATION: MetricSubsystem = MetricSubsystem::Replication; pub const NOTIFICATION: MetricSubsystem = MetricSubsystem::Notification; pub const SCANNER: MetricSubsystem = MetricSubsystem::Scanner; + pub const COMPRESSION: MetricSubsystem = MetricSubsystem::Compression; } #[cfg(test)] diff --git a/crates/obs/src/metrics/schema/mod.rs b/crates/obs/src/metrics/schema/mod.rs index a82108700..be672fb17 100644 --- a/crates/obs/src/metrics/schema/mod.rs +++ b/crates/obs/src/metrics/schema/mod.rs @@ -22,6 +22,7 @@ pub mod cluster_health; pub mod cluster_iam; pub mod cluster_notification; pub mod cluster_usage; +pub mod compression; pub mod entry; pub mod ilm; pub mod node_bucket; diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 597397c33..d79ea3fdd 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -22,17 +22,17 @@ use crate::metrics::collectors::{ BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, BucketStats, BucketUsageStats, - ClusterConfigStats, ClusterHealthStats, ClusterStats, ClusterUsageStats, CpuStats, DiskStats, DriveCountStats, - DriveDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, - ProcessStatusType, ReplicationStats, ResourceStats, ScannerStats, + ClusterConfigStats, ClusterHealthStats, ClusterStats, ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, + DriveCountStats, DriveDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, + ProcessStats, ProcessStatusType, ReplicationStats, ResourceStats, ScannerStats, }; use crate::metrics::runtime_sources::{ ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot, replication_stats_handle, }; use crate::metrics::{ BucketOperations, BucketOptions, ObsEcstoreResult, ObsStore, StorageAdminApi, obs_get_quota_config, - obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_load_data_usage_from_backend, - obs_resolve_object_store_handle, + obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_load_compression_total_from_memory, + obs_load_data_usage_from_backend, obs_resolve_object_store_handle, }; use chrono::Utc; use rustfs_common::heal_channel::HealScanMode; @@ -1100,6 +1100,28 @@ pub async fn collect_scanner_metric_stats() -> Option { }) } +/// Collect cluster-level compression statistics. +pub async fn collect_compression_cluster_stats() -> Option { + let compression_data_usage = obs_load_compression_total_from_memory().await?; + + let original_bytes_total = compression_data_usage.original_bytes_total; + let compressed_bytes_total = compression_data_usage.compressed_bytes_total; + let bytes_saved_total = original_bytes_total.saturating_sub(compressed_bytes_total); + let compression_ratio = if original_bytes_total > 0 { + compressed_bytes_total as f64 / original_bytes_total as f64 + } else { + 0.0 + }; + let compression_operations_total = compression_data_usage.compression_operations_total; + + Some(CompressionClusterStats { + original_bytes_total, + compressed_bytes_total, + bytes_saved_total, + compression_ratio, + compression_operations_total, + }) +} #[cfg(test)] mod tests { use super::*; diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index 30bd45188..a132ebde1 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -21,6 +21,8 @@ pub(crate) use rustfs_ecstore::api::capacity::{ get_total_usable_capacity as obs_get_total_usable_capacity, get_total_usable_capacity_free as obs_get_total_usable_capacity_free, }; +pub(crate) use rustfs_ecstore::api::compression::is_disk_compression_enabled as obs_is_disk_compression_enabled; +pub(crate) use rustfs_ecstore::api::data_usage::load_compression_total_from_memory as obs_load_compression_total_from_memory; pub(crate) use rustfs_ecstore::api::data_usage::load_data_usage_from_backend as obs_load_data_usage_from_backend; pub(crate) use rustfs_ecstore::api::error::Result as ObsEcstoreResult; pub(crate) use rustfs_ecstore::api::runtime::{ @@ -36,7 +38,7 @@ pub(crate) mod metrics { pub(crate) use super::{ ObsBucketBandwidthMonitor, ObsEcstoreResult, ObsReplicationStats, ObsStore, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_global_replication_stats, obs_get_quota_config, obs_get_total_usable_capacity, - obs_get_total_usable_capacity_free, obs_load_data_usage_from_backend, obs_resolve_object_store_handle, - obs_transition_state_handle, + obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, obs_load_compression_total_from_memory, + obs_load_data_usage_from_backend, obs_resolve_object_store_handle, obs_transition_state_handle, }; } diff --git a/rustfs/src/startup_observability.rs b/rustfs/src/startup_observability.rs index ce7009557..fc30739c9 100644 --- a/rustfs/src/startup_observability.rs +++ b/rustfs/src/startup_observability.rs @@ -12,16 +12,21 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::sync::Arc; + use crate::init::{init_auto_tuner, init_update_check, print_server_info}; use crate::startup_runtime_sources; +use crate::storage_api::startup::storage::{ECStore, init_compression_total_memory_from_backend}; use tokio_util::sync::CancellationToken; -pub(crate) async fn init_observability_runtime(ctx: CancellationToken) { +pub(crate) async fn init_observability_runtime(store: Arc, ctx: CancellationToken) { print_server_info(); init_update_check(); crate::allocator_reclaim::init_allocator_reclaim(ctx.clone()); if startup_runtime_sources::observability_metric_enabled() { + // Load persisted compression stats into memory early, before any PUTs can occur. + init_compression_total_memory_from_backend(store).await; startup_runtime_sources::set_put_stage_metrics_enabled(true); startup_runtime_sources::set_get_stage_metrics_enabled(true); startup_runtime_sources::init_metrics_runtime(ctx.clone()); diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 0aa28d1f6..34ae92c3c 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -80,7 +80,7 @@ pub(crate) async fn init_startup_runtime_services( init_auth_integrations().await?; init_notification_runtime(endpoint_pools, buckets).await?; let enable_scanner = init_background_service_runtime(store.clone()).await?; - init_observability_runtime(ctx.clone()).await; + init_observability_runtime(store.clone(), ctx.clone()).await; Ok(StartupServiceRuntime { optional_runtimes, diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index 3fccdc283..741304560 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/rustfs/src/startup_shutdown.rs @@ -125,6 +125,17 @@ pub(crate) async fn run_startup_shutdown_sequence( ); } + // Persist in-memory compression totals to backend before subsystems shut down. + info!( + target: "rustfs::main::handle_shutdown", + event = "compression_total_persist", + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "persisting", + "Compression total persist started" + ); + crate::storage_api::startup::shutdown::store_compression_total_in_backend().await; + let optional_runtime_shutdowns = prepare_optional_runtime_shutdowns(optional_runtimes); info!( diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e57a4f5de..0f376dcd7 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -347,10 +347,10 @@ pub(crate) mod ecstore_config { pub(crate) mod ecstore_data_usage { pub(crate) use rustfs_ecstore::api::data_usage::{ - apply_bucket_usage_memory_overlay, load_data_usage_from_backend, record_bucket_delete_marker_memory, - record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, - refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, - replace_bucket_usage_memory_from_info, + apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend, + record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, + record_bucket_object_write_memory, refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, + replace_bucket_usage_memory_from_info, store_compression_total_in_backend, }; } @@ -1238,3 +1238,11 @@ pub(crate) type GetObjectReader = ::GetOb pub(crate) type ObjectInfo = ::ObjectInfo; pub(crate) type ObjectOptions = ::ObjectOptions; pub(crate) type PutObjReader = ::PutObjectReader; + +pub(crate) async fn store_compression_total_in_backend() { + ecstore_data_usage::store_compression_total_in_backend().await; +} + +pub(crate) async fn init_compression_total_memory_from_backend(store: Arc) { + ecstore_data_usage::init_compression_total_memory_from_backend(store).await +} diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 188c327e1..64cc687b4 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -210,14 +210,14 @@ pub(crate) mod startup { } pub(crate) mod shutdown { - pub(crate) use crate::storage::storage_api::shutdown_background_services; + pub(crate) use crate::storage::storage_api::{shutdown_background_services, store_compression_total_in_backend}; } pub(crate) mod storage { pub(crate) use crate::storage::storage_api::{ - ECStore, EndpointServerPools, init_background_replication, init_ecstore_config, init_global_config_sys, - init_local_disks, init_lock_clients, prewarm_local_disk_id_map, set_global_endpoints, try_migrate_server_config, - update_erasure_type, + ECStore, EndpointServerPools, init_background_replication, init_compression_total_memory_from_backend, + init_ecstore_config, init_global_config_sys, init_local_disks, init_lock_clients, prewarm_local_disk_id_map, + set_global_endpoints, try_migrate_server_config, update_erasure_type, }; } }