diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index c4556c75a..a083fc28d 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -31,9 +31,9 @@ pub use scheduler::{ init_metrics_runtime, metrics_runtime_controller_snapshot, metrics_runtime_status_snapshot, }; 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, + BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsEcstoreResult, ObsStore, StorageAdminApi, + obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_quota_config, 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, + obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, + obs_resolve_object_store_handle, obs_transition_state_handle, }; diff --git a/crates/obs/src/metrics/runtime_sources.rs b/crates/obs/src/metrics/runtime_sources.rs index bffc88416..0e1672a5a 100644 --- a/crates/obs/src/metrics/runtime_sources.rs +++ b/crates/obs/src/metrics/runtime_sources.rs @@ -13,14 +13,11 @@ // limitations under the License. use crate::metrics::{ - ObsBucketBandwidthMonitor, ObsReplicationStats, obs_expiry_state_handle, obs_get_global_bucket_monitor, - obs_get_global_replication_stats, obs_transition_state_handle, + ObsBucketBandwidthMonitor, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_transition_state_handle, }; use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot}; use std::sync::Arc; -pub(crate) type ObsReplicationStatsHandle = Arc; - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) struct ObsIlmRuntimeSnapshot { pub(crate) expiry_pending_tasks: u64, @@ -63,10 +60,6 @@ pub(crate) fn bucket_monitor_available() -> bool { bucket_monitor_handle().is_some() } -pub(crate) fn replication_stats_handle() -> Option { - obs_get_global_replication_stats() -} - pub(crate) async fn ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot { let expiry_state = obs_expiry_state_handle(); let transition_state = obs_transition_state_handle(); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index d79ea3fdd..e715acfd7 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -26,20 +26,19 @@ use crate::metrics::collectors::{ 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::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot}; 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_compression_total_from_memory, - obs_load_data_usage_from_backend, obs_resolve_object_store_handle, + BucketOperations, BucketOptions, ObsEcstoreResult, ObsStore, StorageAdminApi, obs_bucket_replication_stats_snapshot, + obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, + obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, + obs_resolve_object_store_handle, }; use chrono::Utc; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::global_metrics; use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{collections::HashMap, sync::Arc}; use sysinfo::{Networks, System}; use tracing::{instrument, warn}; @@ -79,14 +78,6 @@ fn usize_to_u64_saturating(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } -fn i64_to_u64_floor_zero(value: i64) -> u64 { - u64::try_from(value.max(0)).unwrap_or(0) -} - -fn i32_to_u64_floor_zero(value: i32) -> u64 { - u64::try_from(value.max(0)).unwrap_or(0) -} - async fn load_obs_data_usage_from_backend(store: Arc) -> ObsEcstoreResult { let data_usage = obs_load_data_usage_from_backend(store).await?; @@ -158,116 +149,67 @@ async fn obs_ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot { } async fn obs_bucket_replication_detail_stats() -> Vec { - let Some(stats) = replication_stats_handle() else { - return Vec::new(); - }; - - let all_bucket_stats = stats.get_all().await; - let mut buckets = Vec::with_capacity(all_bucket_stats.len()); - - for (bucket, bucket_stats) in all_bucket_stats { - let proxy = stats.get_proxy_stats(&bucket).await; - let mut total_failed_bytes = 0u64; - let mut total_failed_count = 0u64; - let mut last_min_failed_bytes = 0u64; - let mut last_min_failed_count = 0u64; - let mut last_hour_failed_bytes = 0u64; - let mut last_hour_failed_count = 0u64; - let mut sent_bytes = 0u64; - let mut sent_count = 0u64; - let mut targets = Vec::with_capacity(bucket_stats.stats.len()); - - for (target_arn, target_stats) in bucket_stats.stats { - total_failed_bytes += i64_to_u64_floor_zero(target_stats.fail_stats.size); - total_failed_count += i64_to_u64_floor_zero(target_stats.fail_stats.count); - - let last_min = target_stats.fail_stats.recent_since(Duration::from_secs(60)); - last_min_failed_bytes += i64_to_u64_floor_zero(last_min.size); - last_min_failed_count += i64_to_u64_floor_zero(last_min.count); - - let last_hour = target_stats.fail_stats.recent_since(Duration::from_secs(60 * 60)); - last_hour_failed_bytes += i64_to_u64_floor_zero(last_hour.size); - last_hour_failed_count += i64_to_u64_floor_zero(last_hour.count); - - sent_bytes += i64_to_u64_floor_zero(target_stats.replicated_size); - sent_count += i64_to_u64_floor_zero(target_stats.replicated_count); - - targets.push(BucketReplicationTargetStats { - target_arn, - bandwidth_limit_bytes_per_sec: i64_to_u64_floor_zero(target_stats.bandwidth_limit_bytes_per_sec), - current_bandwidth_bytes_per_sec: target_stats.current_bandwidth_bytes_per_sec, - latency_ms: target_stats.latency.curr, - }); - } - - buckets.push(BucketReplicationStats { - bucket, - total_failed_bytes, - total_failed_count, - last_min_failed_bytes, - last_min_failed_count, - last_hour_failed_bytes, - last_hour_failed_count, - sent_bytes, - sent_count, - proxied_get_requests_total: i64_to_u64_floor_zero(proxy.get_total), - proxied_get_requests_failures: i64_to_u64_floor_zero(proxy.get_failed), - proxied_head_requests_total: i64_to_u64_floor_zero(proxy.head_total), - proxied_head_requests_failures: i64_to_u64_floor_zero(proxy.head_failed), - proxied_put_requests_total: i64_to_u64_floor_zero(proxy.put_total), - proxied_put_requests_failures: i64_to_u64_floor_zero(proxy.put_failed), - proxied_put_tagging_requests_total: i64_to_u64_floor_zero(proxy.put_tag_total), - proxied_put_tagging_requests_failures: i64_to_u64_floor_zero(proxy.put_tag_failed), - proxied_get_tagging_requests_total: i64_to_u64_floor_zero(proxy.get_tag_total), - proxied_get_tagging_requests_failures: i64_to_u64_floor_zero(proxy.get_tag_failed), - proxied_delete_tagging_requests_total: i64_to_u64_floor_zero(proxy.delete_tag_total), - proxied_delete_tagging_requests_failures: i64_to_u64_floor_zero(proxy.delete_tag_failed), - targets, - }); - } - - buckets + obs_bucket_replication_stats_snapshot() + .await + .into_iter() + .map(|stats| BucketReplicationStats { + bucket: stats.bucket, + total_failed_bytes: stats.total_failed_bytes, + total_failed_count: stats.total_failed_count, + last_min_failed_bytes: stats.last_min_failed_bytes, + last_min_failed_count: stats.last_min_failed_count, + last_hour_failed_bytes: stats.last_hour_failed_bytes, + last_hour_failed_count: stats.last_hour_failed_count, + sent_bytes: stats.sent_bytes, + sent_count: stats.sent_count, + proxied_get_requests_total: stats.proxied_get_requests_total, + proxied_get_requests_failures: stats.proxied_get_requests_failures, + proxied_head_requests_total: stats.proxied_head_requests_total, + proxied_head_requests_failures: stats.proxied_head_requests_failures, + proxied_put_requests_total: stats.proxied_put_requests_total, + proxied_put_requests_failures: stats.proxied_put_requests_failures, + proxied_put_tagging_requests_total: stats.proxied_put_tagging_requests_total, + proxied_put_tagging_requests_failures: stats.proxied_put_tagging_requests_failures, + proxied_get_tagging_requests_total: stats.proxied_get_tagging_requests_total, + proxied_get_tagging_requests_failures: stats.proxied_get_tagging_requests_failures, + proxied_delete_tagging_requests_total: stats.proxied_delete_tagging_requests_total, + proxied_delete_tagging_requests_failures: stats.proxied_delete_tagging_requests_failures, + targets: stats + .targets + .into_iter() + .map(|target| BucketReplicationTargetStats { + target_arn: target.target_arn, + bandwidth_limit_bytes_per_sec: target.bandwidth_limit_bytes_per_sec, + current_bandwidth_bytes_per_sec: target.current_bandwidth_bytes_per_sec, + latency_ms: target.latency_ms, + }) + .collect(), + }) + .collect() } async fn obs_site_replication_stats() -> ReplicationStats { - let Some(stats) = replication_stats_handle() else { - return ReplicationStats::default(); - }; - - let site_metrics = stats.get_sr_metrics_for_node().await; let current_data_transfer_rate = obs_bucket_replication_bandwidth_stats() .into_iter() .flatten() .map(|stat| stat.current_bandwidth_bytes_per_sec) .sum::(); - - let all_bucket_stats = stats.get_all().await; - let average_data_transfer_rate = all_bucket_stats - .values() - .flat_map(|bucket| bucket.stats.values()) - .map(|stat| stat.xfer_rate_lrg.avg + stat.xfer_rate_sml.avg) - .sum::(); - let max_data_transfer_rate = all_bucket_stats - .values() - .flat_map(|bucket| bucket.stats.values()) - .map(|stat| stat.xfer_rate_lrg.peak + stat.xfer_rate_sml.peak) - .sum::(); - let recent_backlog_count = stats.mrf_stats.values().copied().filter(|value| *value > 0).sum::(); + let stats = obs_replication_site_stats_snapshot(current_data_transfer_rate).await; ReplicationStats { - average_active_workers: site_metrics.active_workers.avg, - average_queued_bytes: site_metrics.queued.avg.bytes, - average_queued_count: site_metrics.queued.avg.count, - average_data_transfer_rate, - active_workers: i32_to_u64_floor_zero(site_metrics.active_workers.curr), - current_data_transfer_rate, - last_minute_queued_bytes: i64_to_u64_floor_zero(site_metrics.queued.last_minute.bytes), - last_minute_queued_count: i64_to_u64_floor_zero(site_metrics.queued.last_minute.count), - max_active_workers: i32_to_u64_floor_zero(site_metrics.active_workers.max), - max_queued_bytes: i64_to_u64_floor_zero(site_metrics.queued.max.bytes), - max_queued_count: i64_to_u64_floor_zero(site_metrics.queued.max.count), - max_data_transfer_rate, - recent_backlog_count: i64_to_u64_floor_zero(recent_backlog_count), + average_active_workers: stats.average_active_workers, + average_queued_bytes: stats.average_queued_bytes, + average_queued_count: stats.average_queued_count, + average_data_transfer_rate: stats.average_data_transfer_rate, + active_workers: stats.active_workers, + current_data_transfer_rate: stats.current_data_transfer_rate, + last_minute_queued_bytes: stats.last_minute_queued_bytes, + last_minute_queued_count: stats.last_minute_queued_count, + max_active_workers: stats.max_active_workers, + max_queued_bytes: stats.max_queued_bytes, + max_queued_count: stats.max_queued_count, + max_data_transfer_rate: stats.max_data_transfer_rate, + recent_backlog_count: stats.recent_backlog_count, } } diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index a132ebde1..1b28f2e72 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -12,11 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::time::Duration; + pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config; -pub(crate) use rustfs_ecstore::api::bucket::replication::{ - ReplicationStats as ObsReplicationStats, get_global_replication_stats as obs_get_global_replication_stats, -}; +use rustfs_ecstore::api::bucket::replication::get_global_replication_stats; 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, @@ -32,13 +32,194 @@ pub(crate) use rustfs_ecstore::api::runtime::{ pub(crate) use rustfs_ecstore::api::storage::ECStore as ObsStore; use rustfs_storage_api as storage_contracts; +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct ObsBucketReplicationTargetStatsSnapshot { + pub(crate) target_arn: String, + pub(crate) bandwidth_limit_bytes_per_sec: u64, + pub(crate) current_bandwidth_bytes_per_sec: f64, + pub(crate) latency_ms: f64, +} + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct ObsBucketReplicationStatsSnapshot { + pub(crate) bucket: String, + pub(crate) total_failed_bytes: u64, + pub(crate) total_failed_count: u64, + pub(crate) last_min_failed_bytes: u64, + pub(crate) last_min_failed_count: u64, + pub(crate) last_hour_failed_bytes: u64, + pub(crate) last_hour_failed_count: u64, + pub(crate) sent_bytes: u64, + pub(crate) sent_count: u64, + pub(crate) proxied_get_requests_total: u64, + pub(crate) proxied_get_requests_failures: u64, + pub(crate) proxied_head_requests_total: u64, + pub(crate) proxied_head_requests_failures: u64, + pub(crate) proxied_put_requests_total: u64, + pub(crate) proxied_put_requests_failures: u64, + pub(crate) proxied_put_tagging_requests_total: u64, + pub(crate) proxied_put_tagging_requests_failures: u64, + pub(crate) proxied_get_tagging_requests_total: u64, + pub(crate) proxied_get_tagging_requests_failures: u64, + pub(crate) proxied_delete_tagging_requests_total: u64, + pub(crate) proxied_delete_tagging_requests_failures: u64, + pub(crate) targets: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq)] +pub(crate) struct ObsReplicationSiteStatsSnapshot { + pub(crate) average_active_workers: f64, + pub(crate) average_queued_bytes: i64, + pub(crate) average_queued_count: i64, + pub(crate) average_data_transfer_rate: f64, + pub(crate) active_workers: u64, + pub(crate) current_data_transfer_rate: f64, + pub(crate) last_minute_queued_bytes: u64, + pub(crate) last_minute_queued_count: u64, + pub(crate) max_active_workers: u64, + pub(crate) max_queued_bytes: u64, + pub(crate) max_queued_count: u64, + pub(crate) max_data_transfer_rate: f64, + pub(crate) recent_backlog_count: u64, +} + +fn i64_to_u64_floor_zero(value: i64) -> u64 { + u64::try_from(value.max(0)).unwrap_or(0) +} + +fn i32_to_u64_floor_zero(value: i32) -> u64 { + u64::try_from(value.max(0)).unwrap_or(0) +} + +pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec { + let Some(stats) = get_global_replication_stats() else { + return Vec::new(); + }; + + let all_bucket_stats = stats.get_all().await; + let mut buckets = Vec::with_capacity(all_bucket_stats.len()); + + for (bucket, bucket_stats) in all_bucket_stats { + let proxy = stats.get_proxy_stats(&bucket).await; + let mut total_failed_bytes = 0u64; + let mut total_failed_count = 0u64; + let mut last_min_failed_bytes = 0u64; + let mut last_min_failed_count = 0u64; + let mut last_hour_failed_bytes = 0u64; + let mut last_hour_failed_count = 0u64; + let mut sent_bytes = 0u64; + let mut sent_count = 0u64; + let mut targets = Vec::with_capacity(bucket_stats.stats.len()); + + for (target_arn, target_stats) in bucket_stats.stats { + total_failed_bytes = total_failed_bytes.saturating_add(i64_to_u64_floor_zero(target_stats.fail_stats.size)); + total_failed_count = total_failed_count.saturating_add(i64_to_u64_floor_zero(target_stats.fail_stats.count)); + + let last_min = target_stats.fail_stats.recent_since(Duration::from_secs(60)); + last_min_failed_bytes = last_min_failed_bytes.saturating_add(i64_to_u64_floor_zero(last_min.size)); + last_min_failed_count = last_min_failed_count.saturating_add(i64_to_u64_floor_zero(last_min.count)); + + let last_hour = target_stats.fail_stats.recent_since(Duration::from_secs(60 * 60)); + last_hour_failed_bytes = last_hour_failed_bytes.saturating_add(i64_to_u64_floor_zero(last_hour.size)); + last_hour_failed_count = last_hour_failed_count.saturating_add(i64_to_u64_floor_zero(last_hour.count)); + + sent_bytes = sent_bytes.saturating_add(i64_to_u64_floor_zero(target_stats.replicated_size)); + sent_count = sent_count.saturating_add(i64_to_u64_floor_zero(target_stats.replicated_count)); + + targets.push(ObsBucketReplicationTargetStatsSnapshot { + target_arn, + bandwidth_limit_bytes_per_sec: i64_to_u64_floor_zero(target_stats.bandwidth_limit_bytes_per_sec), + current_bandwidth_bytes_per_sec: target_stats.current_bandwidth_bytes_per_sec, + latency_ms: target_stats.latency.curr, + }); + } + + buckets.push(ObsBucketReplicationStatsSnapshot { + bucket, + total_failed_bytes, + total_failed_count, + last_min_failed_bytes, + last_min_failed_count, + last_hour_failed_bytes, + last_hour_failed_count, + sent_bytes, + sent_count, + proxied_get_requests_total: i64_to_u64_floor_zero(proxy.get_total), + proxied_get_requests_failures: i64_to_u64_floor_zero(proxy.get_failed), + proxied_head_requests_total: i64_to_u64_floor_zero(proxy.head_total), + proxied_head_requests_failures: i64_to_u64_floor_zero(proxy.head_failed), + proxied_put_requests_total: i64_to_u64_floor_zero(proxy.put_total), + proxied_put_requests_failures: i64_to_u64_floor_zero(proxy.put_failed), + proxied_put_tagging_requests_total: i64_to_u64_floor_zero(proxy.put_tag_total), + proxied_put_tagging_requests_failures: i64_to_u64_floor_zero(proxy.put_tag_failed), + proxied_get_tagging_requests_total: i64_to_u64_floor_zero(proxy.get_tag_total), + proxied_get_tagging_requests_failures: i64_to_u64_floor_zero(proxy.get_tag_failed), + proxied_delete_tagging_requests_total: i64_to_u64_floor_zero(proxy.delete_tag_total), + proxied_delete_tagging_requests_failures: i64_to_u64_floor_zero(proxy.delete_tag_failed), + targets, + }); + } + + buckets +} + +pub(crate) async fn obs_replication_site_stats_snapshot(current_data_transfer_rate: f64) -> ObsReplicationSiteStatsSnapshot { + let Some(stats) = get_global_replication_stats() else { + return ObsReplicationSiteStatsSnapshot::default(); + }; + + let site_metrics = stats.get_sr_metrics_for_node().await; + let all_bucket_stats = stats.get_all().await; + // These fields keep the existing metric semantics: cluster-wide sums across bucket targets. + let average_data_transfer_rate = all_bucket_stats + .values() + .flat_map(|bucket| bucket.stats.values()) + .map(|stat| stat.xfer_rate_lrg.avg + stat.xfer_rate_sml.avg) + .sum::(); + let max_data_transfer_rate = all_bucket_stats + .values() + .flat_map(|bucket| bucket.stats.values()) + .map(|stat| stat.xfer_rate_lrg.peak + stat.xfer_rate_sml.peak) + .sum::(); + let recent_backlog_count = stats.mrf_stats.values().copied().filter(|value| *value > 0).sum::(); + + ObsReplicationSiteStatsSnapshot { + average_active_workers: site_metrics.active_workers.avg, + average_queued_bytes: site_metrics.queued.avg.bytes, + average_queued_count: site_metrics.queued.avg.count, + average_data_transfer_rate, + active_workers: i32_to_u64_floor_zero(site_metrics.active_workers.curr), + current_data_transfer_rate, + last_minute_queued_bytes: i64_to_u64_floor_zero(site_metrics.queued.last_minute.bytes), + last_minute_queued_count: i64_to_u64_floor_zero(site_metrics.queued.last_minute.count), + max_active_workers: i32_to_u64_floor_zero(site_metrics.active_workers.max), + max_queued_bytes: i64_to_u64_floor_zero(site_metrics.queued.max.bytes), + max_queued_count: i64_to_u64_floor_zero(site_metrics.queued.max.count), + max_data_transfer_rate, + recent_backlog_count: i64_to_u64_floor_zero(recent_backlog_count), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn obs_replication_numeric_conversions_floor_negative_values() { + assert_eq!(i64_to_u64_floor_zero(-1), 0); + assert_eq!(i64_to_u64_floor_zero(42), 42); + assert_eq!(i32_to_u64_floor_zero(-1), 0); + assert_eq!(i32_to_u64_floor_zero(42), 42); + } +} + pub(crate) mod metrics { pub(crate) use super::storage_contracts::{BucketOperations, BucketOptions, StorageAdminApi}; 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_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, + ObsBucketBandwidthMonitor, ObsEcstoreResult, ObsStore, obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, + obs_get_global_bucket_monitor, obs_get_quota_config, 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_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle, }; } diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 4338952f8..56bf1c83c 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -189,6 +189,10 @@ Required contracts before crate movement: - `ReplicationObjectBridge`: app and SetDisks object write/delete replication decisions and scheduling are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_object_bridge.rs`. +- `ObsReplicationStatsSnapshot`: observability reads replication bucket/site + metrics through obs-local snapshot DTOs in + `crates/obs/src/metrics/storage_api.rs` instead of carrying the ECStore + replication stats handle through collectors. - `ReplicationScannerBridge`: scanner-originated replication heal scheduling is exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs`. diff --git a/docs/architecture/obs-ecstore-dependency-inventory.md b/docs/architecture/obs-ecstore-dependency-inventory.md index a756b2c5b..1cdb2fe96 100644 --- a/docs/architecture/obs-ecstore-dependency-inventory.md +++ b/docs/architecture/obs-ecstore-dependency-inventory.md @@ -21,7 +21,7 @@ until the contracts below are extracted. | `rustfs_ecstore::api::bucket::metadata_sys::get_quota_config` | `stats_collector.rs` | Behavior dependency | Reads per-bucket quota limits used in bucket usage metrics. | | `rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor` | `runtime_sources.rs`, `stats_collector.rs` | Type and runtime dependency | Reads replication bandwidth reports from the global bucket monitor handle. | | `rustfs_ecstore::api::runtime::bucket_monitor` | `runtime_sources.rs` | Runtime dependency | Resolves the global bucket bandwidth monitor for metric collection. | -| `rustfs_ecstore::api::bucket::replication::{GLOBAL_REPLICATION_STATS, ReplicationStats}` | `runtime_sources.rs`, `stats_collector.rs` | Type and runtime dependency | Reads replication status, transfer, failure, and site-replication stats. | +| `rustfs_ecstore::api::bucket::replication::get_global_replication_stats` | `storage_api.rs` snapshot helpers | Runtime dependency | Reads replication status, transfer, failure, and site-replication stats, then projects them into obs-local snapshot DTOs. | | `rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, GLOBAL_TransitionState}` | `runtime_sources.rs`, `stats_collector.rs` | Runtime dependency | Reads lifecycle expiry and transition queue counters. | | `rustfs_ecstore::api::error::Result` as `ObsEcstoreResult` | `stats_collector.rs` | Type dependency | Preserves ECStore error propagation while data-usage behavior remains ECStore-owned. | @@ -29,11 +29,10 @@ until the contracts below are extracted. The remaining coupling is not just a dependency declaration problem: -- type coupling: `ObsStore`, `ObsEcstoreResult`, `ObsReplicationStats`, - `ObsBucketBandwidthMonitor`, `StorageAdminApi`, `BucketOperations`, and - `BucketOptions`; +- type coupling: `ObsStore`, `ObsEcstoreResult`, `ObsBucketBandwidthMonitor`, + `StorageAdminApi`, `BucketOperations`, and `BucketOptions`; - runtime handle coupling: object-store handle, bucket monitor, replication - stats, expiry state, and transition state; + stats inside snapshot helpers, expiry state, and transition state; - behavior coupling: data-usage loading, quota lookup, and capacity math. Removing `rustfs-ecstore` from `crates/obs/Cargo.toml` is unsafe until those @@ -43,8 +42,8 @@ three categories have replacement contracts and compile coverage. 1. Keep all direct ECStore and storage-api imports centralized in `crates/obs/src/metrics/storage_api.rs`. -2. Keep projecting ECStore data-usage output into obs-local DTOs before - collectors consume it. +2. Keep projecting ECStore data-usage and replication stats output into + obs-local DTOs before collectors consume it. 3. Introduce obs-owned provider traits for storage info, bucket info, quota, data usage, replication, bandwidth, and lifecycle queue snapshots. 4. Implement those traits in ECStore or an ECStore-owned adapter crate after the @@ -58,6 +57,8 @@ The architecture guard enforces this inventory boundary: - `crates/obs/src/metrics/storage_api.rs` is the only `rustfs-obs` source file allowed to reference `rustfs_ecstore` or `rustfs_storage_api`; +- raw replication stats handles and ECStore replication stat methods must stay + behind the snapshot helpers in `crates/obs/src/metrics/storage_api.rs`; - `rustfs-obs` must not add `storage_compat.rs` or `ecstore_compat.rs` passthrough bridges; - future extraction PRs must update this inventory and the guard in the same diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 29665abab..91a6972f1 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -201,6 +201,7 @@ REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.t REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt" ADMIN_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/admin_replication_dto_boundary_bypass_hits.txt" APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/app_replication_dto_boundary_bypass_hits.txt" +OBS_REPLICATION_STATS_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/obs_replication_stats_boundary_bypass_hits.txt" REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_bandwidth_boundary_bypass_hits.txt" REPLICATION_CONFIG_STORE_BYPASS_HITS_FILE="${TMP_DIR}/replication_config_store_bypass_hits.txt" REPLICATION_ERROR_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_error_boundary_bypass_hits.txt" @@ -2509,6 +2510,24 @@ if [[ -s "$REPLICATION_FACADE_BYPASS_HITS_FILE" ]]; then report_failure "replication facade imports must stay in local storage_api boundaries: $(paste -sd '; ' "$REPLICATION_FACADE_BYPASS_HITS_FILE")" fi +( + cd "$ROOT_DIR" + { + rg -n --with-filename '\b(ObsReplicationStats|obs_get_global_replication_stats|replication_stats_handle|get_sr_metrics_for_node|get_proxy_stats|mrf_stats)\b' \ + crates/obs/src/metrics \ + --glob '*.rs' \ + --glob '!crates/obs/src/metrics/storage_api.rs' || true + rg -n --with-filename 'rustfs_ecstore::api::bucket::replication' \ + crates/obs/src/metrics \ + --glob '*.rs' \ + --glob '!crates/obs/src/metrics/storage_api.rs' || true + } +) >"$OBS_REPLICATION_STATS_BOUNDARY_BYPASS_HITS_FILE" + +if [[ -s "$OBS_REPLICATION_STATS_BOUNDARY_BYPASS_HITS_FILE" ]]; then + report_failure "obs replication stats access must stay behind crates/obs/src/metrics/storage_api.rs snapshot helpers: $(paste -sd '; ' "$OBS_REPLICATION_STATS_BOUNDARY_BYPASS_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename 'pub(?:\(crate\))?\s+use\s+(?:config|datatypes|replication_pool|replication_resyncer|replication_state|rule)::\*;' \